mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
perf(emulator): accumulate MJPEG chunks instead of re-joining partial frames (#32967)
* perf(emulator): accumulate MJPEG chunks instead of re-joining partial frames A simulator frame spans many socket chunks, and extractJpegFrames re-concatenated the growing partial frame on every chunk, which is quadratic in chunks per frame on the Electron main thread. Hold chunks until one could complete the frame, then join once. Held chunks are copied, so no caller chunk memory is retained. * perf(emulator): ignore empty MJPEG chunks in the accumulator --------- Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { extractJpegFrames } from './mjpeg-frame-parser'
|
||||
import { MjpegFrameAccumulator, extractJpegFrames } from './mjpeg-frame-parser'
|
||||
|
||||
const JPEG_A = Buffer.from([0xff, 0xd8, 0x01, 0x02, 0xff, 0xd9])
|
||||
const JPEG_B = Buffer.from([0xff, 0xd8, 0x03, 0x04, 0xff, 0xd9])
|
||||
@@ -40,3 +40,49 @@ describe('extractJpegFrames', () => {
|
||||
expect(result.pending).toEqual(Buffer.from([0xff, 0xd8, 0x09]))
|
||||
})
|
||||
})
|
||||
|
||||
describe('MjpegFrameAccumulator', () => {
|
||||
const LONG_JPEG = Buffer.concat([
|
||||
Buffer.from([0xff, 0xd8]),
|
||||
Buffer.from(Array.from({ length: 40 }, (_, index) => (index % 5 === 0 ? 0xff : index))),
|
||||
Buffer.from([0xff, 0xd9])
|
||||
])
|
||||
const stream = Buffer.concat([
|
||||
Buffer.from('--frame\r\nContent-Type: image/jpeg\r\n\r\n'),
|
||||
LONG_JPEG,
|
||||
Buffer.from('\r\n--frame\r\n\r\n'),
|
||||
JPEG_B
|
||||
])
|
||||
|
||||
it.each([1, 2, 3, 7, 16, stream.length])(
|
||||
'returns the same frames as extractJpegFrames for %i-byte chunks',
|
||||
(size) => {
|
||||
const accumulator = new MjpegFrameAccumulator()
|
||||
const frames: Buffer[] = []
|
||||
let pending: Buffer<ArrayBufferLike> = Buffer.alloc(0)
|
||||
const expected: Buffer[] = []
|
||||
for (let offset = 0; offset < stream.length; offset += size) {
|
||||
const chunk = stream.subarray(offset, offset + size)
|
||||
frames.push(...accumulator.push(chunk).map((frame) => Buffer.from(frame)))
|
||||
const result = extractJpegFrames(pending, chunk)
|
||||
pending = result.pending
|
||||
expected.push(...result.frames.map((frame) => Buffer.from(frame)))
|
||||
}
|
||||
|
||||
expect(frames).toEqual([LONG_JPEG, JPEG_B])
|
||||
expect(frames).toEqual(expected)
|
||||
}
|
||||
)
|
||||
|
||||
it('does not retain caller chunk memory while a frame is buffered', () => {
|
||||
const accumulator = new MjpegFrameAccumulator()
|
||||
const first = Buffer.from(LONG_JPEG.subarray(0, 10))
|
||||
const middle = Buffer.from(LONG_JPEG.subarray(10, 30))
|
||||
expect(accumulator.push(first)).toEqual([])
|
||||
expect(accumulator.push(middle)).toEqual([])
|
||||
first.fill(0)
|
||||
middle.fill(0)
|
||||
|
||||
expect(accumulator.push(LONG_JPEG.subarray(30))).toEqual([LONG_JPEG])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -48,3 +48,41 @@ export function extractJpegFrames(
|
||||
|
||||
return { frames, pending: Buffer.alloc(0) }
|
||||
}
|
||||
|
||||
// Why: a frame spans many socket chunks; re-joining the growing tail on every chunk is
|
||||
// quadratic in chunks per frame, so hold chunks until one could complete the frame.
|
||||
export class MjpegFrameAccumulator {
|
||||
private parts: Buffer<ArrayBufferLike>[] = []
|
||||
private bytes = 0
|
||||
|
||||
constructor(private readonly maxPendingBytes = DEFAULT_MAX_PENDING_BYTES) {}
|
||||
|
||||
push(chunk: Buffer<ArrayBufferLike>): Buffer<ArrayBufferLike>[] {
|
||||
if (chunk.length === 0) {
|
||||
return []
|
||||
}
|
||||
const head = this.parts[0]
|
||||
if (
|
||||
head?.[0] === 0xff &&
|
||||
head[1] === 0xd8 &&
|
||||
!(this.parts.at(-1)?.at(-1) === 0xff && chunk[0] === 0xd9) &&
|
||||
!chunk.includes(JPEG_END) &&
|
||||
this.bytes + chunk.length <= this.maxPendingBytes
|
||||
) {
|
||||
// Copied like extractJpegFrames' pending, so no caller chunk memory is retained.
|
||||
this.parts.push(Buffer.from(chunk))
|
||||
this.bytes += chunk.length
|
||||
return []
|
||||
}
|
||||
const pending = this.parts.length === 1 ? this.parts[0] : Buffer.concat(this.parts, this.bytes)
|
||||
const result = extractJpegFrames(pending, chunk, this.maxPendingBytes)
|
||||
this.parts = result.pending.length > 0 ? [result.pending] : []
|
||||
this.bytes = result.pending.length
|
||||
return result.frames
|
||||
}
|
||||
|
||||
reset(): void {
|
||||
this.parts = []
|
||||
this.bytes = 0
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { request as httpRequest, type ClientRequest, type IncomingMessage } from 'node:http'
|
||||
import { request as httpsRequest } from 'node:https'
|
||||
import { extractJpegFrames } from './mjpeg-frame-parser'
|
||||
import { MjpegFrameAccumulator } from './mjpeg-frame-parser'
|
||||
|
||||
const RECONNECT_DELAY_MS = 1_000
|
||||
const REQUEST_TIMEOUT_MS = 10_000
|
||||
@@ -41,7 +41,7 @@ function requestForUrl(url: URL, response: (res: IncomingMessage) => void): Clie
|
||||
}
|
||||
|
||||
export class MjpegFrameStream {
|
||||
private pending: Buffer<ArrayBufferLike> = Buffer.alloc(0)
|
||||
private readonly accumulator = new MjpegFrameAccumulator()
|
||||
private reconnectTimer: NodeJS.Timeout | null = null
|
||||
private request: ClientRequest | null = null
|
||||
private stopped = false
|
||||
@@ -67,7 +67,7 @@ export class MjpegFrameStream {
|
||||
}
|
||||
this.request?.destroy()
|
||||
this.request = null
|
||||
this.pending = Buffer.alloc(0)
|
||||
this.accumulator.reset()
|
||||
}
|
||||
|
||||
private scheduleReconnect(): void {
|
||||
@@ -114,9 +114,7 @@ export class MjpegFrameStream {
|
||||
}
|
||||
|
||||
private handleChunk(chunk: Buffer<ArrayBufferLike>): void {
|
||||
const result = extractJpegFrames(this.pending, chunk)
|
||||
this.pending = result.pending
|
||||
for (const frame of result.frames) {
|
||||
for (const frame of this.accumulator.push(chunk)) {
|
||||
const now = Date.now()
|
||||
if (this.lastFrameAt > 0 && now - this.lastFrameAt < MIN_FRAME_INTERVAL_MS) {
|
||||
continue
|
||||
|
||||
Reference in New Issue
Block a user