Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204// Many streams run at once, each from its own pooled engine instance (src/core.js's// per-slot pool, which keeps released instances until a timer trims it). Interleaves// dozens of streams one chunk at a time (string and byte chunks, every chunk size 1-23),// opens and closes streams mid-flight so instances get released and reused, and grows// every memory region (input, output, frame stack, errors) while other// streams hold instances of their own. Each stream's output and errors must equal its own// whole-string result, and a repeated burst must not keep instantiating more engines than// the pool already settled on. Also proves a reused instance actually comes back clean:// streams that finish at staggered times hand their instance to a newcomer while siblings// are still open, and a stream abandoned mid-namespace/class/expression (frame stack,// anchors, and errors all non-empty) must leave its instance fit for the very next acquire.// Usage: node test/concurrent-streams.mjsimport assert from "node:assert/strict";import fs from "node:fs";
const { transformSync, engine } = await import("../src/index.js");
const lines = (f) => fs.readFileSync(f, "utf8").split("\n").filter(Boolean).map((l) => JSON.parse(l));const small = [ ...lines("corpus/adversarial.jsonl"), ...lines("corpus/regressions.jsonl"), ...lines("corpus/downlevel.jsonl").map((c) => c.src),].filter((s) => s.length < 3000);const dom = fs.readFileSync("node_modules/typescript/lib/lib.dom.d.ts", "utf8");// Inputs that grow one region each: frames (deep nesting), errors (many non-erasable constructs),// input retained under an anchor (a long namespace), and plain input/output volume.const big = [ "let a = " + "(".repeat(6000) + "1" + ")".repeat(6000) + ";\n", "enum E { A }\nlet x: number = 1;\n".repeat(3000), "namespace N {\n" + " export const x: number = 1;\n".repeat(8000) + "}\n", dom.slice(0, 400000),];
function whole(mode, src) { const errors = []; try { return { out: transformSync(src, { mode, onError: (e) => errors.push(e.start, e.end) }), errors }; } catch (e) { return { threw: e.message }; }}
const enc = new TextEncoder();// A stream over src in chunks of `size` characters, sent as UTF-8 bytes when `bytes` (except for a leading BOM,// which the byte path drops by design).function stream(mode, src, size, bytes) { if (src.startsWith("")) bytes = false; const parts = []; for (let i = 0; i < src.length; i += size) parts.push(bytes ? enc.encode(src.slice(i, i + size)) : src.slice(i, i + size)); const errors = []; const gen = transformSync(parts, { mode, onError: (e) => errors.push(e.start, e.end) }); return { gen, out: "", errors, src, threw: undefined, done: false };}
// Advances every live stream by one chunk, round-robin, until all finish. `spawn` may add streams// mid-flight; `cut` may abandon one early (its generator's return() closes its instance).function interleave(live, spawn) { let round = 0; while (live.some((s) => !s.done)) { for (const s of live) { if (s.done) continue; try { const r = s.gen.next(); if (r.done) s.done = true; else s.out += r.value; } catch (e) { s.threw = e.message; s.done = true; } } spawn?.(live, round++); }}
function check(mode, live, label) { for (const s of live) { if (s.cut) continue; const want = whole(mode, s.src); if (want.threw) { assert.ok(s.threw, `${label}: stream should throw like the whole-string call`); continue; } assert.equal(s.threw, undefined, `${label}: stream threw ${s.threw}`); assert.equal(s.out, want.out, `${label}: interleaved output differs from the whole-string result`); assert.deepEqual(s.errors, want.errors, `${label}: interleaved errors differ from the whole-string result`); }}
// Counts fresh WebAssembly.Instance creations, so the burst section below can check that// the pool settles instead of re-instantiating on every repeat.let instances = 0;const OrigInstance = WebAssembly.Instance;WebAssembly.Instance = class extends OrigInstance { constructor(...args) { super(...args); instances++; }};
for (const mode of ["erase", "downlevel"]) { engine(mode);
// 1. Every small input at once, each with its own chunk size (1-23) and chunk kind // (string or bytes), all interleaved concurrently: each one must hold its own pooled // instance rather than clobbering another stream's parse state. let live = small.map((src, i) => stream(mode, src, 1 + (i % 23), i % 2 === 1)); interleave(live); check(mode, live, `${mode} small x${live.length}`);
// 2. Big inputs that grow the frame stack, error list, and input/output regions, // interleaved with small ones; streams join and leave mid-flight (including one cut off // early via the generator's return(), which must release its instance rather than leak it). live = big.map((src, i) => stream(mode, src, 4096 + i * 1000, i % 2 === 1)); let next = 0; interleave(live, (all, round) => { if (round % 3 === 0 && next < small.length) all.push(stream(mode, small[next++], 7, false)); if (round % 5 === 4) { const victim = all.find((s) => !s.done && !s.cut && s.src.length < 3000); if (victim) { victim.gen.return(); victim.cut = victim.done = true; } } }); check(mode, live, `${mode} big+churn x${live.length}`);
// 3. A burst of concurrent streams past the pool's idle minimum must settle: once the pool // has kept what a burst released, a repeated, identically-shaped burst must not keep // instantiating more and more engines. (A pool that trims could still // instantiate some per burst — that's the pool working as designed, not a leak; what must // not happen is the count climbing burst over burst.) const burst = () => { const l = big.concat(small.slice(0, 60)).map((src, i) => stream(mode, src, 5000, i % 2 === 1)); interleave(l); check(mode, l, `${mode} burst`); }; burst(); const afterFirst = instances; burst(); const createdSecond = instances - afterFirst; burst(); const createdThird = instances - afterFirst - createdSecond; assert.equal(createdThird, createdSecond, `${mode}: a repeated burst must settle, not keep instantiating more instances each time`);
// 4. Dirty reuse: an instance release() takes back must come back genuinely clean to // whoever acquire()s it next, even while sibling streams are still open. Two ways an // instance gets released dirty: // // (a) Staggered reuse: streams with small, varied chunk sizes finish at staggered times // while others remain mid-flight; a new stream opened right after one finishes is very // likely handed the instance that sibling just released, with others still live around it. { const group = small.slice(0, 50).map((src, i) => stream(mode, src, 1 + (i % 5), i % 3 === 0)); let joined = 50; interleave(group, (all) => { // Every time at least one sibling has finished this round, throw a fresh stream at the // pool immediately, so its acquire() lands on a just-freed instance while others are // still open around it. if (joined < small.length && all.some((s) => s.done)) { group.push(stream(mode, small[joined], 1 + (joined % 5), joined % 3 === 1)); joined++; } }); check(mode, group, `${mode} staggered reuse`); }
// (b) Abandon mid-construct: a stream cut off with `.return()` deep inside a namespace, a // class, and an ambiguous expression leaves its frame stack, anchors, and reported errors // all non-empty when release() takes the instance back. `held` streams stay open // throughout, so the dirtied instance is the *only* thing released in between, and the // pool hands the most recently released instance to the very next acquire(). { const held = Array.from({ length: 6 }, () => stream(mode, small[0], 4096, false)); for (const s of held) { const r = s.gen.next(); if (r.done) s.done = true; else s.out += r.value; }
const dirty = "namespace Deep {\n export class C {\n static v = " + "(".repeat(200) + "1" + ")".repeat(80) + ";\n static w = a ?? b as any as unknown && 2;\n"; const d = stream(mode, dirty, 11, false); for (let i = 0; i < 10 && !d.done; i++) { const r = d.gen.next(); if (r.done) d.done = true; } assert.equal(d.done, false, `${mode}: the dirty stream should still be mid-construct when abandoned`); d.gen.return(); // abandoned inside the namespace/class/expression; the instance releases dirty
const fresh = stream(mode, small[1], 9, false); interleave([fresh]); check(mode, [fresh], `${mode} dirty-reuse`);
for (const s of held) interleave([s]); check(mode, held, `${mode} dirty-reuse holders`); }}WebAssembly.Instance = OrigInstance;console.log(`concurrent-streams: ok (${small.length} small + ${big.length} large streams interleaved per mode across the pool; bursts settle)`);