perf(emulator): return MJPEG frames as views instead of per-frame copies (#6707)
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Neil
GitHub
parent
a6083c47dc
commit
c96d8e3b5a
|
|
@ -0,0 +1,176 @@
|
|||
#!/usr/bin/env node
|
||||
// Benchmark: main-process heap allocation of the emulator MJPEG frame parser
|
||||
// across a realistic stream burst.
|
||||
//
|
||||
// extractJpegFrames() splits an MJPEG byte stream into JPEG frames. The old
|
||||
// implementation copied every extracted frame via Buffer.from(subarray(...))
|
||||
// and copied the incoming chunk via Buffer.from(chunk) when no bytes were
|
||||
// pending. Both copies are unnecessary: each frame is consumed synchronously
|
||||
// by the IPC layer (which copies into a transferable ArrayBuffer), and the only
|
||||
// state retained across calls (`pending`) is independently copied out. The fix
|
||||
// returns frame *views* and reads the chunk directly, so the parser allocates
|
||||
// ~0 frame-sized buffers per second instead of one copy per frame.
|
||||
//
|
||||
// This script runs both implementations over the same synthetic 30fps stream
|
||||
// and reports total bytes allocated (via process.memoryUsage / gc deltas).
|
||||
import { performance, PerformanceObserver } from 'node:perf_hooks'
|
||||
|
||||
// Count bytes copied by Buffer.from inside the parser, deterministically. Each
|
||||
// implementation calls trackedCopy() wherever it would allocate a copy, so the
|
||||
// reported "bytes copied" is exact and independent of GC timing.
|
||||
let copiedBytes = 0
|
||||
function trackedCopy(buf) {
|
||||
copiedBytes += buf.length
|
||||
return Buffer.from(buf)
|
||||
}
|
||||
|
||||
const FPS = Number.parseInt(process.env.ORCA_MJPEG_BENCH_FPS ?? '30', 10)
|
||||
const SECONDS = Number.parseInt(process.env.ORCA_MJPEG_BENCH_SECONDS ?? '30', 10)
|
||||
const FRAME_BYTES = Number.parseInt(process.env.ORCA_MJPEG_BENCH_FRAME_BYTES ?? '184320', 10) // ~180 KiB
|
||||
|
||||
for (const [name, value] of [
|
||||
['ORCA_MJPEG_BENCH_FPS', FPS],
|
||||
['ORCA_MJPEG_BENCH_SECONDS', SECONDS],
|
||||
['ORCA_MJPEG_BENCH_FRAME_BYTES', FRAME_BYTES]
|
||||
]) {
|
||||
if (!Number.isInteger(value) || value <= 0) {
|
||||
throw new Error(`${name} must be a positive integer, received ${value}`)
|
||||
}
|
||||
}
|
||||
|
||||
const JPEG_START = Buffer.from([0xff, 0xd8])
|
||||
const JPEG_END = Buffer.from([0xff, 0xd9])
|
||||
|
||||
function makeFrame(seed) {
|
||||
const body = Buffer.alloc(FRAME_BYTES - 4)
|
||||
// Avoid an accidental 0xff 0xd9 inside the body so the frame boundary is clean.
|
||||
body.fill(seed % 251 || 1)
|
||||
return Buffer.concat([JPEG_START, body, JPEG_END])
|
||||
}
|
||||
|
||||
// --- old implementation: copy chunk + copy every frame -----------------------
|
||||
function extractOld(pending, chunk, maxPendingBytes = 2 * 1024 * 1024) {
|
||||
let cursor = pending.length > 0 ? Buffer.concat([pending, chunk]) : trackedCopy(chunk)
|
||||
const frames = []
|
||||
while (cursor.length > 0) {
|
||||
const frameStart = cursor.indexOf(JPEG_START)
|
||||
if (frameStart < 0) {
|
||||
const keepLastByte = cursor.at(-1) === 0xff
|
||||
return { frames, pending: keepLastByte ? Buffer.from([0xff]) : Buffer.alloc(0) }
|
||||
}
|
||||
if (frameStart > 0) {
|
||||
cursor = cursor.subarray(frameStart)
|
||||
}
|
||||
const frameEnd = cursor.indexOf(JPEG_END, JPEG_START.length)
|
||||
if (frameEnd < 0) {
|
||||
const tail = cursor.length <= maxPendingBytes ? cursor : cursor.subarray(-maxPendingBytes)
|
||||
return { frames, pending: trackedCopy(tail) }
|
||||
}
|
||||
const nextOffset = frameEnd + JPEG_END.length
|
||||
frames.push(trackedCopy(cursor.subarray(0, nextOffset)))
|
||||
cursor = cursor.subarray(nextOffset)
|
||||
}
|
||||
return { frames, pending: Buffer.alloc(0) }
|
||||
}
|
||||
|
||||
// --- new implementation: views, no chunk copy --------------------------------
|
||||
function extractNew(pending, chunk, maxPendingBytes = 2 * 1024 * 1024) {
|
||||
let cursor = pending.length > 0 ? Buffer.concat([pending, chunk]) : chunk
|
||||
const frames = []
|
||||
while (cursor.length > 0) {
|
||||
const frameStart = cursor.indexOf(JPEG_START)
|
||||
if (frameStart < 0) {
|
||||
const keepLastByte = cursor.at(-1) === 0xff
|
||||
return { frames, pending: keepLastByte ? Buffer.from([0xff]) : Buffer.alloc(0) }
|
||||
}
|
||||
if (frameStart > 0) {
|
||||
cursor = cursor.subarray(frameStart)
|
||||
}
|
||||
const frameEnd = cursor.indexOf(JPEG_END, JPEG_START.length)
|
||||
if (frameEnd < 0) {
|
||||
const tail = cursor.length <= maxPendingBytes ? cursor : cursor.subarray(-maxPendingBytes)
|
||||
return { frames, pending: trackedCopy(tail) }
|
||||
}
|
||||
const nextOffset = frameEnd + JPEG_END.length
|
||||
frames.push(cursor.subarray(0, nextOffset))
|
||||
cursor = cursor.subarray(nextOffset)
|
||||
}
|
||||
return { frames, pending: Buffer.alloc(0) }
|
||||
}
|
||||
|
||||
function runStream(extract, frames) {
|
||||
// Simulate one network chunk per frame (the common MJPEG transport shape):
|
||||
// each chunk delivers exactly one whole JPEG, so `pending` is empty per call.
|
||||
let totalFrameBytes = 0
|
||||
for (const frame of frames) {
|
||||
const result = extract(Buffer.alloc(0), frame)
|
||||
for (const f of result.frames) {
|
||||
totalFrameBytes += f.length
|
||||
}
|
||||
}
|
||||
return totalFrameBytes
|
||||
}
|
||||
|
||||
function measure(label, extract, frames) {
|
||||
let gcCount = 0
|
||||
let gcPauseMs = 0
|
||||
const observer = new PerformanceObserver((list) => {
|
||||
for (const entry of list.getEntries()) {
|
||||
gcCount += 1
|
||||
gcPauseMs += entry.duration
|
||||
}
|
||||
})
|
||||
observer.observe({ entryTypes: ['gc'] })
|
||||
|
||||
global.gc?.()
|
||||
copiedBytes = 0
|
||||
const before = process.memoryUsage()
|
||||
const start = performance.now()
|
||||
const consumed = runStream(extract, frames)
|
||||
const elapsed = performance.now() - start
|
||||
const after = process.memoryUsage()
|
||||
observer.disconnect()
|
||||
return {
|
||||
label,
|
||||
elapsedMs: elapsed,
|
||||
consumedFrameBytes: consumed,
|
||||
copiedMb: copiedBytes / (1024 * 1024),
|
||||
rssDeltaMb: (after.rss - before.rss) / (1024 * 1024),
|
||||
gcCount,
|
||||
gcPauseMs
|
||||
}
|
||||
}
|
||||
|
||||
const totalFrames = FPS * SECONDS
|
||||
const frames = Array.from({ length: totalFrames }, (_, i) => makeFrame(i))
|
||||
const streamMb = (totalFrames * FRAME_BYTES) / (1024 * 1024)
|
||||
|
||||
console.log(
|
||||
`MJPEG parser alloc benchmark: ${FPS}fps × ${SECONDS}s = ${totalFrames} frames, ` +
|
||||
`${(FRAME_BYTES / 1024).toFixed(0)} KiB/frame (${streamMb.toFixed(1)} MiB streamed)\n`
|
||||
)
|
||||
|
||||
if (!global.gc) {
|
||||
console.log('(run with `node --expose-gc` for accurate heap deltas)\n')
|
||||
}
|
||||
|
||||
const oldResult = measure('before (Buffer.from copies)', extractOld, frames)
|
||||
const newResult = measure('after (subarray views) ', extractNew, frames)
|
||||
|
||||
for (const r of [oldResult, newResult]) {
|
||||
console.log(
|
||||
`${r.label} copied=${r.copiedMb.toFixed(1)} MiB ` +
|
||||
`rssΔ=${r.rssDeltaMb.toFixed(1)} MiB gc=${r.gcCount} (${r.gcPauseMs.toFixed(1)}ms) ` +
|
||||
`time=${r.elapsedMs.toFixed(1)}ms`
|
||||
)
|
||||
}
|
||||
|
||||
const savedMb = oldResult.copiedMb - newResult.copiedMb
|
||||
console.log(
|
||||
`\nEliminated ${savedMb.toFixed(1)} MiB of transient frame copies over ${SECONDS}s ` +
|
||||
`(~${(savedMb / SECONDS).toFixed(2)} MiB/s of avoided allocation + GC pressure on the main process).`
|
||||
)
|
||||
console.log(
|
||||
`Same frame bytes delivered (${(oldResult.consumedFrameBytes / (1024 * 1024)).toFixed(1)} MiB) ` +
|
||||
`with ${newResult.copiedMb.toFixed(1)} MiB copied instead of ${oldResult.copiedMb.toFixed(1)} MiB.`
|
||||
)
|
||||
|
|
@ -27,4 +27,19 @@ describe('extractJpegFrames', () => {
|
|||
|
||||
expect(second.frames).toEqual([JPEG_A])
|
||||
})
|
||||
|
||||
// Regression: frames are returned as views into the chunk (no per-frame copy),
|
||||
// so the retained `pending` must still be an independent copy — otherwise a
|
||||
// later mutation of the source chunk could corrupt buffered partial frames.
|
||||
it('does not retain a view aliased to the input chunk', () => {
|
||||
const chunk = Buffer.concat([JPEG_A, Buffer.from([0xff, 0xd8, 0x09])]) // A + partial B
|
||||
const result = extractJpegFrames(Buffer.alloc(0), chunk)
|
||||
|
||||
expect(result.frames).toEqual([JPEG_A])
|
||||
expect(result.pending).toEqual(Buffer.from([0xff, 0xd8, 0x09]))
|
||||
|
||||
// Mutating the original chunk after parsing must not change pending.
|
||||
chunk.fill(0)
|
||||
expect(result.pending).toEqual(Buffer.from([0xff, 0xd8, 0x09]))
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -22,7 +22,12 @@ export function extractJpegFrames(
|
|||
chunk: Buffer<ArrayBufferLike>,
|
||||
maxPendingBytes = DEFAULT_MAX_PENDING_BYTES
|
||||
): MjpegFrameParseResult {
|
||||
let cursor = pending.length > 0 ? Buffer.concat([pending, chunk]) : Buffer.from(chunk)
|
||||
// Why: when there are no leftover bytes the chunk already holds whole frames,
|
||||
// so read it directly instead of copying — frames below are views consumed
|
||||
// synchronously (the IPC layer copies into a transferable ArrayBuffer), and
|
||||
// any retained `pending` is copied out via trimPendingBuffer, so no chunk
|
||||
// memory is held across calls. At ~30fps this avoids a full-frame copy/frame.
|
||||
let cursor = pending.length > 0 ? Buffer.concat([pending, chunk]) : chunk
|
||||
const frames: Buffer[] = []
|
||||
|
||||
while (cursor.length > 0) {
|
||||
|
|
@ -41,7 +46,9 @@ export function extractJpegFrames(
|
|||
}
|
||||
|
||||
const nextOffset = frameEnd + JPEG_END.length
|
||||
frames.push(Buffer.from(cursor.subarray(0, nextOffset)))
|
||||
// A view, not a copy: the caller consumes each frame synchronously before
|
||||
// the next chunk arrives, and `cursor` is only ever re-sliced (never mutated).
|
||||
frames.push(cursor.subarray(0, nextOffset))
|
||||
cursor = cursor.subarray(nextOffset)
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue