diff --git a/bench/README.md b/bench/README.md index b214288f933..7053312d74a 100644 --- a/bench/README.md +++ b/bench/README.md @@ -432,19 +432,24 @@ unavailable CPU clock fail the run; timing never does. The nightly runs reconnects x 64 characters, and keeps its run directory as the `bench-browser` artifact. -The in-memory frame, retained-history, gap ledger, bandwidth allocation, and resume-history paths also run directly in Chromium: +The in-memory frame, retained-history, gap ledger, bandwidth allocation, cached-group +replay, and resume-history paths also run directly in Chromium: ```bash nix develop --command bun bench/browser/micro.ts --bench frames --out .tmp/frames.json base=/path/to/release current=. aa=. nix develop --command bun bench/browser/micro.ts --bench drift --out .tmp/drift.json before=/path/to/main current=. aa=. nix develop --command bun bench/browser/micro.ts --bench ledger --out .tmp/ledger.json before=/path/to/main current=. aa=. nix develop --command bun bench/browser/micro.ts --bench bandwidth --out .tmp/bandwidth.json before=/path/to/main current=. aa=. +nix develop --command bun bench/browser/micro.ts --bench replay --out .tmp/replay.json before=/path/to/main current=. aa=. nix develop --command bun bench/browser/micro.ts --bench resume-history --out .tmp/resume-history.json before=/path/to/main current=. aa=. ``` -Each variant uses its own workspace source and a persistent page, with one full +Each variant uses its own workspace source and a persistent, cross-origin isolated +page, so `performance.now()` resolves 5 µs rather than 100 µs, with one full warmup before three alternating measured rounds. `aa` must bundle identically to -`current`. The JSON preserves every CSV sample and the browser version. `frames` +`current`. The JSON preserves every CSV sample and the browser version. Replay +and resume-history also record each variant's main-thread GC count, pause and +reclaimed heap and the rows from one traced run after the timed rounds. `frames` runs the checkout's existing `js/net/bench/frames.ts`; `drift` uses the same fixture against every checkout, sweeping retained groups, publishers, and readers. Replay and group setup stay outside drift timing; each parked read checks its live @@ -463,23 +468,36 @@ Its `time_mode` column is 0 for untimed, 1 for millisecond timestamps, and 2 for mixed millisecond/microsecond timestamps, which exercise exact cross-scale comparison. +`replay` sweeps retained groups (64, 512, 2048) against subscribers (1, 8) for +monotonic, reverse, and shuffled arrivals. `join` rows time each subscriber +joining the cache and replaying every retained group; `append` rows time each of +64 arrivals fanned into every subscriber's unread backlog, landing inside the +backlog for reverse and shuffled orders. `max_us` is the slowest single operation, +the longest the event loop was held. An untimed pass first prints each shape's +`Array.prototype.sort` calls, elements and comparator visits, ahead of the timing +header so the nightly report keeps only timings as series. Every subscriber then +drains, requiring each sequence once, ascending, with its payload. Each shape then +closes its readers and track and advances its own manual clock past cache expiry, +outside timing, so completed shapes cannot accumulate through prune timers. + `resume-history` moves long-lived logical fronts across lite sessions over the in-memory transport, make-before-break as a session draining after GOAWAY does. It sweeps fronts, names read per route, and route count, with `stable` names (the same ones on every route) against `churn` (new names first answered by each route), each read by info lookup, fetch, or subscription and then closed. Once a -manual clock ages out every cache, full collections from Chromium's exposed `gc` -let weak references count the retired routes' sessions, transports, broadcast -handles, and delivered payload buffers still reachable. These counts are exact, -so any nonzero one fails the run once every sample is written, leaving a retaining -`before` revision's counts in the JSON. A concluded control must release everything, +manual clock ages out every cache, asynchronous major collections from Chromium's +exposed `gc` run without stack heap pointers, so dead stack slots cannot pin a route +even under tracing. Weak references count the retired routes' sessions, transports, +broadcast handles, and delivered payload buffers still reachable. These counts are exact, +so any nonzero one in a timed or traced pass fails the run once every sample is +written, leaving a retaining `before` revision's counts in the JSON. A concluded control must release everything, or the fixture fails. `size_bytes` is the JS heap the fronts and their serving -route hold, from Chromium's precise memory info; `serve_us` (clamped to the -page's coarse timer) and `read_us` include no setup. Bun's collector scans its +route hold, from Chromium's precise memory info; `serve_us` and +`read_us` include no setup. Bun's collector scans its stack conservatively, so the fixture refuses to run there. -The browser smoke lane preserves one measured round of all five fixtures, and drift, -ledger and bandwidth also run in the nightly JS benchmark leg. Timing is +The browser smoke lane preserves one measured round of all six fixtures, and drift, +ledger, bandwidth and replay also run in the nightly JS benchmark leg. Timing is observational, with the A/A spread showing the host's noise floor. ## Pattern and container coverage diff --git a/bench/browser/driver.ts b/bench/browser/driver.ts index 8f74dc838b3..9f2646f61c2 100644 --- a/bench/browser/driver.ts +++ b/bench/browser/driver.ts @@ -17,6 +17,7 @@ import type { Media } from "./page/media.ts"; import type { Dial, Mode, Racing } from "./page/racing.ts"; import type { Delivery, Transport } from "./page/transport.ts"; import { compare, type Row, summarize } from "./summary.ts"; +import { collections, GC_CATEGORIES, rendererPid, type TraceEvent, traceEvents } from "./trace.ts"; const { values } = parseArgs({ options: { @@ -275,52 +276,13 @@ async function open(variant: string, kind: string, counted = false): Promise context.close() }; + return { page, cdp, pid: await pagePid(page), errors, close: () => context.close() }; } // The renderer process behind a page, from the frame list a trace starts with. -async function rendererPid(page: Page): Promise { +async function pagePid(page: Page): Promise { await browser.startTracing(page, { categories: ["disabled-by-default-devtools.timeline"] }); - const events = traceEvents(await browser.stopTracing()); - const url = page.url(); - for (const event of events) { - if (event.name !== "TracingStartedInBrowser") continue; - const frames = (event.args?.data?.frames ?? []) as { url: string; processId: number }[]; - const frame = frames.find((frame) => frame.url === url); - if (frame) return frame.processId; - } - throw new Error(`no renderer process found for ${url}`); -} - -interface TraceEvent { - name: string; - ph: string; - pid: number; - tid: number; - dur?: number; - args?: { data?: { frames?: unknown }; usedHeapSizeBefore?: number; usedHeapSizeAfter?: number }; -} - -function traceEvents(buffer: Buffer): TraceEvent[] { - const trace = JSON.parse(buffer.toString()) as { traceEvents?: TraceEvent[] } | TraceEvent[]; - return Array.isArray(trace) ? trace : (trace.traceEvents ?? []); -} - -// GC on one renderer's main thread: total pause, count, and the JS heap each collection reclaimed. -function collections(events: TraceEvent[], pid: number) { - const gcs = events.filter( - (event) => event.pid === pid && event.tid === pid && (event.name === "MinorGC" || event.name === "MajorGC"), - ); - const reclaimed = gcs.reduce( - (sum, event) => sum + Math.max(0, (event.args?.usedHeapSizeBefore ?? 0) - (event.args?.usedHeapSizeAfter ?? 0)), - 0, - ); - return { - count: gcs.length, - minor: gcs.filter((event) => event.name === "MinorGC").length, - pause_ms: gcs.reduce((sum, event) => sum + (event.dur ?? 0), 0) / 1000, - reclaimed_bytes: reclaimed, - }; + return rendererPid(traceEvents(await browser.stopTracing()), page.url()); } async function metrics(opened: Opened) { @@ -397,9 +359,7 @@ async function trace( await run(); return undefined; } - await browser.startTracing(page, { - categories: ["devtools.timeline", "v8", "disabled-by-default-v8.gc", "disabled-by-default-devtools.timeline"], - }); + await browser.startTracing(page, { categories: GC_CATEGORIES }); let buffer: Buffer | undefined; try { await run(); diff --git a/bench/browser/micro.ts b/bench/browser/micro.ts index ba5df6ee595..8694098287c 100644 --- a/bench/browser/micro.ts +++ b/bench/browser/micro.ts @@ -2,6 +2,7 @@ import { resolve } from "node:path"; import { parseArgs } from "node:util"; import { chromium, type Page } from "playwright"; +import { collections, GC_CATEGORIES, rendererPid, traceEvents } from "./trace.ts"; import { workspace } from "./workspace.ts"; const { values, positionals } = parseArgs({ @@ -14,10 +15,16 @@ const { values, positionals } = parseArgs({ }, }); const benchmark = values.bench; -const expectedRows = ( - { frames: 26, drift: 36, ledger: 12, bandwidth: 44, "resume-history": 72 } as Record -)[benchmark]; -if (expectedRows === undefined) throw new Error("--bench must be frames, drift, ledger, bandwidth, or resume-history"); +// The rows each benchmark prints per run. +const ROWS: Record = { + frames: 26, + drift: 36, + ledger: 12, + bandwidth: 44, + replay: 72, + "resume-history": 72, +}; +if (!Object.hasOwn(ROWS, benchmark)) throw new Error(`--bench must be one of ${Object.keys(ROWS).join(", ")}`); const rounds = Number(values.rounds); if (!Number.isSafeInteger(rounds) || rounds < 1) throw new Error("--rounds must be a positive integer"); const roots = new Map(); @@ -48,7 +55,7 @@ for (const [name, root] of roots) { : undefined, ); build.onLoad( - { filter: /\/bench\/(frames|drift|ledger|bandwidth|resume-history)\.ts$/ }, + { filter: /\/bench\/(frames|drift|ledger|bandwidth|replay|resume-history)\.ts$/ }, async ({ path }) => { const source = await Bun.file(path).text(); if (benchmark !== "frames") @@ -83,8 +90,13 @@ const server = Bun.serve({ }); } if (!scripts.has(name)) return new Response("unknown variant", { status: 404 }); + // Cross-origin isolated, so performance.now() resolves 5 us instead of 100 us. return new Response(``, { - headers: { "Content-Type": "text/html" }, + headers: { + "Content-Type": "text/html", + "Cross-Origin-Opener-Policy": "same-origin", + "Cross-Origin-Embedder-Policy": "require-corp", + }, }); }, }); @@ -95,6 +107,18 @@ const browser = await chromium.launch({ args: benchmark === "resume-history" ? ["--js-flags=--expose-gc", "--enable-precise-memory-info"] : [], }); const samples: { round: number; variant: string; rows: string[] }[] = []; +const traced: { variant: string; rows: string[] }[] = []; +// Main-thread GC across one whole run per variant, from a traced pass after the timed rounds. +const gc: Record> = {}; +const save = () => + Bun.write( + values.out, + JSON.stringify( + { benchmark, browser: browser.version(), warmup: 1, roots: Object.fromEntries(roots), samples, traced, gc }, + null, + 2, + ), + ); try { const pages = new Map(); const rows = new Map(); @@ -110,34 +134,47 @@ try { void page.close(); }); await page.goto(`http://127.0.0.1:${server.port}/${name}`); + if (!(await page.evaluate("crossOriginIsolated"))) throw new Error(`${name} page is not cross-origin isolated`); await page.waitForFunction(() => typeof (globalThis as unknown as { measure: unknown }).measure === "function"); await page.evaluate("globalThis.measure()"); console.log(`warmed ${name}`); } + // One run of `variant`'s page, requiring every row. + const measure = async (variant: string) => { + rows.set(variant, []); + await pages.get(variant)?.evaluate("globalThis.measure()"); + const measured = rows.get(variant) ?? []; + if (measured.length !== ROWS[benchmark]) throw new Error(`incomplete rows: ${variant}`); + return measured; + }; for (let round = 0; round < rounds; round++) { const names = [...roots.keys()]; if (round % 2) names.reverse(); for (const variant of names) { - rows.set(variant, []); - await pages.get(variant)?.evaluate("globalThis.measure()"); - const measured = rows.get(variant); - if (measured?.length !== expectedRows) throw new Error(`incomplete rows: ${variant}`); - samples.push({ round, variant, rows: measured }); - await Bun.write( - values.out, - JSON.stringify( - { benchmark, browser: browser.version(), warmup: 1, roots: Object.fromEntries(roots), samples }, - null, - 2, - ), - ); + samples.push({ round, variant, rows: await measure(variant) }); + await save(); console.log(`measured ${round} ${variant}`); } } + // Traced apart from the timed rounds, so tracing overhead never enters a timing. + for (const [variant, page] of pages) { + if (benchmark !== "replay" && benchmark !== "resume-history") continue; + await browser.startTracing(page, { categories: GC_CATEGORIES }); + let buffer: Buffer | undefined; + try { + traced.push({ variant, rows: await measure(variant) }); + } finally { + buffer = await browser.stopTracing(); + } + const events = traceEvents(buffer); + gc[variant] = collections(events, rendererPid(events, page.url())); + await save(); + console.log(`traced ${variant}`); + } // Reachability is exact here, so a retired route still reachable fails the run, once every // variant's samples are written for comparison. if (benchmark === "resume-history") { - const retaining = samples.filter(({ rows }) => + const retaining = [...samples, ...traced].filter(({ rows }) => rows.some((row) => row .split(",") diff --git a/bench/browser/run.sh b/bench/browser/run.sh index 4f768742823..ca8d613244c 100755 --- a/bench/browser/run.sh +++ b/bench/browser/run.sh @@ -88,7 +88,7 @@ bun install --frozen-lockfile cd "$HERE" # The in-memory paths share Chromium installation and artifact retention with this lane. -for benchmark in frames drift ledger bandwidth resume-history; do +for benchmark in frames drift ledger bandwidth replay resume-history; do bun micro.ts --bench "$benchmark" --rounds 1 --out "$HARNESS_RUN/$benchmark.json" \ current="$WORKSPACE" aa="$WORKSPACE" done diff --git a/bench/browser/trace.ts b/bench/browser/trace.ts new file mode 100644 index 00000000000..4e529516851 --- /dev/null +++ b/bench/browser/trace.ts @@ -0,0 +1,51 @@ +/** Chrome trace parsing shared by the browser benchmark runners. */ + +/** Categories a GC-attributing trace records. */ +export const GC_CATEGORIES = [ + "devtools.timeline", + "v8", + "disabled-by-default-v8.gc", + "disabled-by-default-devtools.timeline", +]; + +export interface TraceEvent { + name: string; + ph: string; + pid: number; + tid: number; + dur?: number; + args?: { data?: { frames?: unknown }; usedHeapSizeBefore?: number; usedHeapSizeAfter?: number }; +} + +export function traceEvents(buffer: Buffer): TraceEvent[] { + const trace = JSON.parse(buffer.toString()) as { traceEvents?: TraceEvent[] } | TraceEvent[]; + return Array.isArray(trace) ? trace : (trace.traceEvents ?? []); +} + +/** The renderer process showing `url`, from the frame list a trace starts with. */ +export function rendererPid(events: TraceEvent[], url: string): number { + for (const event of events) { + if (event.name !== "TracingStartedInBrowser") continue; + const frames = (event.args?.data?.frames ?? []) as { url: string; processId: number }[]; + const frame = frames.find((frame) => frame.url === url); + if (frame) return frame.processId; + } + throw new Error(`no renderer process found for ${url}`); +} + +/** GC on one renderer's main thread: total pause, count, and the JS heap each collection reclaimed. */ +export function collections(events: TraceEvent[], pid: number) { + const gcs = events.filter( + (event) => event.pid === pid && event.tid === pid && (event.name === "MinorGC" || event.name === "MajorGC"), + ); + const reclaimed = gcs.reduce( + (sum, event) => sum + Math.max(0, (event.args?.usedHeapSizeBefore ?? 0) - (event.args?.usedHeapSizeAfter ?? 0)), + 0, + ); + return { + count: gcs.length, + minor: gcs.filter((event) => event.name === "MinorGC").length, + pause_ms: gcs.reduce((sum, event) => sum + (event.dur ?? 0), 0) / 1000, + reclaimed_bytes: reclaimed, + }; +} diff --git a/js/justfile b/js/justfile index 9e021b09830..c69c42d0672 100644 --- a/js/justfile +++ b/js/justfile @@ -57,7 +57,7 @@ test-changed list: coverage package="": python3 ../sh/js/coverage.py '{{ package }}' -# Run one net benchmark: broadcasts, forward, requests, fills, reader, frames, track, resume, observed, drift, ledger, bandwidth, varint, lite-varint, priority, subscribe-drop, refresh, or compression. +# Run one net benchmark: broadcasts, forward, requests, fills, reader, frames, track, resume, observed, drift, ledger, bandwidth, replay, varint, lite-varint, priority, subscribe-drop, refresh, or compression. bench name: bun install --frozen-lockfile bun net/bench/{{ name }}.ts diff --git a/js/net/bench/replay.ts b/js/net/bench/replay.ts new file mode 100644 index 00000000000..9fcf10cd75f --- /dev/null +++ b/js/net/bench/replay.ts @@ -0,0 +1,171 @@ +/** Sweep cached-group replay to joining subscribers, and arrivals into unread backlogs. */ +import { Manual, Milli, Nano } from "@moq/clock"; +import { Track } from "@moq/net"; + +type Order = "monotonic" | "reverse" | "mixed"; +const ORDERS: Order[] = ["monotonic", "reverse", "mixed"]; +type Kind = "join" | "append"; +const PAYLOAD = 8; +// Arrivals timed per append row, so a sample spans enough work to read. +const APPENDS = 64; +const REPS = 5; +// Every retained group, however old: replay is the work being measured. +const SUBSCRIPTION = { groups: { start: { included: 0 } }, maxDelay: null }; + +// The arrival order of sequences 0..count-1: ascending, descending, or a seeded shuffle. +function arrivals(count: number, order: Order): number[] { + const sequences = Array.from({ length: count }, (_, i) => i); + if (order === "reverse") sequences.reverse(); + if (order === "mixed") { + let seed = 42; + for (let i = count - 1; i > 0; i--) { + seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0; + const j = seed % (i + 1); + [sequences[i], sequences[j]] = [sequences[j], sequences[i]]; + } + } + return sequences; +} + +// Write one finished group whose payload names its sequence, so delivery can be checked. +function publish(track: Track.Producer, sequence: number): void { + const group = track.createGroup(sequence); + const payload = new Uint8Array(PAYLOAD); + new DataView(payload.buffer).setUint32(0, sequence); + group.writeFrame({ payload }); + group.finish(); +} + +// Read everything a subscriber buffered, requiring every sequence exactly once, ascending. +function drain(subscriber: Track.Subscriber, count: number): void { + let next = 0; + for (let group = subscriber.tryRecvGroup(); group; group = subscriber.tryRecvGroup(), next++) { + if (group.sequence !== next) throw new Error(`group ${group.sequence} delivered, expected ${next}`); + const payload = group.tryReadFrame()?.payload; + if (payload?.byteLength !== PAYLOAD || new DataView(payload.buffer, payload.byteOffset).getUint32(0) !== next) + throw new Error(`group ${next} payload mismatch`); + } + if (next !== count) throw new Error(`${next} of ${count} groups delivered`); +} + +type Counts = { sorts: number; elements: number; compares: number }; + +// Run `fn` counting Array.prototype.sort calls, the elements handed to them, and comparator visits. +function counted(fn: () => void): Counts { + const sort = Array.prototype.sort; + const counts = { sorts: 0, elements: 0, compares: 0 }; + Array.prototype.sort = function (this: T[], compare?: (a: T, b: T) => number): T[] { + counts.sorts++; + counts.elements += this.length; + if (!compare) return sort.call(this) as T[]; + return sort.call(this, (a: T, b: T) => { + counts.compares++; + return compare(a, b); + }) as T[]; + }; + try { + fn(); + } finally { + Array.prototype.sort = sort; + } + return counts; +} + +// Runs each measured operation through `op`. +type Operations = (op: (fn: () => void) => void) => void; + +// A track retaining `groups` groups that arrived in `order`, then either `subscribers` joins that +// replay all of them, or APPENDS more arrivals into that many unread backlogs; reverse and mixed +// appends land inside the backlog. `measure` runs the operations. Every subscriber then drains, +// so skipped or reordered work fails the run. +async function shape( + kind: Kind, + groups: number, + order: Order, + subscribers: number, + measure: (ops: Operations) => void, +) { + // Freeze cache age during the operation, then expire this shape before starting another. + const clock = new Manual(); + const installed = clock.install(); + const total = kind === "join" ? groups : groups + APPENDS; + const sequences = arrivals(total, order); + const track = new Track.Producer(kind).accept(); + const readers: Track.Subscriber[] = []; + try { + if (kind === "append") for (let i = 0; i < subscribers; i++) readers.push(track.subscribe(SUBSCRIPTION)); + for (const sequence of sequences.slice(0, groups)) publish(track, sequence); + measure((op) => { + if (kind === "join") { + for (let i = 0; i < subscribers; i++) op(() => readers.push(track.subscribe(SUBSCRIPTION))); + } else { + for (const sequence of sequences.slice(groups)) op(() => publish(track, sequence)); + } + }); + for (const reader of readers) drain(reader, total); + } finally { + for (const reader of readers) reader.close(); + track.finish(); + track.close(); + try { + await clock.advance(Nano.fromMilli(Milli(60_000))); + } finally { + installed.close(); + } + } +} + +const median = (values: number[]) => values.sort((a, b) => a - b)[values.length >> 1]; + +/** + * Print sort counts per shape from an untimed pass, then timing per shape: `join` rows time each + * subscriber joining the retained groups, `append` rows each arrival into every subscriber's + * unread backlog. The counts table precedes the timing header, so the nightly report records only + * the timings as series. `max_us` is the slowest single operation, which held the event loop that + * long. Timings are the median of REPS runs. + */ +export async function run(): Promise { + const shapes: [Kind, number, Order, number][] = []; + for (const groups of [64, 512, 2048]) + for (const order of ORDERS) + for (const subscribers of [1, 8]) + for (const kind of ["join", "append"] as const) shapes.push([kind, groups, order, subscribers]); + + console.log("groups,kind,order,subscribers,sorts,sorted_elements,compares"); + for (const [kind, groups, order, subscribers] of shapes) { + let counts: Counts | undefined; + await shape(kind, groups, order, subscribers, (ops) => { + counts = counted(() => ops((op) => op())); + }); + if (!counts) throw new Error("shape was not counted"); + console.log(`${groups},${kind},${order},${subscribers},${counts.sorts},${counts.elements},${counts.compares}`); + } + + console.log("groups,kind,order,subscribers,us_per_op,max_us"); + for (const [kind, groups, order, subscribers] of shapes) { + const perOp: number[] = []; + const slowest: number[] = []; + for (let rep = 0; rep < REPS; rep++) { + let count = 0; + let max = 0; + let elapsed = 0; + await shape(kind, groups, order, subscribers, (ops) => { + const start = performance.now(); + ops((op) => { + const begin = performance.now(); + op(); + max = Math.max(max, performance.now() - begin); + count++; + }); + elapsed = performance.now() - start; + }); + perOp.push((elapsed * 1000) / count); + slowest.push(max * 1000); + } + console.log( + `${groups},${kind},${order},${subscribers},${median(perOp).toFixed(1)},${median(slowest).toFixed(0)}`, + ); + } +} + +if (import.meta.main) await run(); diff --git a/js/net/bench/resume-history.ts b/js/net/bench/resume-history.ts index 05e076e711e..4659f679eef 100644 --- a/js/net/bench/resume-history.ts +++ b/js/net/bench/resume-history.ts @@ -36,13 +36,16 @@ const url = new URL("https://localhost/resume-history"); type History = "stable" | "churn"; type Access = "lookup" | "fetch" | "subscribe"; -/** Full collections, each from a fresh task so no frame of the caller keeps a target. */ +/** Full collections with no stack heap pointers, so dead stack slots cannot pin a target. */ async function collect(): Promise { - const gc = (globalThis as { gc?: () => void }).gc; + const gc = (globalThis as { gc?: (options: { type: "major"; execution: "async" }) => Promise }).gc; if (!gc) throw new Error("resume-history needs a precise collector: run it with bench/browser/micro.ts"); for (let pass = 0; pass < 3; pass++) { - await new Promise((resolve) => setTimeout(resolve, 0)); - gc(); + // V8's synchronous gc scans the embedder stack conservatively even from a fresh task. + // Async gc runs with StackState::kNoHeapPointers, including while tracing is enabled. + const collected = gc({ type: "major", execution: "async" }); + if (!(collected instanceof Promise)) throw new Error("resume-history needs Chromium's asynchronous gc"); + await collected; } } diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 75377bdec6d..c3ab1e5fed3 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -3566,6 +3566,39 @@ test("an untimed track takes groups in any order", () => { } }); +test("replay and backlogs deliver sparse out-of-order arrivals ascending without sorting", () => { + const producer = new TrackProducer("test").accept({}); + const all = { groups: { start: { included: 0 } }, maxDelay: null }; + const write = (sequence: number) => { + const group = producer.createGroup(sequence); + group.writeFrame({ payload: enc.encode(`${sequence}`) }); + group.finish(); + }; + const drain = (reader: ReturnType) => { + const seen: number[] = []; + for (let group = reader.tryRecvGroup(); group; group = reader.tryRecvGroup()) { + const frame = group.tryReadFrame(); + expect(frame && dec.decode(frame.payload)).toBe(`${group.sequence}`); + seen.push(group.sequence); + } + return seen; + }; + const sort = spyOn(Array.prototype, "sort"); + try { + const live = producer.subscribe(all); + for (const sequence of [40, 10, 30, 0, 20]) write(sequence); + const joined = producer.subscribe(all); + // Arrivals below, between and above the joiner's replayed cache. + for (const sequence of [25, 50, 5, 35]) write(sequence); + expect(sort).not.toHaveBeenCalled(); + const expected = [0, 5, 10, 20, 25, 30, 35, 40, 50]; + expect(drain(live)).toEqual(expected); + expect(drain(joined)).toEqual(expected); + } finally { + sort.mockRestore(); + } +}); + // A received track as a wire subscription leaves it: the feed starts at group 0 and its end is // pending, so lower groups may still arrive behind a higher one. function pendingTail(end: number, maxDelay = Milli(60_000)) { diff --git a/js/net/src/track.ts b/js/net/src/track.ts index d26162c0a70..290d25aea5b 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -513,6 +513,15 @@ function timelineInsert(timeline: GroupConsumer[], group: GroupConsumer): void { else timeline.splice(index, 0, group); } +// Add a group to a sink's sequence-sorted unread buffer, after any group with the same +// sequence, as a stable sort would. Checked for the usual in-order append first, so neither a +// replayed cache nor a growing backlog re-sorts the buffer on every arrival. +function bufferInsert(groups: GroupConsumer[], group: GroupConsumer): void { + const last = groups.at(-1); + if (!last || last.sequence <= group.sequence) groups.push(group); + else groups.splice(timelineIndex(groups, group.sequence + 1), 0, group); +} + // The shared state behind a Producer / Subscriber pair. Package-internal // wiring, unexported so it never appears in the published type declarations. class TrackState { @@ -1125,10 +1134,7 @@ export class Producer { void dst.readable().then(() => sink.timelineChanged.update((revision) => revision + 1)); sink.latest = Math.max(sink.latest ?? 0, dst.sequence); sink.received.set(Math.max(sink.received.peek(), dst.sequence + 1)); - sink.groups.mutate((groups) => { - groups.push(dst); - groups.sort((a, b) => a.sequence - b.sequence); - }); + sink.groups.mutate((groups) => bufferInsert(groups, dst)); } // Drop a cached group's mirror from every sink so no consumer can pin it. diff --git a/quest/m2/performance-survey/README.md b/quest/m2/performance-survey/README.md index 0a4585c5b58..a3eb9d83aa8 100644 --- a/quest/m2/performance-survey/README.md +++ b/quest/m2/performance-survey/README.md @@ -122,6 +122,15 @@ reproduced the missing `moq-archive/producer` selection. The existing [CI audit PR](https://github.com/moq-dev/moq/pull/6266) already fixes it. No duplicate quest or workflow edit belongs here. +Cached JS group joins mattered. In Chromium 156, a subscriber joining 2,048 +retained groups ran about 2.1M sort comparator visits and took 12-18 ms; an +arrival into 8 subscribers' 2,048-group unread backlogs took about 100-113 µs. +Ordered insertion removed every sort: the same join took 0.5-2 ms and the +arrival 9-11 µs. Those original timings used a whole-run frozen clock; the +fixture now expires each shape outside timing so completed tracks cannot +accumulate through prune timers. The original whole-run GC totals are withdrawn. +The `js/replay` nightly series and the browser smoke's `replay.json` protect it. + ### Kernel and placement assessment - Already implemented in [moq-uring](/rs/moq-uring/README.md): multishot @@ -187,7 +196,6 @@ No duplicate quest or workflow edit belongs here. - [Task capacity after bursts](/quest/m2/performance-survey/task-capacity.md) - memory a task set keeps after peak occupancy subsides - [QUIC receive-state reuse](/quest/m2/performance-survey/quic-receive-reuse.md) - reassembly metadata retained by recycled completed streams - [Cache metadata](/quest/m2/performance-survey/cache-metadata.md) - bookkeeping capacity after cached groups expire -- [Cached JS group joins](/quest/m2/performance-survey/js-cached-groups.md) - repeated sorting while replaying retained groups - [PCM batch processing](/quest/m2/performance-survey/audio-batches.md) - repeated tail movement during encoding and resampling - [Fragmented MKV import](/quest/m2/performance-survey/mkv-fragments.md) - accumulated-buffer copies and parser replay - [HLS request fan-out](/quest/m2/performance-survey/hls-serving.md) - repeated fetch, remux and playlist construction per viewer diff --git a/quest/m2/performance-survey/js-cached-groups.md b/quest/m2/performance-survey/js-cached-groups.md deleted file mode 100644 index deba6ffccff..00000000000 --- a/quest/m2/performance-survey/js-cached-groups.md +++ /dev/null @@ -1,32 +0,0 @@ -# [M] Joining a cached JS track avoids repeated full sorts - -## Goal - -A subscriber joining retained groups, or falling behind while groups arrive, -does no unnecessary whole-backlog sorting. - -## Plan - -[Track replay](/js/net/src/track.ts) mirrors every cached group on join; -each mirror pushes and sorts the entire unread group array. Even ordered -replay sorts arrays of sizes 1 through G. -The separate timeline index already has an append fast path and ordered -insertion. Existing track timing drains immediately and misses growing queues. - -Benchmark retained groups × joining subscribers, then unread backlog × active -subscribers, in a real browser. Hold payload work fixed where possible and -include monotonic, reverse and mixed arrivals. Report comparator visits, -join/append latency, allocation/GC cost and event-loop delay. Prove replay -contents, ascending delivery, sparse arrivals, eviction, close and retained -mirror lifetimes remain identical. Reuse existing ordering machinery if it -makes the result simpler. - -This is distinct from subscription update traffic in -[moq-dev/moq#6391](https://github.com/moq-dev/moq/pull/6391) and the storage -contract of the JS group tail and chunked frames from -[moq-dev/moq#7411](https://github.com/moq-dev/moq/pull/7411). - -Follow the [survey's benchmark contract](/quest/m2/performance-survey/README.md): -measure the baseline before optimizing, retain paired evidence, verify equivalent -work, schedule coverage nightly, and accept a measured no-win. Expected public -API and wire impact: none; report the actual impact in the implementation PR. diff --git a/sh/gh/nightly.sh b/sh/gh/nightly.sh index 0310b6e643f..d682cd95a3c 100755 --- a/sh/gh/nightly.sh +++ b/sh/gh/nightly.sh @@ -139,7 +139,7 @@ case "$1" in done ;; js) - for benchmark in broadcasts requests fills forward reader frames drift ledger bandwidth varint lite-varint track resume priority subscribe-drop refresh observed compression; do + for benchmark in broadcasts requests fills forward reader frames drift ledger bandwidth replay varint lite-varint track resume priority subscribe-drop refresh observed compression; do javascript "$benchmark" just js bench "$benchmark" done javascript catalog just js catalog-bench