diff --git a/config/scripts/scrcpy-frame-buffering-benchmark.mjs b/config/scripts/scrcpy-frame-buffering-benchmark.mjs new file mode 100644 index 00000000000..c84c80e6f55 --- /dev/null +++ b/config/scripts/scrcpy-frame-buffering-benchmark.mjs @@ -0,0 +1,134 @@ +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import { performance } from 'node:perf_hooks' +import { transform } from 'esbuild' +import { buildCounterbalancedSchedule } from './counterbalanced-benchmark-schedule.mjs' +import { summarizeBenchmarkSamples } from './benchmark-sample-summary.mjs' + +// git show :src/main/emulator/android/scrcpy-video-frame-parser.ts | node config/scripts/scrcpy-frame-buffering-benchmark.mjs +async function load(source) { + const { code } = await transform(source, { loader: 'ts', format: 'esm' }) + return import(`data:text/javascript;base64,${Buffer.from(code).toString('base64')}`) +} +const before = (await load(readFileSync(0, 'utf8'))).parseScrcpyVideoFrames +const after = ( + await load(readFileSync('src/main/emulator/android/scrcpy-video-frame-parser.ts', 'utf8')) +).parseScrcpyVideoFrames +const { RelayFrameBuffer } = await load(readFileSync('src/shared/relay-frame-buffer.ts', 'utf8')) + +function reader(arm) { + if (arm === 'before') { + let pending = Buffer.alloc(0) + return { + read(chunk) { + // Match the baseline session's copy before invoking its production parser. + const result = before(Buffer.alloc(0), Buffer.concat([pending, chunk])) + pending = result.pending + return result.frames + }, + pending: () => pending + } + } + const pending = new RelayFrameBuffer() + return { + read(chunk) { + // Include the session's mandatory ownership copy in the queued parser arm. + if (chunk.length > 0) { + pending.append(Buffer.from(chunk)) + } + return after(pending) + }, + pending: () => (pending.length > 0 ? pending.peek(pending.length) : Buffer.alloc(0)) + } +} + +function packet(size, meta = 123n) { + const frame = Buffer.alloc(size + 12, 7) + frame.writeBigUInt64BE(meta, 0) + frame.writeUInt32BE(size, 8) + return frame +} + +let seed = 42 +const random = (max) => { + seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0 + return seed % max +} +let differentialChunks = 0 +for (let trial = 0; trial < 1000; trial += 1) { + const stream = Buffer.concat( + Array.from({ length: 1 + random(8) }, (_, index) => + packet(random(256), (BigInt(random(4)) << 62n) | BigInt(index)) + ) + ) + const oldReader = reader('before') + const newReader = reader('after') + for (let offset = 0; offset < stream.length;) { + const size = 1 + random(128) + const chunk = stream.subarray(offset, offset + size) + assert.deepEqual(newReader.read(chunk), oldReader.read(chunk)) + assert.deepEqual(newReader.pending(), oldReader.pending()) + assert.deepEqual(newReader.read(Buffer.alloc(0)), oldReader.read(Buffer.alloc(0))) + differentialChunks += 1 + offset += size + } +} + +const results = [] +for (const frameBytes of [32, 4096, 65_536, 1_048_576]) { + const frame = packet(frameBytes) + const expected = reader('before').read(frame) + for (const chunkBytes of new Set([frame.length, 65_536, 4096, 1024])) { + if (chunkBytes > frame.length) { + continue + } + const chunks = [] + for (let offset = 0; offset < frame.length; offset += chunkBytes) { + chunks.push(frame.subarray(offset, offset + chunkBytes)) + } + const iterations = Math.max(10, Math.floor(4_194_304 / frameBytes)) + function run(arm, repeats) { + const parser = reader(arm) + let frames + const started = performance.now() + for (let iteration = 0; iteration < repeats; iteration += 1) { + for (const chunk of chunks) { + frames = parser.read(chunk) + } + } + const ms = performance.now() - started + assert.deepEqual(frames, expected) + assert.equal(parser.pending().length, 0) + return ms + } + run('before', iterations) + run('after', iterations) + /** @type {{ before: number[], after: number[] }} */ + const samples = { before: [], after: [] } + for (const pair of buildCounterbalancedSchedule(8, 'before', 'after')) { + for (const arm of pair) { + samples[arm].push(run(arm, iterations)) + } + } + results.push({ + frameBytes, + chunkBytes, + iterations, + meanMicrosecondsPerFrame: Object.fromEntries( + Object.entries(samples).map(([arm, values]) => [ + arm, + (values.reduce((sum, ms) => sum + ms, 0) * 1000) / values.length / iterations + ]) + ), + before: summarizeBenchmarkSamples(samples.before), + after: summarizeBenchmarkSamples(samples.after) + }) + } +} +console.log( + JSON.stringify( + { node: process.version, platform: process.platform, differentialChunks, results }, + null, + 2 + ) +) diff --git a/src/main/emulator/android/scrcpy-stream-session.test.ts b/src/main/emulator/android/scrcpy-stream-session.test.ts new file mode 100644 index 00000000000..108014a6078 --- /dev/null +++ b/src/main/emulator/android/scrcpy-stream-session.test.ts @@ -0,0 +1,172 @@ +import { EventEmitter } from 'node:events' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { ScrcpyStreamSession } from './scrcpy-stream-session' + +const io = vi.hoisted(() => ({ spawn: vi.fn(), connect: vi.fn() })) +vi.mock('node:child_process', () => ({ spawn: io.spawn })) +vi.mock('node:net', () => ({ connect: io.connect })) +vi.mock('../emulator-probe', () => ({ emulatorProbe: vi.fn(), emulatorProbeError: vi.fn() })) + +class TestSocket extends EventEmitter { + destroy = vi.fn() + setTimeout = vi.fn() +} + +function packet(size: number, meta = 123n): Buffer { + const result = Buffer.alloc(12 + size, 7) + result.writeBigUInt64BE(meta, 0) + result.writeUInt32BE(size, 8) + return result +} + +function handshake(): Buffer { + const result = Buffer.alloc(77) + result.write('test-device', 1) + result.write('h264', 65) + result.writeUInt32BE(1080, 69) + result.writeUInt32BE(2400, 73) + return result +} + +async function startSession() { + const video = new TestSocket() + const control = new TestSocket() + const server = Object.assign(new EventEmitter(), { + stdout: new EventEmitter(), + stderr: new EventEmitter(), + kill: vi.fn() + }) + io.spawn.mockReturnValue(server) + io.connect.mockReturnValueOnce(video).mockReturnValueOnce(control) + const callbacks = { onMeta: vi.fn(), onFrame: vi.fn(), onError: vi.fn(), onClose: vi.fn() } + const runner = vi.fn().mockResolvedValue({ code: 0, stdout: '', stderr: '' }) + const started = ScrcpyStreamSession.start( + { + runner, + sdk: { sdkRoot: 'sdk', adb: 'adb', emulator: 'emulator', avdmanager: 'avdmanager' }, + serial: 'test-device', + localJarPath: 'server.jar', + localPort: 12345 + }, + callbacks + ) + await vi.waitFor(() => expect(io.connect).toHaveBeenCalledTimes(1)) + return { video, control, server, callbacks, started } +} + +beforeEach(() => { + io.spawn.mockReset() + io.connect.mockReset() +}) + +describe('ScrcpyStreamSession video buffering', () => { + it('accepts bytewise handshake and frames, including empty chunks and empty frames', async () => { + const { video, callbacks, started } = await startSession() + const header = handshake() + for (let index = 0; index < header.length - 1; index += 1) { + video.emit('data', header.subarray(index, index + 1)) + } + expect(callbacks.onMeta).not.toHaveBeenCalled() + video.emit('data', header.subarray(-1)) + const session = await started + expect(callbacks.onMeta).toHaveBeenCalledWith({ codecId: 'h264', width: 1080, height: 2400 }) + const stream = Buffer.concat([packet(0, 1n << 63n), packet(3, (1n << 62n) | 5n)]) + for (const byte of stream) { + video.emit('data', Buffer.alloc(0)) + video.emit('data', Buffer.from([byte])) + } + expect(callbacks.onFrame.mock.calls.map(([frame]) => frame)).toEqual([ + { config: true, keyFrame: false, pts: 0n, data: Buffer.alloc(0) }, + { config: false, keyFrame: true, pts: 5n, data: Buffer.alloc(3, 7) } + ]) + session.close() + }) + + it('owns pending bytes and emitted frames independently of input chunks', async () => { + const { video, callbacks, started } = await startSession() + video.emit('data', handshake()) + const session = await started + const first = packet(4) + const second = packet(6, 456n) + const chunk = Buffer.concat([first, second.subarray(0, 14)]) + video.emit('data', chunk) + chunk.fill(0) + const tail = Buffer.from(second.subarray(14)) + video.emit('data', tail) + tail.fill(0) + expect(callbacks.onFrame.mock.calls.map(([frame]) => frame)).toEqual([ + { config: false, keyFrame: false, pts: 123n, data: Buffer.alloc(4, 7) }, + { config: false, keyFrame: false, pts: 456n, data: Buffer.alloc(6, 7) } + ]) + session.close() + }) + + it('emits initial metadata and frames before startup resolves', async () => { + const { video, callbacks, started } = await startSession() + const events: string[] = [] + callbacks.onMeta.mockImplementation(() => events.push('meta')) + callbacks.onFrame.mockImplementation(() => events.push('frame')) + const ready = started.then((session) => { + events.push('ready') + return session + }) + video.emit('data', Buffer.concat([handshake(), packet(3), packet(5)])) + expect(events).toEqual(['meta', 'frame', 'frame']) + const session = await ready + expect(events).toEqual(['meta', 'frame', 'frame', 'ready']) + session.close() + }) + + it('fails an already started session on a corrupt batch without delivering partial results', async () => { + const { video, callbacks, started } = await startSession() + video.emit('data', handshake()) + const session = await started + const corrupt = packet(0) + corrupt.writeUInt32BE(16 * 1024 * 1024 + 1, 8) + video.emit('data', Buffer.concat([packet(1), corrupt])) + expect(callbacks.onFrame).not.toHaveBeenCalled() + expect(callbacks.onError).toHaveBeenCalledExactlyOnceWith(expect.stringMatching(/desynced/)) + expect(callbacks.onClose).toHaveBeenCalledTimes(1) + session.close() + expect(callbacks.onClose).toHaveBeenCalledTimes(1) + }) + + it('does not resolve startup or deliver earlier frames if the first batch is desynced', async () => { + const { video, callbacks, started, server } = await startSession() + const corrupt = packet(0) + corrupt.writeUInt32BE(16 * 1024 * 1024 + 1, 8) + const rejected = expect(started).rejects.toThrow(/desynced/) + video.emit('data', Buffer.concat([handshake(), packet(1), corrupt])) + await rejected + expect(callbacks.onMeta).toHaveBeenCalledTimes(1) + expect(callbacks.onFrame).not.toHaveBeenCalled() + expect(callbacks.onError).toHaveBeenCalledTimes(1) + expect(callbacks.onClose).toHaveBeenCalledTimes(1) + expect(server.kill).toHaveBeenCalledTimes(1) + expect(video.destroy).toHaveBeenCalledTimes(1) + }) + + it('does not repeatedly concatenate a growing fragmented frame', async () => { + const { video, callbacks, started } = await startSession() + video.emit('data', handshake()) + const session = await started + const frame = packet(1024 * 1024) + const concat = Buffer.concat + let concatenatedBytes = 0 + const spy = vi.spyOn(Buffer, 'concat').mockImplementation((buffers, length) => { + concatenatedBytes += length ?? buffers.reduce((sum, part) => sum + part.length, 0) + return concat(buffers, length) + }) + try { + for (let offset = 0; offset < frame.length; offset += 4096) { + video.emit('data', frame.subarray(offset, offset + 4096)) + } + } finally { + spy.mockRestore() + session.close() + } + expect(callbacks.onFrame).toHaveBeenCalledTimes(1) + expect(callbacks.onFrame.mock.calls[0][0].data).toEqual(frame.subarray(12)) + expect(concatenatedBytes).toBeLessThanOrEqual(frame.length * 2) + }) +}) diff --git a/src/main/emulator/android/scrcpy-stream-session.ts b/src/main/emulator/android/scrcpy-stream-session.ts index 0333e9ee5db..b8c65aecf89 100644 --- a/src/main/emulator/android/scrcpy-stream-session.ts +++ b/src/main/emulator/android/scrcpy-stream-session.ts @@ -1,6 +1,7 @@ import { spawn, type ChildProcess } from 'node:child_process' import { connect, type Socket } from 'node:net' import { randomBytes } from 'node:crypto' +import { RelayFrameBuffer } from '../../../shared/relay-frame-buffer' import type { AndroidCommandRunner } from './android-command-runner' import type { AndroidSdkPaths } from './android-sdk-discovery' import { ensureAdbOk } from './android-adb-result' @@ -14,7 +15,6 @@ import { import { parseScrcpyVideoFrames, parseScrcpyVideoMeta, - type ScrcpyFrameParseResult, type ScrcpyVideoFrame, type ScrcpyVideoMeta } from './scrcpy-video-frame-parser' @@ -57,7 +57,7 @@ export class ScrcpyStreamSession { private server: ChildProcess | null = null private videoSocket: Socket | null = null private controlSocket: Socket | null = null - private pendingVideo: Buffer = Buffer.alloc(0) + private readonly pendingVideo = new RelayFrameBuffer() private metaSeen = false private headerStripped = false private closed = false @@ -206,47 +206,48 @@ export class ScrcpyStreamSession { } private handleVideoChunk(chunk: Buffer): void { - let buffer = Buffer.concat([this.pendingVideo, chunk]) + const buffer = this.pendingVideo + if (chunk.length > 0) { + // Socket chunks and emitted frames must not share mutable pending storage. + buffer.append(Buffer.from(chunk)) + } // The first socket carries a 1-byte readiness marker + the 64-byte device name. if (!this.headerStripped) { const headerLen = DUMMY_BYTE + DEVICE_NAME_BYTES if (buffer.length < headerLen) { - this.pendingVideo = buffer return } - buffer = Buffer.from(buffer.subarray(headerLen)) + buffer.discard(headerLen) this.headerStripped = true } let shouldResolveReady = false if (!this.metaSeen) { - const meta = parseScrcpyVideoMeta(buffer) - if (!meta) { - this.pendingVideo = buffer + if (buffer.length < 12) { return } + const meta = parseScrcpyVideoMeta(buffer.peek(12))! this.metaSeen = true emulatorProbe('scrcpy.meta', meta) this.callbacks.onMeta(meta) shouldResolveReady = true - buffer = Buffer.from(buffer.subarray(12)) + buffer.discard(12) } // The parser throws on a desynced stream (e.g. an absurd frame size); catch // it here so it fails the session via the normal teardown path rather than // surfacing as an unhandled exception in this socket 'data' listener. - let result: ScrcpyFrameParseResult + let frames: ScrcpyVideoFrame[] try { - result = parseScrcpyVideoFrames(Buffer.alloc(0), buffer) + frames = parseScrcpyVideoFrames(buffer) } catch (error) { this.fail(error instanceof Error ? error.message : String(error)) return } - this.pendingVideo = result.pending if (shouldResolveReady) { this.resolveReady?.() this.resolveReady = null this.rejectReady = null } - for (const frame of result.frames) { + for (const frame of frames) { this.callbacks.onFrame(frame) } } diff --git a/src/main/emulator/android/scrcpy-video-frame-parser.test.ts b/src/main/emulator/android/scrcpy-video-frame-parser.test.ts index ecbceeff3ca..78f1771eaa4 100644 --- a/src/main/emulator/android/scrcpy-video-frame-parser.test.ts +++ b/src/main/emulator/android/scrcpy-video-frame-parser.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from 'vitest' -import { parseScrcpyVideoFrames, parseScrcpyVideoMeta } from './scrcpy-video-frame-parser' +import { RelayFrameBuffer } from '../../../shared/relay-frame-buffer' +import { + MAX_PENDING_CHUNKS, + parseScrcpyVideoFrames, + parseScrcpyVideoMeta +} from './scrcpy-video-frame-parser' const CONFIG = 1n << 63n const KEY = 1n << 62n @@ -28,7 +33,9 @@ describe('parseScrcpyVideoMeta', () => { describe('parseScrcpyVideoFrames', () => { it('extracts config and key frames with their flags and data', () => { const stream = Buffer.concat([frame(CONFIG, [0, 0, 0, 1]), frame(KEY | 123n, [1, 2, 3])]) - const { frames, pending } = parseScrcpyVideoFrames(Buffer.alloc(0), stream) + const pending = new RelayFrameBuffer() + pending.append(stream) + const frames = parseScrcpyVideoFrames(pending) expect(pending.length).toBe(0) expect(frames).toHaveLength(2) expect(frames[0]).toMatchObject({ config: true, keyFrame: false }) @@ -39,19 +46,89 @@ describe('parseScrcpyVideoFrames', () => { it('buffers a partial frame across chunks', () => { const full = frame(5n, [9, 9, 9, 9]) - const r1 = parseScrcpyVideoFrames(Buffer.alloc(0), full.subarray(0, 14)) - expect(r1.frames).toHaveLength(0) - expect(r1.pending.length).toBe(14) - const r2 = parseScrcpyVideoFrames(r1.pending, full.subarray(14)) - expect(r2.frames).toHaveLength(1) - expect([...r2.frames[0].data]).toEqual([9, 9, 9, 9]) - expect(r2.pending.length).toBe(0) + const pending = new RelayFrameBuffer() + pending.append(full.subarray(0, 14)) + expect(parseScrcpyVideoFrames(pending)).toHaveLength(0) + expect(pending.length).toBe(14) + pending.append(full.subarray(14)) + const frames = parseScrcpyVideoFrames(pending) + expect(frames).toHaveLength(1) + expect([...frames[0].data]).toEqual([9, 9, 9, 9]) + expect(pending.length).toBe(0) + }) + + it('does not retain a consumed large packet behind a one-byte pending suffix', () => { + const first = Buffer.alloc(4 * 1024 * 1024 + 12, 7) + first.writeBigUInt64BE(123n, 0) + first.writeUInt32BE(first.length - 12, 8) + const second = frame(KEY | 456n, [1, 2, 3]) + const chunk = Buffer.concat([first, second.subarray(0, 1)]) + const pending = new RelayFrameBuffer() + pending.append(chunk) + + const frames = parseScrcpyVideoFrames(pending) + expect(frames).toHaveLength(1) + expect(frames[0]).toMatchObject({ config: false, keyFrame: false, pts: 123n }) + expect(frames[0].data.equals(first.subarray(12))).toBe(true) + expect(pending.length).toBe(1) + expect(pending.peek(1)).toEqual(second.subarray(0, 1)) + expect(pending.peek(1).buffer === chunk.buffer).toBe(false) + expect(pending.peek(1).buffer.byteLength).toBeLessThan(chunk.length) + + chunk.fill(0xff) + pending.append(second.subarray(1)) + expect(parseScrcpyVideoFrames(pending)).toEqual([ + { config: false, keyFrame: true, pts: 456n, data: Buffer.from([1, 2, 3]) } + ]) + expect(pending.length).toBe(0) + }) + + it('keeps mostly live chunk storage instead of recopying a large pending frame', () => { + const first = frame(123n, [1, 2, 3]) + const second = Buffer.alloc(4 * 1024 * 1024 + 12, 7) + second.writeBigUInt64BE(KEY | 456n, 0) + second.writeUInt32BE(second.length - 12, 8) + const split = 3 * 1024 * 1024 + const chunk = Buffer.concat([first, second.subarray(0, split)]) + const pending = new RelayFrameBuffer() + pending.append(chunk) + + expect(parseScrcpyVideoFrames(pending)).toHaveLength(1) + expect(pending.length).toBe(split) + expect(pending.peek(1).buffer === chunk.buffer).toBe(true) + + pending.append(second.subarray(split)) + const frames = parseScrcpyVideoFrames(pending) + expect(frames).toHaveLength(1) + expect(frames[0]).toMatchObject({ config: false, keyFrame: true, pts: 456n }) + expect(frames[0].data.equals(second.subarray(12))).toBe(true) + expect(pending.length).toBe(0) + }) + + it('bounds queued fragment count for a large frame delivered one byte at a time', () => { + const full = Buffer.alloc(256 * 1024 + 12, 7) + full.writeBigUInt64BE(KEY | 789n, 0) + full.writeUInt32BE(full.length - 12, 8) + const pending = new RelayFrameBuffer() + let maxChunks = 0 + let frames: ReturnType = [] + for (const byte of full) { + pending.append(Buffer.from([byte])) + frames = parseScrcpyVideoFrames(pending) + maxChunks = Math.max(maxChunks, pending.chunkCount) + } + expect(maxChunks).toBeLessThanOrEqual(MAX_PENDING_CHUNKS) + expect(frames).toHaveLength(1) + expect(frames[0]).toMatchObject({ config: false, keyFrame: true, pts: 789n }) + expect(frames[0].data.equals(full.subarray(12))).toBe(true) + expect(pending.length).toBe(0) }) it('holds an incomplete header until more bytes arrive', () => { - const result = parseScrcpyVideoFrames(Buffer.alloc(0), Buffer.from([0, 1, 2])) - expect(result.frames).toHaveLength(0) - expect(result.pending.length).toBe(3) + const pending = new RelayFrameBuffer() + pending.append(Buffer.from([0, 1, 2])) + expect(parseScrcpyVideoFrames(pending)).toHaveLength(0) + expect(pending.length).toBe(3) }) it('throws on a desynced frame size instead of buffering toward OOM', () => { @@ -59,6 +136,8 @@ describe('parseScrcpyVideoFrames', () => { // never be satisfied, leaving the whole buffer pending forever. const header = Buffer.alloc(12) header.writeUInt32BE(64 * 1024 * 1024, 8) - expect(() => parseScrcpyVideoFrames(Buffer.alloc(0), header)).toThrow(/desynced/) + const pending = new RelayFrameBuffer() + pending.append(header) + expect(() => parseScrcpyVideoFrames(pending)).toThrow(/desynced/) }) }) diff --git a/src/main/emulator/android/scrcpy-video-frame-parser.ts b/src/main/emulator/android/scrcpy-video-frame-parser.ts index bbada077cf9..fd702823350 100644 --- a/src/main/emulator/android/scrcpy-video-frame-parser.ts +++ b/src/main/emulator/android/scrcpy-video-frame-parser.ts @@ -3,11 +3,15 @@ // socket. The socket reader (scrcpy-stream-session) feeds chunks here; this file // has no I/O so the framing is unit-testable. +import type { RelayFrameBuffer } from '../../../shared/relay-frame-buffer' + const FRAME_HEADER_SIZE = 12 const CODEC_META_SIZE = 12 // scrcpy frames are well under this at the configured max_size; a larger // size means a desynced stream — fail fast instead of buffering toward OOM. const MAX_FRAME_BYTES = 16 * 1024 * 1024 +// Caps per-object overhead when a socket delivers one frame as many tiny chunks. +export const MAX_PENDING_CHUNKS = 1024 // Top two bits of the 64-bit PTS field carry packet flags. const CONFIG_FLAG = 1n << 63n const KEY_FRAME_FLAG = 1n << 62n @@ -47,33 +51,38 @@ export type ScrcpyVideoFrame = { data: Buffer } -export type ScrcpyFrameParseResult = { frames: ScrcpyVideoFrame[]; pending: Buffer } - -// Extracts complete frames from `pending + chunk`, returning the leftover bytes -// of any partially-received frame so the caller can prepend them to the next chunk. -export function parseScrcpyVideoFrames(pending: Buffer, chunk: Buffer): ScrcpyFrameParseResult { - const buffer = pending.length > 0 ? Buffer.concat([pending, chunk]) : chunk +// Leave partial frames queued so fragmented payloads are not recopied on every chunk. +export function parseScrcpyVideoFrames(buffer: RelayFrameBuffer): ScrcpyVideoFrame[] { const frames: ScrcpyVideoFrame[] = [] - let offset = 0 - while (buffer.length - offset >= FRAME_HEADER_SIZE) { - const meta = buffer.readBigUInt64BE(offset) - const size = buffer.readUInt32BE(offset + 8) + while (buffer.length >= FRAME_HEADER_SIZE) { + const header = buffer.peek(FRAME_HEADER_SIZE) + const meta = header.readBigUInt64BE(0) + const size = header.readUInt32BE(8) if (size > MAX_FRAME_BYTES) { throw new Error(`scrcpy frame size ${size} exceeds ${MAX_FRAME_BYTES}; stream desynced`) } - const dataStart = offset + FRAME_HEADER_SIZE - if (buffer.length - dataStart < size) { + if (buffer.length < FRAME_HEADER_SIZE + size) { break } + const packet = buffer.take(FRAME_HEADER_SIZE + size) frames.push({ config: (meta & CONFIG_FLAG) !== 0n, keyFrame: (meta & KEY_FRAME_FLAG) !== 0n, pts: meta & PTS_MASK, - data: Buffer.from(buffer.subarray(dataStart, dataStart + size)) + data: Buffer.from(packet.subarray(FRAME_HEADER_SIZE)) }) - offset = dataStart + size } - return { frames, pending: offset > 0 ? Buffer.from(buffer.subarray(offset)) : buffer } + if (frames.length > 0 && buffer.length > 0) { + const pendingHead = buffer.peek(1) + // Compact only mostly consumed allocations larger than the reusable Buffer slab. + if (pendingHead.buffer.byteLength > Math.max(Buffer.poolSize, pendingHead.length * 2)) { + buffer.append(Buffer.from(buffer.drain())) + } + } + if (buffer.chunkCount > MAX_PENDING_CHUNKS) { + buffer.append(buffer.drain()) + } + return frames } diff --git a/src/shared/relay-frame-buffer.ts b/src/shared/relay-frame-buffer.ts index a851426c6d8..77933fcf53b 100644 --- a/src/shared/relay-frame-buffer.ts +++ b/src/shared/relay-frame-buffer.ts @@ -7,6 +7,10 @@ export class RelayFrameBuffer { return this.bytes } + get chunkCount(): number { + return this.chunks.length - this.head + } + append(chunk: Buffer): void { this.chunks.push(chunk) this.bytes += chunk.length