Daan MeijerandClaude Opus 5.5
168605148f
decodeFrames joined each network piece (~64 KB) onto the frame so far: a chunk of MBs copied over and over, 5.4% of a worker's time in an end-to-end profile (the bench reads files, so it never showed). Pieces are now joined once a frame is complete. A batch post's gzip (1.6%) runs on a thread-pool thread. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
59 lines
2.2 KiB
TypeScript
59 lines
2.2 KiB
TypeScript
/**
|
|
* A range's chunks on the wire (REBUILD.md, prefetch): one streaming
|
|
* response, a frame per chunk: from (8 bytes), to (8 bytes), length
|
|
* (4 bytes), then the chunk as the news server sent it (compressed XOVER).
|
|
*/
|
|
export interface Frame {
|
|
from: number
|
|
to: number
|
|
data: Buffer
|
|
}
|
|
|
|
const HEADER = 20
|
|
|
|
/** An empty frame, sent while the next chunk is still being fetched, so the response doesn't go quiet. */
|
|
export const KEEPALIVE: Frame = { from: 0, to: 0, data: Buffer.alloc(0) }
|
|
|
|
export function isKeepalive(frame: Frame) {
|
|
return frame.data.length === 0 && frame.from === 0
|
|
}
|
|
|
|
export function encodeFrame(frame: Frame): Buffer {
|
|
const head = Buffer.alloc(HEADER)
|
|
head.writeBigUInt64BE(BigInt(frame.from), 0)
|
|
head.writeBigUInt64BE(BigInt(frame.to), 8)
|
|
head.writeUInt32BE(frame.data.length, 16)
|
|
return Buffer.concat([head, frame.data])
|
|
}
|
|
|
|
/**
|
|
* The frames of a byte stream, as they complete. Pieces are kept in a list
|
|
* and joined once a frame is complete: joining each piece onto what came
|
|
* before copied a chunk of MBs once per 64 KB (~5% of a worker's time;
|
|
* LOG.md 2026-10-09).
|
|
*/
|
|
export async function* decodeFrames(stream: AsyncIterable<Uint8Array> | Iterable<Uint8Array>): AsyncGenerator<Frame> {
|
|
let pieces: Buffer[] = []
|
|
let length = 0
|
|
/** Bytes needed before the next frame can be cut: its header first, then all of it. */
|
|
let needed = HEADER
|
|
for await (const piece of stream) {
|
|
pieces.push(Buffer.isBuffer(piece) ? piece : Buffer.from(piece.buffer, piece.byteOffset, piece.byteLength))
|
|
length += piece.byteLength
|
|
while (length >= needed) {
|
|
let buffer = pieces.length === 1 ? pieces[0] : Buffer.concat(pieces, length)
|
|
pieces = [buffer]
|
|
if (needed === HEADER) {
|
|
needed = HEADER + buffer.readUInt32BE(16)
|
|
if (length < needed) break
|
|
}
|
|
yield { from: Number(buffer.readBigUInt64BE(0)), to: Number(buffer.readBigUInt64BE(8)), data: buffer.subarray(HEADER, needed) }
|
|
buffer = buffer.subarray(needed)
|
|
pieces = buffer.length > 0 ? [buffer] : []
|
|
length = buffer.length
|
|
needed = HEADER
|
|
}
|
|
}
|
|
if (length > 0) throw new Error(`Stream ended inside a frame (${length} bytes left)`)
|
|
}
|