Merge main

This commit is contained in:
Chris Tate
2026-10-15 19:50:35 -05:00
69 changed files with 5682 additions and 428 deletions
+1
View File
@@ -18,6 +18,7 @@ Application-shaped workloads for measuring the speed of compiled executables. Ea
| `value-objects` | cpu | small immutable records and class instances: vector helpers, method chains, particle updates |
| `optional-fields` | cpu | records and class instances with `number \| undefined` and other scalar-union fields |
| `record-boundaries` | cpu | records built from unchecked array reads passed to and returned from annotated functions |
| `number-maps` | cpu | number-keyed Map/Set: a dense-id table, sparse ids in many small maps, get/has/set mixes |
| `validate-errors` | cpu | throw/catch on a fraction of inputs, small call chains |
| `csv-numbers` | cpu | `split`, `Number`, `parseInt`, `toFixed` |
| `log-lines` | cpu | 200k immediate `console.log`/`process.stdout.write` lines into the runner's pipe |
+7
View File
@@ -105,6 +105,13 @@
"8"
]
},
{
"name": "number-maps",
"entry": "workloads/number-maps.ts",
"args": [
"4"
]
},
{
"name": "validate-errors",
"entry": "workloads/validate-errors.ts",
@@ -0,0 +1,90 @@
// Number-keyed Map and Set work: a node table indexed by dense ids, scope
// tables keyed by sparse ids from one global counter, and caches that mix
// get, has and set on the same keys.
interface Item {
id: number;
parent: number;
flags: number;
}
// Dense ids: items 0..n-1, walked through their parent chains.
function denseIds(n: number, rounds: number): number {
const items = new Map<number, Item>();
for (let id = 0; id < n; id++) items.set(id, { id, parent: id === 0 ? -1 : (id * 7) >> 3, flags: id & 15 });
let total = 0;
for (let r = 0; r < rounds; r++) {
for (let id = r & 7; id < n; id += 8) {
let item = items.get(id);
let depth = 0;
while (item !== undefined && depth < 4) {
total += item.flags;
item = items.get(item.parent);
depth++;
}
}
}
return total;
}
// Sparse ids: one counter hands out ids to many scopes, so each scope's
// keys are spread far apart. Lookups test membership, then read.
function sparseIds(scopes: number, perScope: number, rounds: number): number {
const tables: Map<number, Item>[] = [];
const probes: number[] = [];
let next = 1000;
for (let s = 0; s < scopes; s++) tables.push(new Map<number, Item>());
for (let i = 0; i < perScope; i++) {
for (let s = 0; s < scopes; s++) {
const id = next;
next += 1 + ((id * 31) & 3);
tables[s]!.set(id, { id, parent: s, flags: id & 7 });
if ((i & 1) === 0) probes.push(id);
else probes.push(id + 1);
}
}
let total = 0;
for (let r = 0; r < rounds; r++) {
for (let p = 0; p < probes.length; p++) {
const id = probes[p]!;
const table = tables[p % scopes]!;
if (table.has(id)) total += table.get(id)!.flags;
else total++;
}
}
return total;
}
// Mixed get/set: counting relation keys, a memo cache filled on misses,
// and a visited set.
function mixed(ids: number[], rounds: number): number {
const counts = new Map<number, number>();
const memo = new Map<number, number>();
let total = 0;
for (let r = 0; r < rounds; r++) {
const visited = new Set<number>();
for (let i = 0; i < ids.length; i++) {
const id = ids[i]!;
const relation = id * 4 + (r & 3);
counts.set(relation, (counts.get(relation) ?? 0) + 1);
let cached = memo.get(id);
if (cached === undefined) {
cached = (id * 2654435761) % 1000;
memo.set(id, cached);
}
total += cached;
if (!visited.has(id)) {
visited.add(id);
total++;
}
}
}
return total + counts.size + memo.size;
}
const scale = Number(process.argv[2] ?? "1");
const ids: number[] = [];
for (let i = 0; i < 20000; i++) ids.push(5000 + ((i * 7919) % 12000) * 3);
const dense = denseIds(50000, Math.floor(40 * scale));
const sparse = sparseIds(64, 40, Math.floor(400 * scale));
const mix = mixed(ids, Math.floor(30 * scale));
console.log("dense", dense, "sparse", sparse, "mixed", mix);
@@ -1 +1 @@
- **Streams and readline:** Stream chunks are Buffers or strings. Object-mode streams of other values, `Readable.from` over arrays of other values, the `Readable` iterator helpers (`map`, `filter`, `toArray`, and the others), `stream.compose`, and generator stages in `pipeline` are unsupported. `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags. Stream constructors accept a `signal` option in an options object literal, and `stream.addAbortSignal` accepts any stream. `node:readline` and `node:readline/promises` interfaces read `process.stdin` or a readable stream without terminal mode: `question`, `prompt`, `write`, `pause`, `resume`, `'line'`/`'close'`/`'pause'`/`'resume'`/`'error'` listeners, and `for await` compile. Terminal mode, `completer`, history options, and chained listener registration are unsupported.
- **Streams and readline:** Stream chunks are Buffers or strings. Object-mode streams of other values, `Readable.from` over arrays of other values or over generators and other iterables, and generator stages in `pipeline` are unsupported. `Readable.from` accepts `objectMode` and `highWaterMark` options. The `Readable` operators (`map`, `filter`, `forEach`, `toArray`, `some`, `find`, `every`, `flatMap`, `drop`, `take`, and `reduce`) accept `concurrency`, `highWaterMark`, and `signal` options; their callbacks take only the chunk, `map` returns strings or Buffers, and `flatMap` returns strings or arrays of them. `readable.compose` is unsupported. `stream.compose` accepts two or more streams, or one `Duplex`; functions, iterables, and web streams as stages are unsupported, and the `ERR_INVALID_ARG_VALUE` message for a stream that is no longer readable or writable names the stream's class instead of printing the stream. `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags. Stream constructors accept a `signal` option in an options object literal, and `stream.addAbortSignal` accepts any stream. `node:readline` and `node:readline/promises` interfaces read `process.stdin` or a readable stream without terminal mode: `question`, `prompt`, `write`, `pause`, `resume`, `'line'`/`'close'`/`'pause'`/`'resume'`/`'error'` listeners, and `for await` compile. Terminal mode, `completer`, history options, and chained listener registration are unsupported.
@@ -2,6 +2,6 @@
"schemaVersion": 3,
"nodeVersion": "24.22.0",
"nodeCommit": "f14cd3aa7bf9ca8df2cc5b8bb14beb0f335b1d93",
"artifactVersion": "c7c956da1d50af685bf1",
"artifactVersion": "cce4b946b6f8d3072e83",
"rowCount": 3880
}
+47 -47
View File
@@ -659,9 +659,9 @@
"apiEntries": 140,
"static": {
"supported": 0,
"partial": 38,
"partial": 53,
"refused": 0,
"not-implemented": 102,
"not-implemented": 87,
"by-design": 0,
"unreviewed": 0,
"not-applicable": 0
@@ -61525,9 +61525,9 @@
"inherited": true
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -61640,9 +61640,9 @@
"inherited": true
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -62031,9 +62031,9 @@
"inherited": true
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -62146,9 +62146,9 @@
"inherited": true
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63388,9 +63388,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63411,9 +63411,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63434,9 +63434,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63457,9 +63457,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63480,9 +63480,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63503,9 +63503,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63526,9 +63526,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63549,9 +63549,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63572,9 +63572,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63595,9 +63595,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -63618,9 +63618,9 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"detail": "No static compiler lowering is registered for this API yet.",
"verification": "registry-gap"
"status": "partial",
"detail": "Implemented for the documented static async-iteration subset.",
"verification": "test-backed"
}
},
{
@@ -8,8 +8,8 @@
"tiers": {
"static": {
"replace-refusal": 45,
"verify-gap": 2362,
"audit-partial": 978,
"verify-gap": 2347,
"audit-partial": 993,
"classify": 2,
"implement": 85
}
@@ -61820,12 +61820,15 @@
"priority": "high",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.stream.compose",
"tests": [
"tests/corpus/5342-stream-compose.ts",
"tests/corpus/5343-stream-compose-edges.ts"
]
}
}
},
@@ -61945,12 +61948,15 @@
"priority": "high",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.stream.Readable.from",
"tests": [
"tests/corpus/5340-readable-from-options.ts",
"tests/corpus/5341-readable-from-dynamic.cjs"
]
}
}
},
@@ -62409,12 +62415,15 @@
"priority": "high",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.stream.compose",
"tests": [
"tests/corpus/5342-stream-compose.ts",
"tests/corpus/5343-stream-compose-edges.ts"
]
}
}
},
@@ -62534,12 +62543,15 @@
"priority": "high",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.stream.Readable.from",
"tests": [
"tests/corpus/5340-readable-from-options.ts",
"tests/corpus/5341-readable-from-dynamic.cjs"
]
}
}
},
@@ -63990,12 +64002,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.map",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64015,12 +64032,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.filter",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64040,12 +64062,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.forEach",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64065,12 +64092,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.toArray",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64090,12 +64122,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.some",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64115,12 +64152,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.find",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64140,12 +64182,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.every",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64165,12 +64212,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.flatMap",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64190,12 +64242,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.drop",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64215,12 +64272,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.take",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -64240,12 +64302,17 @@
"priority": "low",
"tiers": {
"static": {
"status": "not-implemented",
"action": "verify-gap",
"verification": "registry-gap",
"confidence": "low",
"source": "compiler-unmatched:stream",
"tests": []
"status": "partial",
"action": "audit-partial",
"verification": "test-backed",
"confidence": "high",
"source": "compiler-feature:stream.readable.reduce",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
}
}
},
@@ -716,9 +716,9 @@
"apiEntries": 140,
"static": {
"supported": 0,
"partial": 38,
"partial": 53,
"refused": 0,
"not-implemented": 102,
"not-implemented": 87,
"by-design": 0,
"unreviewed": 0,
"not-applicable": 0
@@ -85802,7 +85802,15 @@
"tests/corpus/5300-stream-add-abort-signal.ts",
"tests/corpus/5301-stream-add-abort-signal-errors.cjs",
"tests/corpus/5302-stream-promises-options.ts",
"tests/corpus/5303-stream-finished-callback-opts.cjs"
"tests/corpus/5303-stream-finished-callback-opts.cjs",
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs",
"tests/corpus/5340-readable-from-options.ts",
"tests/corpus/5341-readable-from-dynamic.cjs",
"tests/corpus/5342-stream-compose.ts",
"tests/corpus/5343-stream-compose-edges.ts"
]
},
"anchorSource": "chapter"
@@ -86120,8 +86128,12 @@
"inherited": true
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.stream.compose",
"tests": [
"tests/corpus/5342-stream-compose.ts",
"tests/corpus/5343-stream-compose-edges.ts"
]
},
"anchorSource": "exact"
},
@@ -86235,8 +86247,12 @@
"inherited": true
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.stream.Readable.from",
"tests": [
"tests/corpus/5340-readable-from-options.ts",
"tests/corpus/5341-readable-from-dynamic.cjs"
]
},
"anchorSource": "exact"
},
@@ -86695,8 +86711,12 @@
"inherited": true
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.stream.compose",
"tests": [
"tests/corpus/5342-stream-compose.ts",
"tests/corpus/5343-stream-compose-edges.ts"
]
},
"anchorSource": "ancestor"
},
@@ -86810,8 +86830,12 @@
"inherited": true
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.stream.Readable.from",
"tests": [
"tests/corpus/5340-readable-from-options.ts",
"tests/corpus/5341-readable-from-dynamic.cjs"
]
},
"anchorSource": "ancestor"
},
@@ -88294,8 +88318,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.map",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88317,8 +88347,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.filter",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88340,8 +88376,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.forEach",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88363,8 +88405,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.toArray",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88386,8 +88434,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.some",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88409,8 +88463,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.find",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88432,8 +88492,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.every",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88455,8 +88521,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.flatMap",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88478,8 +88550,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.drop",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88501,8 +88579,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.take",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -88524,8 +88608,14 @@
"inherited": false
},
"static": {
"status": "not-implemented",
"evidence": "compiler-unmatched:stream"
"status": "partial",
"evidence": "compiler-feature:stream.readable.reduce",
"tests": [
"tests/corpus/5335-stream-operators-values.ts",
"tests/corpus/5336-stream-operators-promise-timing.ts",
"tests/corpus/5337-stream-operators-concurrency.ts",
"tests/corpus/5338-stream-operators-checks.cjs"
]
},
"anchorSource": "exact"
},
@@ -186,6 +186,9 @@
{"chapter": "readline", "symbols": ["readlinePromises.Interface", "readlinePromises.createInterface", "rl.question"], "status": "partial", "evidence": ["tests/corpus/5295-readline-promises-question.ts", "tests/corpus/5296-readline-lines.ts", "tests/corpus/5298-readline-abort.cjs", "tests/corpus/5299-readline-stdin.ts"]},
{"chapter": "readline", "symbols": ["readline.Interface", "rl.pause", "rl.resume", "rl.prompt", "rl.setPrompt", "rl.getPrompt", "rl.write", "rl", "'close'", "'error'", "'line'", "'pause'", "'resume'"], "status": "partial", "evidence": ["tests/corpus/5295-readline-promises-question.ts", "tests/corpus/5296-readline-lines.ts", "tests/corpus/5297-readline-callback-events.cjs", "tests/corpus/5299-readline-stdin.ts"]},
{"chapter": "stream", "symbols": ["stream.addAbortSignal"], "status": "partial", "evidence": ["tests/corpus/5300-stream-add-abort-signal.ts", "tests/corpus/5301-stream-add-abort-signal-errors.cjs"]},
{"chapter": "stream", "symbols": ["readable.map", "readable.filter", "readable.forEach", "readable.toArray", "readable.some", "readable.find", "readable.every", "readable.flatMap", "readable.drop", "readable.take", "readable.reduce"], "status": "partial", "evidence": ["tests/corpus/5335-stream-operators-values.ts", "tests/corpus/5336-stream-operators-promise-timing.ts", "tests/corpus/5337-stream-operators-concurrency.ts", "tests/corpus/5338-stream-operators-checks.cjs"]},
{"chapter": "stream", "symbols": ["stream.Readable.from"], "status": "partial", "evidence": ["tests/corpus/5340-readable-from-options.ts", "tests/corpus/5341-readable-from-dynamic.cjs"]},
{"chapter": "stream", "symbols": ["stream.compose"], "status": "partial", "evidence": ["tests/corpus/5342-stream-compose.ts", "tests/corpus/5343-stream-compose-edges.ts"]},
{"chapter": "worker_threads", "symbols": ["Worker"], "status": "partial", "evidence": ["tests/corpus/native-worker-messages.ts", "tests/corpus/native-worker-commonjs.cjs", "tests/corpus/native-worker-environment.ts", "tests/corpus/native-worker-roots/main.ts"]},
{"chapter": "worker_threads", "symbols": ["MessagePort", "parentPort", "worker_threads.workerData", "isMainThread", "threadId"], "status": "partial", "evidence": ["tests/corpus/native-worker-commonjs.cjs", "tests/corpus/native-worker-port-lifetime.ts", "tests/corpus/native-worker-references.ts"]},
{"chapter": "worker_threads", "symbols": ["worker.postMessage", "port.postMessage"], "status": "partial", "evidence": ["tests/corpus/native-worker-message-graphs.ts", "tests/corpus/native-worker-shared-views.ts", "tests/corpus/native-worker-clone-errors.ts", "tests/corpus/native-worker-delayed-messages.ts"]},
+57
View File
@@ -5527,6 +5527,14 @@ declare module "stream" {
signal?: AbortSignal;
}
/* The operators' options: concurrency and highWaterMark bound map's
* in-flight callbacks and queued results. */
interface ArrayOptions {
concurrency?: number;
highWaterMark?: number;
signal?: AbortSignal;
}
class Readable extends EventEmitter {
constructor(opts?: ReadableOptions);
/* The underscore-method surface (@types/node declares these on the
@@ -5557,6 +5565,51 @@ declare module "stream" {
readonly destroyed: boolean;
readonly errored: Error | null;
static from(iterable: unknown, options?: ReadableOptions): Readable;
/* The Readable operators: chunks are strings or Buffers, map and
* flatMap callbacks return strings or Buffers (flatMap: arrays of
* them, or strings), and callbacks that declare the options argument
* are refused. */
map(
fn: (data: any, options?: Pick<ArrayOptions, "signal">) => any,
options?: ArrayOptions,
): Readable;
filter(
fn: (data: any, options?: Pick<ArrayOptions, "signal">) => boolean | Promise<boolean>,
options?: ArrayOptions,
): Readable;
forEach(
fn: (data: any, options?: Pick<ArrayOptions, "signal">) => void | Promise<void>,
options?: ArrayOptions,
): Promise<void>;
toArray(options?: Pick<ArrayOptions, "signal">): Promise<any[]>;
some(
fn: (data: any, options?: Pick<ArrayOptions, "signal">) => boolean | Promise<boolean>,
options?: ArrayOptions,
): Promise<boolean>;
find(
fn: (data: any, options?: Pick<ArrayOptions, "signal">) => boolean | Promise<boolean>,
options?: ArrayOptions,
): Promise<any>;
every(
fn: (data: any, options?: Pick<ArrayOptions, "signal">) => boolean | Promise<boolean>,
options?: ArrayOptions,
): Promise<boolean>;
flatMap(
fn: (data: any, options?: Pick<ArrayOptions, "signal">) => any,
options?: ArrayOptions,
): Readable;
drop(limit: number, options?: Pick<ArrayOptions, "signal">): Readable;
take(limit: number, options?: Pick<ArrayOptions, "signal">): Readable;
reduce<T = any>(
fn: (previous: any, data: any, options?: Pick<ArrayOptions, "signal">) => T,
initial?: undefined,
options?: Pick<ArrayOptions, "signal">,
): Promise<T>;
reduce<T = any>(
fn: (previous: T, data: any, options?: Pick<ArrayOptions, "signal">) => T,
initial: T,
options?: Pick<ArrayOptions, "signal">,
): Promise<T>;
on(event: "data", listener: (chunk: any) => void): this;
on(event: "end" | "close" | "readable" | "pause" | "resume", listener: () => void): this;
on(event: "error", listener: (err: Error) => void): this;
@@ -5773,6 +5826,9 @@ declare module "stream" {
function addAbortSignal<T extends Readable | Writable>(signal: AbortSignal, stream: T): T;
/* Declared so requires typecheck; each fences at its use site. */
function pipeline(...streams: unknown[]): unknown;
/* compose(...streams): a Duplex that writes to the first stream and
* reads from the last, with the streams running as a pipeline. */
function compose(...streams: Array<Readable | Writable>): Duplex;
/* finished's options: readable/writable choose which ends count,
* error: false ignores 'error' events, signal aborts the wait. */
interface FinishedOptions {
@@ -5808,6 +5864,7 @@ declare module "stream" {
Stream,
addAbortSignal,
pipeline,
compose,
finished,
getDefaultHighWaterMark,
promises,
+119 -4
View File
@@ -2940,9 +2940,13 @@ export class LlEmitter {
// Number-key Map/Set lookups probe the runtime's direct index inline:
// while ndense is nonzero every live key is an integer below it, so an
// in-range integral key reads its slot (entry index + 1, 0 = absent) and
// any other key is absent. Maps without the index (ndense 0) and every
// non-integral probe outside the table take the runtime lookup, which
// keeps the complete SameValueZero semantics.
// any other key is absent. Maps without the index (ndense 0) hash the
// key in sc_map_entry_f64_hashed, which scans the entries of a small map
// or probes a compact bucket table as scr_map_find_f64 does, and calls
// the runtime lookup for a wide table. It stays out of line so that the
// direct-index path stays small at every call site. Both only read
// memory, so LLVM merges lookups of one key when nothing is stored in
// between (a has() followed by get(), for example).
if (this.decls.has(NUMBER_MAP_ENTRY_DECL)) {
const sz = this.sizeType;
let mapTmp = 0;
@@ -2978,7 +2982,118 @@ export class LlEmitter {
` %dense.any = icmp ne ${sz} %nd, 0`,
` br i1 %dense.any, label %miss, label %slow`,
`slow:`,
` %r = call ptr @scr_map_entry_f64(ptr %m, double %k)`,
` %hashed = call ptr @sc_map_entry_f64_hashed(ptr %m, double %k)`,
` ret ptr %hashed`,
`}`,
``,
// Hashing (scr_map_find_f64): the SameValueZero key word, its
// scr_key_mix hash (zero remapped to one) and the absent-key filter.
`define internal ptr @sc_map_entry_f64_hashed(ptr %m, double %k) noinline ${FN_ATTRS} {`,
`entry:`,
` %nan = fcmp uno double %k, %k`,
` %zero = fcmp oeq double %k, 0.0`,
` %kz = select i1 %zero, double 0.0, double %k`,
` %kzb = bitcast double %kz to i64`,
` %kb = select i1 %nan, i64 9221120237041090560, i64 %kzb`,
` %x1s = lshr i64 %kb, 30`,
` %x1 = xor i64 %kb, %x1s`,
` %x2 = mul i64 %x1, -4658895280553007687`,
` %x3s = lshr i64 %x2, 27`,
` %x3 = xor i64 %x2, %x3s`,
` %x4 = mul i64 %x3, -7723592293110705685`,
` %x5s = lshr i64 %x4, 31`,
` %x5 = xor i64 %x4, %x5s`,
` %hz = icmp eq i64 %x5, 0`,
` %h = select i1 %hz, i64 1, i64 %x5`,
` %f1s = lshr i64 %h, 44`,
` %f1a = and i64 %f1s, 63`,
` %f1 = shl i64 1, %f1a`,
` %f2s = lshr i64 %h, 50`,
` %f2a = and i64 %f2s, 63`,
` %f2 = shl i64 1, %f2a`,
` %fb = or i64 %f1, %f2`,
` %fp = getelementptr inbounds %ScrMapIx, ptr %m, i32 0, i32 11`,
` %filter = load i64, ptr %fp`,
` %fm = and i64 %filter, %fb`,
` %fin = icmp eq i64 %fm, %fb`,
` br i1 %fin, label %table, label %miss`,
`table:`,
` %nbp = getelementptr inbounds %ScrMapIx, ptr %m, i32 0, i32 14`,
` %nb = load ${sz}, ptr %nbp`,
` %eps = getelementptr inbounds %ScrMapIx, ptr %m, i32 0, i32 13`,
` %ents = load ptr, ptr %eps`,
` %linear = icmp eq ${sz} %nb, 0`,
` br i1 %linear, label %scan, label %buckets`,
// Small maps keep no bucket table: scan the cached entry hashes.
`scan:`,
` %nep = getelementptr inbounds %ScrMapIx, ptr %m, i32 0, i32 9`,
` %ne = load ${sz}, ptr %nep`,
` br label %scan.loop`,
`scan.loop:`,
` %si = phi ${sz} [ 0, %scan ], [ %si.next, %scan.next ]`,
` %sdone = icmp eq ${sz} %si, %ne`,
` br i1 %sdone, label %miss, label %scan.entry`,
`scan.entry:`,
` %se = getelementptr inbounds %ScrMapEntry, ptr %ents, ${sz} %si`,
` %seh = getelementptr inbounds %ScrMapEntry, ptr %se, i32 0, i32 2`,
` %sh = load i64, ptr %seh`,
` %sk = load i64, ptr %se`,
` %shq = icmp eq i64 %sh, %h`,
` %skq = icmp eq i64 %sk, %kb`,
` %sq = and i1 %shq, %skq`,
` br i1 %sq, label %scan.found, label %scan.next`,
`scan.found:`,
` ret ptr %se`,
`scan.next:`,
` %si.next = add ${sz} %si, 1`,
` br label %scan.loop`,
// Compact tables tag each 24-bit entry index (+1) with the hash's
// top byte; wider tables take the runtime lookup.
`buckets:`,
` %wide = icmp ugt ${sz} %nb, 16777216`,
` br i1 %wide, label %runtime, label %probe`,
`probe:`,
` %bp = getelementptr inbounds %ScrMapIx, ptr %m, i32 0, i32 15`,
` %bk = load ptr, ptr %bp`,
` %mask = sub ${sz} %nb, 1`,
` %ht = lshr i64 %h, 32`,
` %ht32 = trunc i64 %ht to i32`,
` %tag = and i32 %ht32, -16777216`,
sz === "i64" ? ` %hi = add i64 %h, 0` : ` %hi = trunc i64 %h to ${sz}`,
` %i0 = and ${sz} %hi, %mask`,
` br label %probe.loop`,
`probe.loop:`,
` %pi = phi ${sz} [ %i0, %probe ], [ %pi.next, %probe.next ]`,
` %bep = getelementptr inbounds i32, ptr %bk, ${sz} %pi`,
` %b = load i32, ptr %bep`,
` %bempty = icmp eq i32 %b, 0`,
` br i1 %bempty, label %miss, label %probe.tag`,
`probe.tag:`,
` %bt = and i32 %b, -16777216`,
` %btq = icmp eq i32 %bt, %tag`,
` br i1 %btq, label %probe.entry, label %probe.next`,
`probe.entry:`,
` %bix = and i32 %b, 16777215`,
` %bix1 = sub i32 %bix, 1`,
sz === "i32" ? ` %bi = add i32 %bix1, 0` : ` %bi = zext i32 %bix1 to ${sz}`,
` %pe = getelementptr inbounds %ScrMapEntry, ptr %ents, ${sz} %bi`,
` %peh = getelementptr inbounds %ScrMapEntry, ptr %pe, i32 0, i32 2`,
` %ph = load i64, ptr %peh`,
` %pk = load i64, ptr %pe`,
` %phq = icmp eq i64 %ph, %h`,
` %pkq = icmp eq i64 %pk, %kb`,
` %pq = and i1 %phq, %pkq`,
` br i1 %pq, label %probe.found, label %probe.next`,
`probe.found:`,
` ret ptr %pe`,
`probe.next:`,
` %pi.inc = add ${sz} %pi, 1`,
` %pi.next = and ${sz} %pi.inc, %mask`,
` br label %probe.loop`,
`miss:`,
` ret ptr null`,
`runtime:`,
` %r = call ptr @scr_map_entry_f64(ptr %m, double %k) memory(read) nounwind willreturn`,
` ret ptr %r`,
`}`,
``,
@@ -941,6 +941,19 @@ export const LIB_FN_SYMS: Record<string, string> = {
"readable.nextChunk": "scr_stream_next_chunk",
"readable.nextChunkDyn": "scr_stream_next_chunk_dyn",
"readable.fromArr": "scr_stream_from_arr",
"readable.fromDyn": "scr_stream_from_dyn",
// The Readable operators: borrowed operands; +1 stream or promise.
"readable.opMap": "scr_stream_op_map",
"readable.opFilter": "scr_stream_op_filter",
"readable.opFlatMap": "scr_stream_op_flat_map",
"readable.opDrop": "scr_stream_op_drop",
"readable.opTake": "scr_stream_op_take",
"readable.opToArray": "scr_stream_op_to_array",
"readable.opForEach": "scr_stream_op_for_each",
"readable.opSome": "scr_stream_op_some",
"readable.opEvery": "scr_stream_op_every",
"readable.opFind": "scr_stream_op_find",
"readable.opReduce": "scr_stream_op_reduce",
"readable.pipe": "scr_stream_pipe",
"writable.cork": "scr_stream_cork",
"writable.uncork": "scr_stream_uncork",
@@ -1323,6 +1336,18 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
"readable.nextChunk",
"readable.nextChunkDyn",
"readable.fromArr",
"readable.fromDyn",
"readable.opMap",
"readable.opFilter",
"readable.opFlatMap",
"readable.opDrop",
"readable.opTake",
"readable.opToArray",
"readable.opForEach",
"readable.opSome",
"readable.opEvery",
"readable.opFind",
"readable.opReduce",
"readable.resume",
"readable.pipe",
"readable.unpipe",
@@ -1347,6 +1372,7 @@ export const USES_TIMERS_LIB_FNS = new Set<string>([
"stream.finishedDyn",
"stream.pipeline",
"stream.pipelineDyn",
"stream.compose",
"sp.finished",
"sp.pipeline",
"sp.pipelineOpts",
@@ -612,6 +612,27 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV
host.emitPendingCheck();
return out;
}
if (e.fn === "stream.compose") {
// compose(count, s1..sn): the stream list rides a stack array.
const countArg = e.args[0]!;
if (countArg.kind !== "numLit")
throw new InternalCompilerError(`llvm emitter bug: ${e.fn} count not a literal`);
const n = countArg.value;
const args = e.args.map((a) => host.emitExpr(a));
const arr = B.slot();
B.entryAllocas.push(`${arr} = alloca [${n} x ptr]`);
for (let i = 0; i < n; i++) {
const p = B.tmp();
B.line(`${p} = getelementptr inbounds [${n} x ptr], ptr ${arr}, i64 0, i64 ${i}`);
B.line(`store ptr ${args[1 + i]!.name}, ptr ${p}`);
}
const t = B.tmp();
host.declare(`declare ptr @scr_stream_compose(double, ptr)`);
B.line(`${t} = call ptr @scr_stream_compose(double ${f64Lit(n)}, ptr ${arr})`);
const out = host.own({ name: t, type: e.type });
host.emitPendingCheck();
return out;
}
if (e.fn === "stream.pipeline" || e.fn === "stream.pipelineDyn") {
// pipeline(count, s1..sn, cb): the destination answers +1. The
// stream list rides a stack array (the C compound literal).
@@ -117,3 +117,68 @@ test("string keys and reference values are retained inline, boolean slots compar
expect(text).toContain("@scr_str_retain");
expect(text).toContain("call ptr @scr_map_iter_val_ref(");
});
/** work(map, k): map.has(k) on a number-keyed map. */
function lookupFixture(): IrModule {
const map: IrType = { kind: "map", key: F64, value: F64 };
const params = [
{ localId: "map", name: "map", type: map },
{ localId: "k", name: "k", type: F64 },
];
const has: IrExpr = {
kind: "mapIntrinsic",
method: "has",
receiver: ref("map", map),
args: [ref("k", F64)],
type: BOOL,
loc,
};
const work: IrFunction = {
name: "work",
params,
returnType: BOOL,
loc,
locals: params.map((p) => ({ id: p.localId, name: p.name, type: p.type, mutable: true })),
body: [{ kind: "return", value: has, loc }],
};
return {
irVersion: 16,
sourceFile: loc.file,
entry: "main",
records: [],
unions: [],
functions: [{ name: "main", params: [], locals: [], returnType: VOID, body: [], loc }, work],
};
}
test("hashed number-key lookups probe inline and call the runtime only for wide tables", () => {
for (const bits of [32, 64] as const) {
const sz = bits === 64 ? "i64" : "i32";
const { work, module } = emitted(lookupFixture(), bits);
expect(work).toContain("call ptr @sc_map_entry_f64(");
const helper = /^define internal ptr @sc_map_entry_f64\(ptr %m, double %k\)[^]*?^}/m.exec(
module,
)![0];
expect(helper).toContain("call ptr @sc_map_entry_f64_hashed(ptr %m, double %k)");
const hashed =
/^define internal ptr @sc_map_entry_f64_hashed\(ptr %m, double %k\) noinline [^]*?^}/m.exec(
module,
)![0];
// The SameValueZero key word (canonical NaN, +0 for -0) and the
// scr_key_mix multipliers 0xbf58476d1ce4e5b9 and 0x94d049bb133111eb.
expect(hashed).toContain("select i1 %nan, i64 9221120237041090560, i64 %kzb");
expect(hashed).toContain("mul i64 %x1, -4658895280553007687");
expect(hashed).toContain("mul i64 %x3, -7723592293110705685");
// Filter (field 11), small-map scan of the entry count (field 9), and
// compact buckets (fields 14 and 15) tagged with the hash's top byte.
expect(hashed).toContain("getelementptr inbounds %ScrMapIx, ptr %m, i32 0, i32 11");
expect(hashed).toContain(`%ne = load ${sz}, ptr %nep`);
expect(hashed).toContain(`%nb = load ${sz}, ptr %nbp`);
expect(hashed).toContain("getelementptr inbounds %ScrMapIx, ptr %m, i32 0, i32 15");
expect(hashed).toContain("and i32 %ht32, -16777216");
expect(hashed).toContain(`icmp ugt ${sz} %nb, 16777216`);
expect(hashed).toContain(
"call ptr @scr_map_entry_f64(ptr %m, double %k) memory(read) nounwind willreturn",
);
}
});
@@ -3,8 +3,10 @@
* Number-keyed collections whose keys are small non-negative integers keep a
* direct index in the runtime (scr_map.c). Reads go through the internal
* sc_map_entry_f64 helper (emitter.ts helperDefs), which LLVM inlines: it
* resolves directly indexed maps without a call and defers to
* scr_map_entry_f64 otherwise. The generated code then reads the entry's
* resolves directly indexed maps without a call and other maps through
* the internal sc_map_entry_f64_hashed probe, which hashes the key as the
* runtime does and defers to scr_map_entry_f64 only for wide bucket
* tables. Both only read memory. The generated code then reads the entry's
* value slot itself and retains a reference value with the value type's own
* retain, which is the adapter the collection was constructed with.
* String-keyed reads use the same entry contract through the internal
@@ -549,6 +549,7 @@ export async function compileCInternal(
// The zlib stream classes reference both the zlib engine and the
// stream unit.
...(opts.zlib && opts.stream ? [rt(join(rtDir, "scr_zlib_stream.c"))] : []),
...(opts.stream && opts.dynAsync ? [rt(join(rtDir, "scr_stream_ops.c"))] : []),
// The readiness-poller backends (scr_platform.h): kqueue on macOS/BSD,
// epoll on Linux, WSAPoll on Windows — each TU is empty off its
// platform, so all three link whenever a poller-using unit does and
@@ -59,13 +59,14 @@ import {
import { classStaticData } from "./class-static-data.js";
import { isJsSourceFile } from "../program.js";
import * as ts from "../ts7/adapter.js";
import { SYMBOL_T } from "../../ir/ir.js";
import { SYMBOL_T, type IrLibFn } from "../../ir/ir.js";
import { isClassCallback } from "./class-callbacks.js";
import { CaughtValueDispatch } from "./caught-value-dispatch.js";
import { refreshDescriptorGuards } from "./class-descriptors.js";
import { ClassConstructionDispatch } from "./class-construction.js";
import { emitterRooted } from "./lower-event-emitter.js";
import { protocolArm, protocolFieldArm } from "./class-protocol-dispatch.js";
import { STREAM_OPERATORS } from "./lower-stream.js";
const STREAM_METHODS = new Set([
"on",
@@ -1089,7 +1090,12 @@ export class ClassDynamicDispatch {
const matching = this.propertyReceivers(lowerer, true).flatMap(({ info }) => {
if (!info || info.fields.has(method) || info.builtinError) return [];
if (emitterRooted(lowerer, info) && EMITTER_METHODS.has(method)) return [info];
if (info.builtinStream) return STREAM_METHODS.has(method) ? [info] : [];
if (info.builtinStream) {
return STREAM_METHODS.has(method) ||
(STREAM_OPERATORS.has(method) && info.builtinStream !== "w")
? [info]
: [];
}
return findMethodOn(lowerer, info, method) || findGenericMethodOn(lowerer, info, method)
? [info]
: [];
@@ -3969,6 +3975,48 @@ export class ClassDynamicDispatch {
type: method === "isPaused" ? BOOL : receiver.type,
loc,
});
if (STREAM_OPERATORS.has(method) && info.builtinStream && info.builtinStream !== "w") {
// The Readable operators over a checked-dynamic receiver: every
// argument is already a checked value.
const arg = (i: number): IrExpr => incoming[i] ?? dynUndefinedExpr(loc);
const readableT: IrType = { kind: "object", className: "%Readable" };
const op = (fn: string, args: IrExpr[], type: IrType): IrStmt[] =>
ret({ kind: "libCall", fn: fn as IrLibFn, args: [receiver, ...args], type, loc });
switch (method) {
case "map":
return op("readable.opMap", [arg(0), arg(1)], readableT);
case "filter":
return op("readable.opFilter", [arg(0), arg(1)], readableT);
case "flatMap":
return op("readable.opFlatMap", [arg(0), arg(1)], readableT);
case "drop":
return op("readable.opDrop", [arg(0), arg(1)], readableT);
case "take":
return op("readable.opTake", [arg(0), arg(1)], readableT);
case "toArray":
return op("readable.opToArray", [arg(0)], { kind: "promise", inner: DYN });
case "forEach":
return op("readable.opForEach", [arg(0), arg(1)], { kind: "promise", inner: VOID });
case "some":
return op("readable.opSome", [arg(0), arg(1)], { kind: "promise", inner: BOOL });
case "every":
return op("readable.opEvery", [arg(0), arg(1)], { kind: "promise", inner: BOOL });
case "find":
return op("readable.opFind", [arg(0), arg(1)], { kind: "promise", inner: DYN });
case "reduce":
return op(
"readable.opReduce",
[
arg(0),
arg(1),
{ kind: "boolLit", value: dispatch.source.args.length >= 2, type: BOOL, loc },
arg(2),
{ kind: "numLit", value: 0, type: F64, loc },
],
{ kind: "promise", inner: DYN },
);
}
}
if (method === "destroy" && (!incoming.length || incoming[0]?.kind === "unitLit"))
return ret({
kind: "libCall",
@@ -317,6 +317,19 @@ function boolRef(localId: string, loc: ReturnType<typeof locOf>): IrExpr {
return { kind: "varRef", localId, type: BOOL, loc };
}
/** The loop's stream: a JavaScript value typed as a stream can arrive as a
* checked-dynamic value (returned through an untyped function), which
* checks back into the stream class. */
function streamSource(
lowerer: Lowerer,
plan: NodeReadablePlan,
loc: ReturnType<typeof locOf>,
): IrExpr {
const v = lowerer.lowerExpr(plan.source);
if (v.type.kind !== "dyn") return v;
return { kind: "dynCheck", value: v, type: plan.type, loc };
}
function lowerNodeReadable(
lowerer: Lowerer,
stmt: ts.ForOfStatement,
@@ -423,7 +436,7 @@ function lowerNodeReadable(
return {
kind: "block",
body: [
{ kind: "varDecl", localId: stream.id, init: lowerer.lowerExpr(plan.source), loc },
{ kind: "varDecl", localId: stream.id, init: streamSource(lowerer, plan, loc), loc },
{
kind: "varDecl",
localId: active.id,
@@ -150,6 +150,7 @@ import {
} from "./lower-console.js";
import {
STREAM_API_MEMBERS,
STREAM_OPERATORS,
lowerStreamMethodCall,
lowerStreamStaticCall,
streamSidesOf,
@@ -8592,7 +8593,7 @@ export function lowerObjectMethodCall(
if (
info &&
!found &&
STREAM_API_MEMBERS.has(access.name.text) &&
(STREAM_API_MEMBERS.has(access.name.text) || STREAM_OPERATORS.has(access.name.text)) &&
streamSidesOf(lowerer, info) !== null
) {
const stream = lowerStreamMethodCall(lowerer, call, access, info);
@@ -38,6 +38,8 @@ import { appendImplicitUndefinedReturn } from "./function-returns.js";
import { bufEncoding, knownBufEncoding } from "./containers/bytes.js";
import { tryLowerExpression } from "./expressions/try-lower-expression.js";
import { sideEffectFreeOptionValue } from "./surfaces.js";
import { lowerDynObjectLiteral } from "./expressions/object-literals.js";
import { dynUndefinedExpr } from "../../ir/build.js";
import {
BOOL,
DYN,
@@ -1346,12 +1348,14 @@ export function lowerStreamUnderscoreAssign(
/* ── the static surface (Readable.from) ───────────────────────────────── */
/** `Readable.from(iterable)` — the array-seeded object-entry form:
* string[] / Buffer[] values (or literals) become a fully-seeded
* readable delivering one WHOLE chunk per element (Node's objectMode
* accounting; hwm 1), and a single string/Buffer argument is a
* one-chunk stream (Node's special case). Called from the method-call
* chain once the receiver resolved to a stream class value. */
/** `Readable.from(iterable, options)`: string[] / Buffer[] sources
* become a readable that pushes one element per _read (whole entries in
* object mode, hwm 1), and a single string/Buffer argument is a
* one-chunk stream (from.js's special case, the mode's default hwm). The
* options lower when they are a literal of `objectMode` and
* `highWaterMark`; the JS lane and `any` sources take the runtime's
* checked-dynamic form. Called from the method-call chain once the
* receiver resolved to a stream class value. */
export function lowerStreamStaticCall(
lowerer: Lowerer,
call: ts.CallExpression,
@@ -1410,49 +1414,97 @@ export function lowerStreamStaticCall(
lowerer.noLowering(`${cls.slice(1)}.from`, call, "Readable.from is the lowered form");
}
const args = call.arguments;
if (args.length !== 1) {
if (args.length > 2) {
lowerer.noLowering(
`Readable.from with ${args.length} arguments`,
call,
"the one-argument form (an array, string, or Buffer) is supported",
"Readable.from(iterable, options) is the supported form",
);
}
const srcT = lowerer.mapTypeOf(lowerer.typeOf(args[0]!));
const type: IrType = { kind: "object", className: "%Readable" };
if (
srcT?.kind === "array" &&
(srcT.elem.kind === "string" || (srcT.elem.kind === "bytes" && srcT.elem.elem === "u8"))
) {
const arr = lowerer.lowerExpr(args[0]!);
const strings = srcT.elem.kind === "string";
const js = isJsSourceFile(call.getSourceFile());
const srcNode = args[0];
const optsNode = args[1];
const srcT = srcNode ? lowerer.mapTypeOf(lowerer.typeOf(srcNode)) : null;
const typedElem = (t: IrType | null | undefined): "string" | "bytes" | null =>
t?.kind === "string" ? "string" : t?.kind === "bytes" && t.elem === "u8" ? "bytes" : null;
const single = typedElem(srcT);
const elem = srcT?.kind === "array" ? typedElem(srcT.elem) : null;
const opts = optsNode ? fromOptions(lowerer, optsNode) : { flags: 0, hwm: null };
if (srcNode && (single || elem) && opts) {
const v = lowerer.lowerExpr(srcNode);
const hwm = opts.hwm ? lowerer.lowerExprExpecting(opts.hwm, F64) : numLit(0, loc);
const arr: IrExpr = single ? { kind: "arrayLit", elems: [v], type: arrayOf(v.type), loc } : v;
// SCR_FROM_STRINGS 1, SCR_FROM_SINGLE 2 (scr_runtime.h); the options
// add SCR_FROM_BYTE_MODE 4 and SCR_FROM_HAS_HWM 8.
const flags = ((single ?? elem) === "string" ? 1 : 0) | (single ? 2 : 0) | opts.flags;
return {
kind: "libCall",
fn: "readable.fromArr",
args: [arr, boolLit(strings, loc)],
args: [arr, numLit(flags, loc), hwm],
type,
loc,
};
}
if (srcT?.kind === "string" || (srcT?.kind === "bytes" && srcT.elem === "u8")) {
// Node special-cases a single string/Buffer: ONE whole chunk.
const v = lowerer.lowerExpr(args[0]!);
const strings = srcT.kind === "string";
const arr: IrExpr = { kind: "arrayLit", elems: [v], type: arrayOf(v.type), loc };
return {
kind: "libCall",
fn: "readable.fromArr",
args: [arr, boolLit(strings, loc)],
type,
loc,
};
const anySrc =
srcT?.kind === "dyn" ||
(srcNode !== undefined &&
srcT === null &&
(lowerer.typeOf(srcNode).flags & ts.TypeFlags.Any) !== 0);
if (js || anySrc) {
// The checked-dynamic form: the runtime takes from.js's ladder
// (ERR_INVALID_ARG_TYPE for a non-iterable, the options, null values).
const src = srcNode ? lowerer.lowerExprExpecting(srcNode, DYN) : dynUndefinedExpr(loc);
const o = lowerOperatorValue(lowerer, optsNode, "Readable.from options", loc, true);
return { kind: "libCall", fn: "readable.fromDyn", args: [src, o], type, loc };
}
if (srcNode && (single || elem)) {
lowerer.noLowering(
"Readable.from with these options",
optsNode!,
"an object literal with 'objectMode' (true or false) and a number 'highWaterMark' is supported",
);
}
lowerer.noLowering(
`Readable.from over a '${srcT ? lowerer.fmt(srcT) : "?"}' source`,
args[0]!,
srcNode ?? call,
"string[] / Buffer[] arrays (and a single string or Buffer) are the supported sources — generators and async iterables have no lowering yet",
);
}
/** Readable.from's typed options: an object literal of `objectMode`
* (a boolean literal) and `highWaterMark` (a number), as SCR_FROM_* flag
* bits and the hwm expression; null for any other shape. */
function fromOptions(
lowerer: Lowerer,
node: ts.Expression,
): { flags: number; hwm: ts.Expression | null } | null {
if (!ts.isObjectLiteralExpression(node)) return null;
let flags = 0;
let hwm: ts.Expression | null = null;
for (const prop of node.properties) {
if (
!ts.isPropertyAssignment(prop) ||
!(ts.isIdentifier(prop.name) || ts.isStringLiteral(prop.name))
) {
return null;
}
const value = prop.initializer;
if (prop.name.text === "objectMode") {
if (value.kind === ts.SyntaxKind.FalseKeyword) flags |= 4;
else if (value.kind === ts.SyntaxKind.TrueKeyword) flags &= ~4;
else return null;
} else if (prop.name.text === "highWaterMark") {
if (lowerer.mapTypeOf(lowerer.typeOf(value))?.kind !== "f64") return null;
flags |= 8;
hwm = value;
} else {
return null;
}
}
return { flags, hwm };
}
/** `const Writable = stream.Writable` — a runtime stream CLASS bound
* through a namespace member: alias plumbing with no storage (both
* declaration walks skip it; uses resolve through builtinStreamInfoOf's
@@ -1491,7 +1543,7 @@ const PIPELINE_STAGE_EXPECTED =
function lowerStreamArg(
lowerer: Lowerer,
node: ts.Expression,
what: "finished" | "pipeline" | "addAbortSignal",
what: "finished" | "pipeline" | "addAbortSignal" | "compose",
): IrExpr {
const v = lowerer.lowerExpr(node);
const info = v.type.kind === "object" ? lowerer.classes.get(v.type.className) : undefined;
@@ -1546,7 +1598,7 @@ function lowerStreamArg(
lowerer.noLowering(
`${what} over a '${lowerer.fmt(v.type)}'`,
node,
what === "pipeline"
what === "pipeline" || what === "compose"
? "stream stages are the lowered surface — Node also accepts iterables (strings, arrays), generators, plain functions, and { readable, writable } pairs, but those have no lowering yet"
: "stream arguments are the lowered surface (iterables, generators, and plain functions have no lowering yet)",
);
@@ -2005,11 +2057,295 @@ export function lowerStreamModuleCall(
};
}
if (bi.member === "compose") return lowerStreamCompose(lowerer, call, loc);
return null;
}
/** `stream.compose(...streams)` over stream instances: a Duplex writing
* to the first stream and reading from the last (scr_stream_compose). The
* classes fix which positions can be readable and writable; a stream in
* the wrong position, a single Readable or Writable (Duplex.from wraps
* those), and functions, iterables or web streams have no lowering. */
function lowerStreamCompose(lowerer: Lowerer, call: ts.CallExpression, loc: SrcLoc): IrExpr {
const args = call.arguments;
const duplex: IrType = { kind: "object", className: "%Duplex" };
if (args.some((a) => ts.isSpreadElement(a))) {
lowerer.noLowering("compose with spread arguments", call, "pass the streams directly");
}
if (args.length === 0) {
return nodeThrowExpr(
1,
"ERR_MISSING_ARGS",
'The "streams" argument must be specified',
duplex,
loc,
);
}
let streams: IrExpr[];
try {
streams = args.map((a) => lowerStreamArg(lowerer, a, "compose"));
} catch (e) {
if (e instanceof StreamArgTypeThrow) return e.expr;
throw e;
}
const sides = streams.map((s) =>
s.type.kind === "object" ? streamSidesOf(lowerer, lowerer.classes.get(s.type.className)) : null,
);
if (streams.length === 1) {
// Duplex.from answers a duplex itself.
if (sides[0] === "rw") return streams[0]!;
lowerer.noLowering(
`compose of a single ${sides[0] === "r" ? "Readable" : "Writable"}`,
args[0]!,
"Duplex.from wraps a lone Readable or Writable, which has no lowering yet; one Duplex, Transform or PassThrough answers itself",
);
}
for (let i = 0; i < streams.length; i++) {
const s = sides[i];
const want =
i < streams.length - 1 && s === "w" ? "readable" : i > 0 && s === "r" ? "writable" : null;
if (want) {
lowerer.noLowering(
`compose with a stream that is not ${want} at position ${i}`,
args[i]!,
`Node throws ERR_INVALID_ARG_VALUE ('streams[${i}]' must be ${want}); every stream but the last must be readable and every stream but the first writable`,
);
}
}
return {
kind: "libCall",
fn: "stream.compose",
args: [numLit(streams.length, loc), ...streams],
type: duplex,
loc,
};
}
/* ── the method surface ───────────────────────────────────────────────── */
/* ── the Readable operators ──────────────────────────────────────────── */
/** The Readable operators (scr_stream_ops.c — Node's operators.js). */
export const STREAM_OPERATORS: ReadonlySet<string> = new Set([
"map",
"filter",
"flatMap",
"drop",
"take",
"toArray",
"forEach",
"some",
"every",
"find",
"reduce",
]);
/** An operator callback as a checked-dynamic function value. Callbacks
* that declare the options argument ({ signal }) are refused: the
* AbortSignal Node passes there has no representation. map and flatMap
* callbacks must produce strings or Buffers (flatMap: arrays of them, or
* strings) — or checked-dynamic values, which the runtime checks. */
function lowerOperatorCallback(
lowerer: Lowerer,
node: ts.Expression,
member: string,
maxParams: number,
): IrExpr {
const loc = locOf(node);
const optionsParam = (at: ts.Node): never =>
lowerer.noLowering(
`${member} callbacks that declare the options argument`,
at,
"the { signal } argument has no lowering — take only the chunk",
);
if (
(ts.isArrowFunction(node) || ts.isFunctionExpression(node)) &&
node.parameters.length > maxParams
) {
optionsParam(node.parameters[maxParams]!);
}
const v = lowerer.lowerExpr(node);
if (v.type.kind === "dyn") return v;
if (
v.type.kind !== "func" &&
isJsSourceFile(node.getSourceFile()) &&
lowerer.dynConvertible(v.type)
) {
// JavaScript passes any value; Node's validateFunction throws.
return { kind: "dynFrom", value: v, type: DYN, loc };
}
if (v.type.kind !== "func") {
lowerer.noLowering(
`${member} with a '${lowerer.fmt(v.type)}' callback`,
node,
"pass a function",
);
}
if (v.type.params.length > maxParams) optionsParam(node);
if (member === "map" || member === "flatMap") {
const ret = v.type.ret.kind === "promise" ? v.type.ret.inner : v.type.ret;
const chunk = (t: IrType): boolean =>
t.kind === "string" || t.kind === "dyn" || (t.kind === "bytes" && t.elem === "u8");
const ok =
member === "map"
? chunk(ret)
: ret.kind === "dyn" || ret.kind === "string" || (ret.kind === "array" && chunk(ret.elem));
if (!ok) {
lowerer.noLowering(
`${member} callbacks that return '${lowerer.fmt(ret)}'`,
node,
member === "map"
? "map results are stream chunks: strings or Buffers (object-mode streams of other values are not lowered)"
: "flatMap results are arrays of strings or Buffers, or strings",
);
}
}
const getRecord = (id: string) => lowerer.shapes.get(id);
const getUnion = (id: string) => lowerer.unions.get(id);
if (!canBoxFuncIntoDyn(v.type, getRecord, getUnion)) {
lowerer.noLowering(
`${member} with this callback`,
node,
"the callback's parameter and result types must cross the checked-dynamic boundary",
);
}
return { kind: "dynFrom", value: v, type: DYN, loc };
}
/** An operator's options (or reduce's initial value): an object literal
* becomes a checked-dynamic object, checked-dynamic values pass through,
* absent arguments are undefined. */
function lowerOperatorValue(
lowerer: Lowerer,
node: ts.Expression | undefined,
what: string,
loc: SrcLoc,
literalObject: boolean,
): IrExpr {
if (node === undefined) return dynUndefinedExpr(loc);
if (literalObject && ts.isObjectLiteralExpression(node))
return lowerDynObjectLiteral(lowerer, node);
const v = lowerer.lowerExpr(node);
if (v.type.kind === "dyn") return v;
if (literalObject && v.type.kind !== "undefinedT" && v.type.kind !== "nullT") {
lowerer.noLowering(
`${what} that are not an object literal`,
node,
"write the options inline, e.g. { concurrency: 2 }",
);
}
if (!lowerer.dynConvertible(v.type)) {
lowerer.noLowering(`${what} of type '${lowerer.fmt(v.type)}'`, node);
}
return { kind: "dynFrom", value: v, type: DYN, loc: locOf(node) };
}
function lowerStreamOperator(
lowerer: Lowerer,
call: ts.CallExpression,
access: ts.PropertyAccessExpression,
member: string,
): IrExpr {
const loc = locOf(call);
const args = call.arguments;
const recv = lowerer.lowerExpr(access.expression);
const readable: IrType = { kind: "object", className: "%Readable" };
const opts = (i: number): IrExpr =>
lowerOperatorValue(lowerer, args[i], `${member} options`, loc, true);
const fnArg = (maxParams: number): IrExpr => {
if (args[0] === undefined) return dynUndefinedExpr(loc);
return lowerOperatorCallback(lowerer, args[0], member, maxParams);
};
const lib = (fn: string, a: IrExpr[], type: IrType): IrExpr => ({
kind: "libCall",
fn: fn as IrLibFn,
args: a,
type,
loc,
});
switch (member) {
case "map":
case "filter":
case "flatMap": {
const fn = fnArg(1);
const fnName =
member === "map"
? "readable.opMap"
: member === "filter"
? "readable.opFilter"
: "readable.opFlatMap";
return lib(fnName, [recv, fn, opts(1)], readable);
}
case "drop":
case "take": {
const n = lowerOperatorValue(lowerer, args[0], `${member}'s count`, loc, false);
return lib(
member === "drop" ? "readable.opDrop" : "readable.opTake",
[recv, n, opts(1)],
readable,
);
}
case "toArray":
return lib("readable.opToArray", [recv, opts(0)], { kind: "promise", inner: DYN });
case "forEach":
return lib("readable.opForEach", [recv, fnArg(1), opts(1)], { kind: "promise", inner: VOID });
case "some":
case "every": {
const fn = fnArg(1);
return lib(member === "some" ? "readable.opSome" : "readable.opEvery", [recv, fn, opts(1)], {
kind: "promise",
inner: BOOL,
});
}
case "find": {
const t = lowerer.typeOf(call);
const inner = lowerer.checker.getTypeArguments(t as ts.TypeReference)[0];
if (inner && !(inner.flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown))) {
lowerer.noLowering(
"find with a type-guard callback",
call,
"the result is a checked-dynamic value — use a boolean predicate and narrow the result",
);
}
return lib("readable.opFind", [recv, fnArg(1), opts(1)], { kind: "promise", inner: DYN });
}
case "reduce": {
const t = lowerer.typeOf(call);
const resultTs = lowerer.checker.getTypeArguments(t as ts.TypeReference)[0];
let kind = 0;
let inner: IrType = DYN;
if (resultTs && !(resultTs.flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown))) {
const mapped = lowerer.mapTypeOf(resultTs);
if (mapped?.kind === "f64") {
kind = 1;
inner = F64;
} else if (mapped?.kind === "bool") {
kind = 2;
inner = BOOL;
} else if (mapped?.kind === "string") {
kind = 3;
inner = STRING;
} else if (mapped?.kind !== "dyn") {
lowerer.noLowering(
`reduce with a '${mapped ? lowerer.fmt(mapped) : lowerer.checker.typeToString(resultTs)}' result`,
call,
"reduce results are numbers, booleans, strings, or checked-dynamic values",
);
}
}
const fn = fnArg(2);
const init = lowerOperatorValue(lowerer, args[1], "reduce's initial value", loc, false);
return lib(
"readable.opReduce",
[recv, fn, init, boolLit(args.length >= 2, loc), opts(2), numLit(kind, loc)],
{ kind: "promise", inner },
);
}
}
throw new InternalCompilerError(`unknown stream operator ${member}`);
}
/** The one supported encoding argument: utf8 (Node's default), as a
* compile-time literal or absent. */
function checkUtf8Encoding(lowerer: Lowerer, member: string, arg: ts.Expression | undefined): void {
@@ -2032,6 +2368,11 @@ export function lowerStreamMethodCall(
info: ClassInfo,
): IrExpr | null {
const member = access.name.text;
if (STREAM_OPERATORS.has(member)) {
const opSides = streamSidesOf(lowerer, info);
if (opSides === "r" || opSides === "rw")
return lowerStreamOperator(lowerer, call, access, member);
}
if (!STREAM_API_MEMBERS.has(member)) return null;
const sides = streamSidesOf(lowerer, info);
if (!sides) return null;
@@ -1145,8 +1145,21 @@ export function dynFallbackType(lowerer: Lowerer, node: ts.Node, t: ts.Type): Ir
symbol && lowerer.isStdlibSymbol(symbol)
? lowerer.checker.getTypeArguments(t as ts.TypeReference)
: [];
if (symbol?.name === "Promise" && args[0] && args[0].flags & ts.TypeFlags.Any)
return jsFallbackPromiseType(lowerer, t);
// Promise<any> and Promise<any[]> (an any[] is itself a checked
// value below): a promise of a checked value.
const anyArray = (x: ts.Type): boolean =>
lowerer.checker.isArrayType(x) &&
((lowerer.checker.getTypeArguments(x as ts.TypeReference)[0]?.flags ?? 0) &
ts.TypeFlags.Any) !==
0;
if (
symbol?.name === "Promise" &&
args[0] &&
(args[0].flags & ts.TypeFlags.Any || anyArray(args[0]))
)
return symbol.name === "Promise" && anyArray(args[0])
? { kind: "promise", inner: DYN }
: jsFallbackPromiseType(lowerer, t);
if (
symbol &&
["Generator", "AsyncGenerator", "IterableIterator"].includes(symbol.name) &&
@@ -71,6 +71,7 @@ export const JS_RELAXED_TSC_CODES: ReadonlySet<number> = new Set([
2556,
2769,
2740, // 2322's elaborated "missing the following properties" form
2559, // a value with no properties in common with an all-optional options type
2305,
2551,
2724,
@@ -242,6 +242,7 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
"stream.finishedDyn",
"stream.pipeline",
"stream.pipelineDyn",
"stream.compose",
"sp.finished",
"sp.pipeline",
"sc.text",
@@ -384,6 +385,13 @@ export const MAY_THROW_LIB_FNS: ReadonlySet<IrLibFn> = new Set([
"https.requestFn",
"https.requestFnCb",
"rl.question",
"readable.fromArr",
"readable.fromDyn",
"readable.opMap",
"readable.opFilter",
"readable.opFlatMap",
"readable.opDrop",
"readable.opTake",
"rl.createSignal",
"stream.finishedOpts",
"stream.finishedOptsSignal",
+28 -3
View File
@@ -3222,6 +3222,10 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
"stream.finishedDyn": { argTypes: [null, DYN], result: VOID },
"stream.pipeline": { argTypes: [F64], result: VOID },
"stream.pipelineDyn": { argTypes: [F64], result: VOID },
/** stream.compose(count, s1..sn): the Duplex over a pipeline of the
* streams (+1). MAY THROW (ERR_INVALID_ARG_VALUE for a stream that is
* no longer readable or writable). */
"stream.compose": { argTypes: [F64], result: VOID },
/** node:stream/promises — the promise forms over the same machinery:
* sp.finished: (s) → a pending void promise the terminal watcher
* settles; sp.pipeline: (count, s1..sn) → the callback pipeline's
@@ -3304,11 +3308,32 @@ export const LIB_FN_SIGS = defineLibFnSignatures({
* +1 promise of the next chunk — buffered content, the EOF sentinel
* (empty Buffer / dyn undefined), or a rejection with the stream's
* error. The Dyn twin boxes chunks by runtime tag (the JS lane).
* readable.fromArr is Readable.from(array): a fully-seeded object-
* entry stream (one whole chunk per element; strings per the flag). */
* readable.fromArr is Readable.from over a typed array, string or
* Buffer: [array, SCR_FROM_* flags, highWaterMark]. readable.fromDyn is
* the JS lane's [source, options]. */
"readable.nextChunk": { argTypes: [null], result: VOID },
"readable.nextChunkDyn": { argTypes: [null], result: VOID },
"readable.fromArr": { argTypes: [null, BOOL], result: VOID },
"readable.fromArr": { argTypes: [null, F64, F64], result: VOID },
"readable.fromDyn": { argTypes: [DYN, DYN], result: VOID },
/** The Readable operators (scr_stream_ops.c): [stream, ...] with every
* other argument a checked-dynamic value (the callback, the count, the
* options). map/filter/flatMap/drop/take answer the object-mode output
* stream (validated as a stream class; THROW Node's validation errors);
* toArray settles a dyn array, find a dyn value, some/every a boolean,
* forEach nothing; reduce: [stream, fn, init, hasInit, options, kind]
* with the payload kind (0 dyn, 1 number, 2 boolean, 3 string). The
* promise-returning ones reject instead of throwing. */
"readable.opMap": { argTypes: [null, DYN, DYN], result: VOID },
"readable.opFilter": { argTypes: [null, DYN, DYN], result: VOID },
"readable.opFlatMap": { argTypes: [null, DYN, DYN], result: VOID },
"readable.opDrop": { argTypes: [null, DYN, DYN], result: VOID },
"readable.opTake": { argTypes: [null, DYN, DYN], result: VOID },
"readable.opToArray": { argTypes: [null, DYN], result: { kind: "promise", inner: DYN } },
"readable.opForEach": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: VOID } },
"readable.opSome": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: BOOL } },
"readable.opEvery": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: BOOL } },
"readable.opFind": { argTypes: [null, DYN, DYN], result: { kind: "promise", inner: DYN } },
"readable.opReduce": { argTypes: [null, DYN, DYN, BOOL, DYN, F64], result: VOID },
"readable.resume": { argTypes: [null], result: VOID },
"readable.isPaused": { argTypes: [null], result: BOOL },
"readable.pipe": { argTypes: [null, null, BOOL], result: VOID },
@@ -231,6 +231,9 @@ function scanRuntimeFeatures(mod: IrModule, stopAt?: keyof RuntimeFeatures): Run
)
features.stream = true;
if (fn.startsWith("zlib.")) features.zlib = true;
// The Readable operators (scr_stream_ops.c) await through the
// dyn-async unit.
if (fn.startsWith("readable.op")) features.dynAsync = true;
if (fn.startsWith("dc.")) features.dc = true;
if (fn.startsWith("assert.")) features.assert = true;
if (
+47 -4
View File
@@ -3960,6 +3960,7 @@ function validateFunction(
e.fn === "passthrough.initDyn" ||
e.fn === "stream.pipeline" ||
e.fn === "stream.pipelineDyn" ||
e.fn === "stream.compose" ||
e.fn === "sp.pipeline" ||
e.fn === "sp.pipelineOpts" ||
e.fn === "sp.pipelineOptsSignal" ||
@@ -5239,6 +5240,7 @@ function validateFunction(
e.fn === "stream.finishedDyn" ||
e.fn === "stream.pipeline" ||
e.fn === "stream.pipelineDyn" ||
e.fn === "stream.compose" ||
e.fn === "stream.addAbortSignal" ||
e.fn === "stream.addAbortSignalFn" ||
e.fn === "stream.finishedOpts" ||
@@ -5329,6 +5331,22 @@ function validateFunction(
}
return true;
}
if (e.fn === "stream.compose") {
// (count, s1..sn) → a Duplex (n >= 2), or the lone stream.
const count = e.args[0];
const n = count?.kind === "numLit" ? count.value : -1;
if (n < 1 || e.args.length !== n + 1) {
err(`libCall ${e.fn} count/arity mismatch`, e.loc);
return true;
}
for (let i = 1; i <= n; i++) {
if (!isStreamObject(e.args[i]?.type)) {
err(`libCall ${e.fn} stage ${i} must be a stream-hierarchy object`, e.loc);
}
}
if (!isStreamObject(e.type)) err(`libCall ${e.fn} must return a stream`, e.loc);
return true;
}
if (e.fn === "stream.pipeline" || e.fn === "stream.pipelineDyn") {
// (count, s1..sn, cb) → the destination's type.
const count = e.args[0];
@@ -5354,10 +5372,35 @@ function validateFunction(
}
return true;
}
if (e.fn === "readable.fromArr") {
// No receiver: the seed array leads; the result is the class.
if (e.args[0]?.type.kind !== "array" || !isStreamObject(e.type)) {
err(`libCall readable.fromArr must take an array and return a stream class`, e.loc);
if (
e.fn === "readable.opMap" ||
e.fn === "readable.opFilter" ||
e.fn === "readable.opFlatMap" ||
e.fn === "readable.opDrop" ||
e.fn === "readable.opTake"
) {
if (!isStreamObject(e.args[0]?.type) || !isStreamObject(e.type)) {
err(`libCall ${e.fn} must take a stream and return a stream class`, e.loc);
}
return true;
}
if (e.fn === "readable.opReduce") {
const inner = e.type.kind === "promise" ? e.type.inner : null;
const kind = e.args[5]?.kind === "numLit" ? e.args[5].value : -1;
const want = [DYN, F64, BOOL, STRING][kind];
if (!isStreamObject(e.args[0]?.type) || !inner || !want || !typeEquals(inner, want)) {
err(`libCall readable.opReduce must return the promise its kind names`, e.loc);
}
return true;
}
if (e.fn === "readable.fromArr" || e.fn === "readable.fromDyn") {
// No receiver: the source leads; the result is the class.
const src = e.fn === "readable.fromArr" ? "array" : "dyn";
if (e.args[0]?.type.kind !== src || !isStreamObject(e.type)) {
err(`libCall ${e.fn} must take a ${src} source and return a stream class`, e.loc);
}
if (e.fn === "readable.fromArr" && e.args[1]?.kind !== "numLit") {
err(`libCall readable.fromArr flags must be a number literal`, e.loc);
}
return true;
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5335-stream-operators-values.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5336-stream-operators-promise-timing.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5337-stream-operators-concurrency.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5338-stream-operators-checks.cjs"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5339-stream-for-await-hops.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5340-readable-from-options.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5341-readable-from-dynamic.cjs"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5342-stream-compose.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5343-stream-compose-edges.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5344-stream-pipeline-timing.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5710-map-number-keys-hashed.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/5712-map-number-overwrite-at-growth.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-dynamic-entry-kinds.cjs"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-dynamic-entry-kinds.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-encoded-from-entries.cjs"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/stream-encoded-from-entries.ts"
],
"diags": []
}
@@ -90,6 +90,8 @@ const optional = [
["scr_stream.c", "stream"],
// The zlib stream classes reference both the zlib engine and the stream unit.
["scr_zlib_stream.c", all("zlib", "stream")],
// The Readable operators await through the dyn-async unit.
["scr_stream_ops.c", all("stream", "dynAsync")],
["scr_loop_kqueue.c", any("netEffective", "dgram")],
["scr_loop_epoll.c", any("netEffective", "dgram")],
["scr_loop_wsapoll.c", any("netEffective", "dgram")],
@@ -80,6 +80,8 @@ const OPTIONAL = [
["scr_stream.c", "stream"],
// The zlib stream classes reference both the zlib engine and the stream unit.
["scr_zlib_stream.c", all("zlib", "stream")],
// The Readable operators await through the dyn-async unit.
["scr_stream_ops.c", all("stream", "dynAsync")],
["scr_loop_kqueue.c", any("netEffective", "dgram")],
["scr_loop_epoll.c", any("netEffective", "dgram")],
["scr_loop_wsapoll.c", any("netEffective", "dgram")],
+3
View File
@@ -533,6 +533,9 @@ void scr_dyn_arr_push_hole(ScrDyn *arr);
ScrDyn *scr_dyn_new_obj(void);
ScrDyn *scr_dyn_iterator(const ScrDyn *value, const ScrStr *spell);
ScrDyn *scr_dyn_iterator_optional(const ScrDyn *value);
/* from.js's probe: 2 when value[Symbol.asyncIterator] is set, 1 when
* value[Symbol.iterator] is, 0 for neither, -1 with an error pending. */
int scr_dyn_iterable_kind(const ScrDyn *value);
void scr_dyn_install_iterator_symbol(ScrDyn *key);
void scr_dyn_install_async_iterator_symbol(ScrDyn *key);
ScrDyn *scr_dyn_array_values_function(void);
+17
View File
@@ -4152,6 +4152,23 @@ static ScrDyn *scr_iterator_method(const ScrDyn *source) {
return scr_builtin_iterable(source) ? scr_builtin_iterator_method(source) : scr_dyn_retain(scr_dyn_undefined());
}
int scr_dyn_iterable_kind(const ScrDyn *value) {
if (value->kind == SCR_DYN_NULL || value->kind == SCR_DYN_UNDEF) return 0;
if (scr_dyn_generator(value)) return value->v.typed_ref.type_key[0] == 'a' ? 2 : 1;
if (scr_async_iterator_symbol) {
ScrDyn *method = scr_dyn_symbol_key_get(value, scr_async_iterator_symbol, false);
if (!method) return -1;
bool set = method->kind != SCR_DYN_UNDEF && method->kind != SCR_DYN_NULL;
scr_dyn_release(method);
if (set) return 2;
}
ScrDyn *method = scr_iterator_method(value);
if (!method) return -1;
bool set = method->kind != SCR_DYN_UNDEF && method->kind != SCR_DYN_NULL;
scr_dyn_release(method);
return set ? 1 : 0;
}
/* Compiled class methods are dispatched before this checked-object path. */
ScrDyn *scr_dyn_iterator_optional(const ScrDyn *source) {
ScrDyn *method = scr_iterator_method(source);
+16 -3
View File
@@ -258,7 +258,11 @@ static inline bool scr_map_lookup_eq(const ScrMap *m, uint64_t stored, uint64_t
/* Compact tables keep a hash fingerprint beside each 24-bit entry index.
* Rejecting a collision here avoids loading an unrelated ordered entry.
* Larger tables keep full-width indices; collection size remains limited
* only by the allocator. Zero is empty in compact tables, so store index+1. */
* only by the allocator. Zero is empty in compact tables, so store index+1.
* The LLVM backend's inline number-key probe (sc_map_entry_f64_hashed)
* reads this encoding for tables of at most 1 << 24 buckets, hashes keys
* as scr_map_find_f64 does, and calls scr_map_entry_f64 for wider tables;
* only the runtime unit test builds a smaller bound. */
#ifndef SCR_MAP_COMPACT_BUCKETS
#define SCR_MAP_COMPACT_BUCKETS ((size_t)1 << 24)
#endif
@@ -999,10 +1003,19 @@ static inline bool scr_map_append_grows_index(const ScrMap *m) {
/* Number keys arrive normalized (+0 for -0, canonical NaN). A directly
* indexed map overwrites or appends through its table; a key the table
* cannot hold converts it to hashing first. */
* cannot hold converts it to hashing first. When an append would grow a
* hashed map's index, the lookup that decides whether the key is new also
* overwrites a key that is already present. */
static void scr_map_set_f64_slot(ScrMap *m, uint64_t key, uint64_t val) {
double k = scr_map_slot_to_f64(key);
if (!m->ndense && scr_map_append_grows_index(m) && scr_map_find_f64(m, k) == SCR_MAP_EMPTY) {
if (!m->ndense && scr_map_append_grows_index(m)) {
size_t e = scr_map_find_f64(m, k);
if (e != SCR_MAP_EMPTY) {
uint64_t old = m->entries[e].val;
m->entries[e].val = val; /* unlink before releasing (cycle collector) */
scr_map_release_val(m, old);
return;
}
scr_map_enter_dense(m, k); /* the append would grow the index */
}
if (m->ndense) {
+5 -1
View File
@@ -119,7 +119,8 @@ _Static_assert(offsetof(ScrMap, key_trace) == offsetof(ScrMap, val_trace) + 3 *
/* The LLVM backend's inline number-key lookup (sc_map_entry_f64) reads
* entries, dense and ndense through %ScrMapIx fields 13, 19 and 20, and
* entries as %ScrMapEntry { key, val, hash }; %ScrMapIx field 11 is the
* i64 filter. */
* i64 filter. Its hashed probe (sc_map_entry_f64_hashed) also reads the
* entry count, nbuckets and buckets as fields 9, 14 and 15. */
_Static_assert(offsetof(ScrMap, filter) ==
(offsetof(ScrMap, nlive) + sizeof(size_t) + _Alignof(uint64_t) - 1) / _Alignof(uint64_t) * _Alignof(uint64_t),
"%ScrMapIx field 11 is ScrMap.filter, directly after nlive");
@@ -132,6 +133,9 @@ _Static_assert(offsetof(ScrMap, nentries) + sizeof(size_t) == offsetof(ScrMap, n
_Static_assert(offsetof(ScrMapEntry, hash) == 16, "inline iteration reads ScrMapEntry.hash at byte 16");
_Static_assert(offsetof(ScrMap, entries) == offsetof(ScrMap, filter) + sizeof(uint64_t) + sizeof(size_t),
"inline number-key lookups read ScrMap.entries as %ScrMapIx field 13");
_Static_assert(offsetof(ScrMap, nbuckets) == offsetof(ScrMap, entries) + sizeof(void *) &&
offsetof(ScrMap, buckets) == offsetof(ScrMap, nbuckets) + sizeof(size_t),
"inline number-key probes read ScrMap.nbuckets and buckets as %ScrMapIx fields 14 and 15");
_Static_assert(offsetof(ScrMap, dense) == offsetof(ScrMap, entries) + 6 * sizeof(void *),
"inline number-key lookups read ScrMap.dense as %ScrMapIx field 19");
_Static_assert(offsetof(ScrMap, ndense) == offsetof(ScrMap, dense) + sizeof(void *),
File diff suppressed because it is too large Load Diff
+56 -9
View File
@@ -160,16 +160,22 @@ ScrClosure *scr_stream_finished(ScrStream *s, ScrClosure *cb, ScrStreamErrInv in
* the slot): success calls with NO arguments, errors box through the
* boundary encoding. cb's cap boxes the dyn callable. */
void scr_stream_finished_dyn_inv(ScrClosure *cb, ScrStream *s, ScrError *err);
/* stream.pipeline(s1..sn, cb) — the callback form over stream arguments:
* chains pipes (end: true), destroys every other stream with the first
* error (Node's destroyer, pipeline order), and calls cb once after the
* LAST 'close' with that error (or NULL). Streams borrowed; cb MOVES;
* answers the destination +1 (pipeline's return value). */
/* stream.pipeline(s1..sn, cb) — the callback form over stream arguments
* (pipeline.js): pipes each pair, destroys the unfinished streams with
* the first error, and calls cb on a nextTick once every stream's eos
* watcher finished, with that error (or NULL). Streams borrowed; cb
* MOVES; answers the destination +1 (pipeline's return value), or NULL
* with ERR_STREAM_UNABLE_TO_PIPE / ERR_INVALID_ARG_TYPE pending. */
ScrStream *scr_stream_pipeline(double n, ScrStream **streams, ScrClosure *cb, ScrStreamErrInv inv);
/* The dyn-valued twins (the callback as a checked-dynamic value —
* mustCall wrappers; borrowed). */
ScrClosure *scr_stream_finished_dyn(ScrStream *s, struct ScrDyn *cb);
ScrStream *scr_stream_pipeline_dyn(double n, ScrStream **streams, struct ScrDyn *cb);
/* stream.compose(s1..sn): a pipeline over the streams and a Duplex that
* writes to the first and reads from the last (compose.js). One stream
* answers itself (Duplex.from of a duplex). +1, or NULL with
* ERR_MISSING_ARGS / ERR_INVALID_ARG_VALUE pending. Streams borrowed. */
ScrStream *scr_stream_compose(double n, ScrStream **streams);
/* Emitted subclass RC/trace delegates for the state block (NULL-safe —
* an exception before super(options) leaves st unset). The registry
@@ -247,10 +253,23 @@ ScrPromise *scr_sc_text(ScrStream *s);
ScrPromise *scr_sc_json(ScrStream *s);
ScrPromise *scr_sc_buffer(ScrStream *s);
ScrPromise *scr_sc_array_buffer(ScrStream *s);
/* Readable.from(array): +1 fully-seeded object-entry stream (one WHOLE
* chunk per element — strings or Buffers per the flag; hwm 1, already
* EOF'd). Borrows arr. */
ScrStream *scr_stream_from_arr(ScrArr *arr, bool strings);
/* Readable.from over a typed source: arr holds the string or Buffer
* elements (one element under SCR_FROM_SINGLE, from.js's string/Buffer
* case). Elements push lazily, one _read at a time, as whole entries
* unless SCR_FROM_BYTE_MODE ({ objectMode: false }). hwm applies under
* SCR_FROM_HAS_HWM. +1, or NULL with ERR_INVALID_ARG_VALUE pending.
* Borrows arr. */
enum {
SCR_FROM_STRINGS = 1,
SCR_FROM_SINGLE = 2,
SCR_FROM_BYTE_MODE = 4,
SCR_FROM_HAS_HWM = 8,
};
ScrStream *scr_stream_from_arr(ScrArr *arr, double flags, double hwm);
/* Readable.from(iterable, options) in the JS lane: strings, Buffers and
* arrays of them; options objectMode and highWaterMark. +1, or NULL with
* the error pending (ERR_INVALID_ARG_TYPE for a non-iterable). Borrows. */
ScrStream *scr_stream_from_dyn(ScrDyn *src, ScrDyn *opts);
ScrStream *scr_stream_pause(ScrStream *s); /* recv +1 */
ScrStream *scr_stream_resume(ScrStream *s); /* recv +1 */
bool scr_stream_is_paused(ScrStream *s);
@@ -279,6 +298,34 @@ ScrStream *scr_stream_iterator_close(ScrStream *s);
* NULL with ERR_INVALID_ARG_TYPE pending. */
ScrStream *scr_stream_add_abort_signal(ScrStream *s, ScrDyn *signal, bool from_option,
ScrAbortAddFn add);
/* Object-mode streams for the stream operators: Readable.from's stream
* (highWaterMark 1) around native callbacks (+1; callbacks move), and its
* push: whole string or Buffer entries, the first push fixing the kind.
* Answers push()'s result, -1 for another kind of value, -2 for a kind
* that differs from the first push's. */
ScrStream *scr_stream_new_object_readable(ScrClosure *read, ScrStreamReadInv read_inv,
ScrClosure *destroy, ScrStreamErrInv destroy_inv);
int scr_stream_push_object(ScrStream *s, const ScrDyn *v);
/* The Readable operators (scr_stream_ops.c, the streamOps feature): the
* callbacks, counts and options are checked-dynamic values (borrowed).
* The stream-returning ones answer the +1 object-mode output stream, or
* NULL with Node's validation error pending; the promise-returning ones
* answer a +1 promise and reject instead of throwing. toArray settles a
* dyn array, find and reduce a dyn value — reduce's `kind` picks the
* payload instead: 0 dyn, 1 number, 2 boolean, 3 string — some and every
* a boolean, forEach nothing. */
ScrStream *scr_stream_op_map(ScrStream *s, ScrDyn *fn, ScrDyn *options);
ScrStream *scr_stream_op_filter(ScrStream *s, ScrDyn *fn, ScrDyn *options);
ScrStream *scr_stream_op_flat_map(ScrStream *s, ScrDyn *fn, ScrDyn *options);
ScrStream *scr_stream_op_drop(ScrStream *s, ScrDyn *number, ScrDyn *options);
ScrStream *scr_stream_op_take(ScrStream *s, ScrDyn *number, ScrDyn *options);
ScrPromise *scr_stream_op_to_array(ScrStream *s, ScrDyn *options);
ScrPromise *scr_stream_op_for_each(ScrStream *s, ScrDyn *fn, ScrDyn *options);
ScrPromise *scr_stream_op_some(ScrStream *s, ScrDyn *fn, ScrDyn *options);
ScrPromise *scr_stream_op_every(ScrStream *s, ScrDyn *fn, ScrDyn *options);
ScrPromise *scr_stream_op_find(ScrStream *s, ScrDyn *fn, ScrDyn *options);
ScrPromise *scr_stream_op_reduce(ScrStream *s, ScrDyn *fn, ScrDyn *init, bool has_init,
ScrDyn *options, double kind);
/* finished(s, options[, cb]) — Node's eos with its listeners and state
* predicates: readable/writable are -1 (the default), 0 or 1; on_error is
* options.error !== false; signal is NULL or borrowed (ERR_INVALID_ARG_TYPE
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,21 @@
import { Readable } from "node:stream";
const bytes = (): Readable => new Readable({ read() { this.push(Buffer.from("ab")); this.push(Buffer.from("cd")); this.push(null); } });
console.log(await bytes().toArray());
console.log(await bytes().map((b: Buffer) => b.toString("hex")).toArray());
console.log(await bytes().filter((b: Buffer) => b[0] === 0x63).toArray());
console.log(await bytes().drop(1).toArray(), await bytes().take(1).toArray());
console.log(await bytes().reduce((n: number, b: Buffer) => n + b.length, 0));
console.log(await bytes().find((b: Buffer) => b.length === 2), await bytes().every((b: Buffer) => b.length === 2));
console.log(await Readable.from(["a", "b"]).flatMap((c: string) => [c, c + c]).toArray());
console.log(await Readable.from(["a", "b"]).map(async (c: string) => c + "!", { concurrency: 2 }).toArray());
try { await Readable.from(["x"]).reduce((a: string, b: string) => a + b); } catch (e) { console.log((e as Error & { code?: string }).code, (e as Error).message); }
try { Readable.from(["a"]).drop(-1); } catch (e) { console.log((e as Error & { code?: string }).code, (e as Error).message); }
try { Readable.from(["a"]).map((x: string) => x, { concurrency: 0 }); } catch (e) { console.log((e as Error & { code?: string }).code, (e as Error).message); }
const ac = new AbortController();
ac.abort();
try { await Readable.from(["a"]).toArray({ signal: ac.signal }); } catch (e) { console.log("toArray aborted", (e as Error).name); }
try { await Readable.from(["a"]).map((x: string) => x, { signal: ac.signal }).toArray(); } catch (e) { console.log("map aborted", (e as Error).name); }
try { await Readable.from(["a"]).reduce((a: string, b: string) => a + b, "", { signal: ac.signal }); } catch (e) { console.log("reduce aborted", (e as Error).name); }
const m = Readable.from(["a"]).map((x: string) => x);
console.log(m.readableObjectMode, m.readableHighWaterMark);
@@ -0,0 +1,14 @@
import { Readable } from "node:stream";
const log: string[] = [];
const p = Readable.from(["a", "b"]).toArray().then((a: unknown) => log.push(`toArray ${String(a)}`));
const q = Readable.from(["x", "y"]).some((c: string) => { log.push(`some ${c}`); return c === "x"; }).then((v: boolean) => log.push(`some -> ${v}`));
const r = Readable.from(["1", "2"]).every((c: string) => { log.push(`every ${c}`); return c !== "2"; }).then((v: boolean) => log.push(`every -> ${v}`));
const s = Readable.from(["k", "l"]).find((c: string) => c === "l").then((v: unknown) => log.push(`find -> ${String(v)}`));
const t = Readable.from(["m", "n"]).forEach((c: string) => { log.push(`each ${c}`); }).then(() => log.push("forEach done"));
const u = Readable.from(["p", "q"]).reduce((acc: string, c: string) => acc + c, ">").then((v: string) => log.push(`reduce -> ${v}`));
let i = 0;
const hop = (): void => { if (i++ < 40) { log.push(`hop ${i}`); Promise.resolve().then(hop); } };
Promise.resolve().then(hop);
await Promise.all([p, q, r, s, t, u]);
await new Promise<void>((res) => setTimeout(res, 5));
console.log(log.join("\n"));
@@ -0,0 +1,44 @@
import { Readable } from "node:stream";
// map output interleaved with microtasks, ticks and immediates
await new Promise<void>((done) => {
const log: string[] = [];
const src = Readable.from(["a", "b", "c"]);
const out = src.map((c: string) => {
log.push(`map ${c}`);
return c.toUpperCase();
});
out.on("data", (c: string) => log.push(`data ${c}`));
out.on("end", () => {
log.push("end");
console.log(log.join("\n"));
done();
});
Promise.resolve()
.then(() => log.push("micro1"))
.then(() => log.push("micro2"))
.then(() => log.push("micro3"));
process.nextTick(() => log.push("tick"));
setImmediate(() => log.push("immediate"));
});
// concurrency 2 keeps output order; take() destroys its source. "a" is
// released only after "c" ends and the others resume on setImmediate, so
// the order does not depend on timer precision.
const log: string[] = [];
const gate = Promise.withResolvers<void>();
const out = Readable.from(["a", "b", "c", "d"]).map(async (c: string) => {
log.push(`start ${c}`);
if (c === "a") await gate.promise;
else await new Promise<void>((r) => setImmediate(r));
log.push(`end ${c}`);
if (c === "c") gate.resolve();
return c + c;
}, { concurrency: 2 });
for await (const v of out) log.push(`got ${String(v)}`);
console.log(log.join("\n"));
const taken = Readable.from(["1", "2", "3", "4"]);
const firstTwo = await taken.map((c: string) => { log.length = 0; return c; }).take(2).toArray();
console.log(firstTwo, taken.destroyed);
const dropped = await Readable.from(["1", "2", "3"]).drop(1.5).toArray();
console.log(dropped);
@@ -0,0 +1,38 @@
'use strict';
const { Readable } = require('node:stream');
const r = () => Readable.from(['a', 'b', 'c']);
function syncCheck(label, f) {
try { f(); console.log(label, 'ok'); } catch (e) { console.log(label, e.name, e.code, e.message); }
}
async function asyncCheck(label, p) {
console.log(label);
try { await p; } catch (e) { console.log(' rejected', e.name, e.code, e.message); }
}
async function main() {
syncCheck('map(5)', () => r().map(5));
syncCheck('filter(null)', () => r().filter(null));
syncCheck('map options 7', () => r().map((x) => x, 7));
syncCheck('map options []', () => r().map((x) => x, []));
syncCheck('map signal {}', () => r().map((x) => x, { signal: {} }));
syncCheck('map concurrency NaN', () => r().map((x) => x, { concurrency: 'two' }));
syncCheck('map concurrency 1.5', () => r().map((x) => x, { concurrency: 1.5 }));
syncCheck('map highWaterMark -1', () => r().map((x) => x, { highWaterMark: -1 }));
syncCheck('take(-0.5)', () => r().take(-0.5));
syncCheck('drop("x")', () => r().drop('x'));
syncCheck('take options null', () => r().take(1, null));
await asyncCheck('forEach(1)', r().forEach(1));
await asyncCheck('toArray(3)', r().toArray(3));
await asyncCheck('reduce(null)', r().reduce(null));
await asyncCheck('some({})', r().some({}));
await asyncCheck('every signal 1', r().every((x) => x, { signal: 1 }));
try { await Readable.from([]).reduce((a, b) => a + b); } catch (e) { console.log(e.name, e.code, e.message); }
console.log(await Readable.from(['x']).reduce((a, b) => a + b));
console.log(await Readable.from(['1', '2']).reduce((a, b) => a + Number(b), 0));
console.log(await r().map((c) => c + c).filter((c) => c !== 'bb').toArray());
console.log(await r().drop(0).take(Infinity).toArray(), await r().take(0).toArray());
const thrower = r().map((c) => { if (c === 'b') throw new Error('bad ' + c); return c; });
thrower.on('error', (e) => console.log('stream error', e.message));
try { for await (const v of thrower) console.log('v', v); } catch (e) { console.log('loop threw', e.message); }
try { await r().forEach(async (c) => { if (c === 'c') throw new Error('async bad'); }); } catch (e) { console.log('forEach rejected', e.message); }
}
main();
@@ -0,0 +1,17 @@
import { Readable } from "node:stream";
const log: string[] = [];
let i = 0;
const hop = (): void => { if (i++ < 12) { log.push(`hop ${i}`); Promise.resolve().then(hop); } };
async function run(): Promise<void> {
for await (const c of Readable.from(["a", "b"])) log.push(`got ${String(c)}`);
log.push("loop done");
}
Promise.resolve().then(hop);
const p = run();
await p;
await new Promise<void>((r) => setTimeout(r, 5));
console.log(log.join(" | "));
// pushes made inside the iterator's own _read arrive as one chunk
const r = new Readable({ read() { this.push(Buffer.from("ab")); this.push(Buffer.from("cd")); this.push(null); } });
for await (const c of r) console.log("chunk", c);
@@ -0,0 +1,48 @@
import { Readable } from "node:stream";
const a = Readable.from(["x", "y"]);
console.log("array", a.readableLength, a.readableHighWaterMark, a.readableObjectMode);
const s = Readable.from("hello");
console.log("string", s.readableLength, s.readableHighWaterMark, s.readableObjectMode);
const b = Readable.from(Buffer.from("hi"));
console.log("buffer", b.readableHighWaterMark, b.readableObjectMode);
const bytes = Readable.from(["ab", "cd"], { objectMode: false });
console.log("objectMode false", bytes.readableObjectMode, bytes.readableHighWaterMark);
bytes.on("data", (c: Buffer) => console.log("bytes data", c));
const hw = Readable.from(["1", "2", "3"], { highWaterMark: 2 });
console.log("hwm 2", hw.readableHighWaterMark);
hw.on("data", (c: string) => console.log("hw data", c, hw.readableLength));
hw.on("close", () => console.log("hw close"));
for (const h of [-1, 1.5, NaN, Infinity]) {
try {
Readable.from(["a"], { highWaterMark: h });
} catch (err) {
console.log((err as NodeJS.ErrnoException).code, (err as Error).message);
}
}
const lazy = Readable.from([Buffer.from("p"), Buffer.from("q")]);
lazy.on("readable", () => {
let c: Buffer | null;
while ((c = lazy.read()) !== null) console.log("read", String(c), lazy.readableLength);
});
lazy.on("end", () => console.log("lazy end"));
lazy.on("close", () => console.log("lazy close"));
const sb = Readable.from(Buffer.from("xyz"), { objectMode: false });
sb.on("data", (c: Buffer) => console.log("single bytes", c, sb.readableHighWaterMark));
const d = Readable.from(["d1", "d2"]);
d.on("close", () => console.log("destroyed close"));
d.destroy();
process.nextTick(() => console.log("tick 1"));
queueMicrotask(() => console.log("micro 1"));
const ord = Readable.from(["o1", "o2", "o3"]);
ord.on("data", (c: string) => {
console.log("ord", c);
process.nextTick(() => console.log("ord tick", c));
});
ord.on("end", () => console.log("ord end"));
ord.on("close", () => console.log("ord close"));
// an `any` source takes the checked-dynamic form, options included
const anySrc: any = JSON.parse('["a","b"]');
const fromAny = Readable.from(anySrc, { objectMode: false });
console.log("any source", fromAny.readableObjectMode, fromAny.readableHighWaterMark);
fromAny.on("data", (c: Buffer) => console.log("any data", c));
@@ -0,0 +1,34 @@
// @no-node-shims
'use strict';
const { Readable } = require('node:stream');
const e = Readable.from([]);
e.on('end', () => console.log('empty end'));
e.resume();
for (const bad of [42, null, undefined, { a: 1 }, true, Promise.resolve(1)]) {
try { Readable.from(bad); console.log('accepted', bad); } catch (err) { console.log(err.code, err.message); }
}
const n = Readable.from(['ok', null, 'later']);
n.on('data', (c) => console.log('null-stream data', c));
n.on('error', (err) => console.log('null-stream error', err.name, err.code, err.message));
n.on('close', () => console.log('null-stream close'));
const bm = Readable.from(['ok', 7], { objectMode: false });
bm.on('data', (c) => console.log('bm data', c));
bm.on('error', (err) => console.log('bm error', err.name, err.code, err.message));
const dynSrc = JSON.parse('["j1","j2"]');
const j = Readable.from(dynSrc, { highWaterMark: 4 });
console.log('dyn', j.readableHighWaterMark, j.readableObjectMode, j.readableLength);
j.on('data', (c) => console.log('dyn data', typeof c, c));
const und = Readable.from(['u'], { highWaterMark: undefined });
console.log('undef hwm', und.readableHighWaterMark);
const und2 = Readable.from(['u'], { objectMode: false, highWaterMark: null });
console.log('null hwm bytes', und2.readableHighWaterMark, und2.readableObjectMode);
try { Readable.from(['x'], { highWaterMark: '2' }); } catch (err) { console.log(err.code, err.message); }
const one = Readable.from('single');
one.on('data', (c) => console.log('single', c));
const opt = Readable.from(Buffer.from('bb'), JSON.parse('{"objectMode":false}'));
opt.on('data', (c) => console.log('opt', c, opt.readableObjectMode));
(async () => {
for await (const c of Readable.from(['f1', 'f2'])) console.log('for await', c);
const arr = await Readable.from(['t1', 't2']).toArray();
console.log('toArray', arr);
})();
+72
View File
@@ -0,0 +1,72 @@
import { compose, PassThrough, Readable, Transform, Writable } from "node:stream";
const upper = new Transform({
transform(chunk: Buffer, _enc, cb) {
cb(null, chunk.toString().toUpperCase());
},
});
const exclaim = new Transform({
transform(chunk: Buffer, _enc, cb) {
cb(null, chunk.toString() + "!");
},
});
const d = compose(upper, exclaim);
console.log("duplex", d.readable, d.writable, d.readableObjectMode);
d.on("data", (c: Buffer) => console.log("data", c.toString()));
d.on("finish", () => console.log("finish"));
d.on("end", () => console.log("end"));
d.on("close", () => console.log("close"));
d.write("a");
d.write("b");
d.end("c");
const p = new PassThrough();
console.log("single", compose(p) === p);
try {
compose();
} catch (err) {
console.log((err as Error).name, (err as NodeJS.ErrnoException).code, (err as Error).message);
}
// A readable head: the composed stream is read-only.
const src = Readable.from(["x", "y"]);
const pt = new PassThrough();
const r = compose(src, pt);
console.log("read-only", r.readable, r.writable);
r.on("data", (c: Buffer) => console.log("r data", c.toString()));
r.on("end", () => console.log("r end"));
r.on("close", () => console.log("r close"));
// A writable tail: the composed stream is write-only.
const seen: string[] = [];
const sink = new Writable({
write(chunk: Buffer, _enc, cb) {
seen.push(chunk.toString());
cb();
},
});
const w = compose(new PassThrough(), sink);
console.log("write-only", w.readable, w.writable);
w.on("finish", () => console.log("w finish", seen.join(",")));
w.on("close", () => console.log("w close"));
w.end("z");
// Errors: a failing stage destroys the composed stream.
const bad = new Transform({
transform(_chunk: Buffer, _enc, cb) {
cb(new Error("boom"));
},
});
const e = compose(new PassThrough(), bad);
e.on("error", (err: Error) => console.log("e error", err.message));
e.on("close", () => console.log("e close"));
e.write("q");
// destroy() before the pipeline ends: AbortError.
const k = compose(new PassThrough(), new PassThrough());
k.on("error", (err: Error) => console.log("k error", err.name, (err as NodeJS.ErrnoException).code));
k.on("close", () => console.log("k close"));
k.destroy();
process.nextTick(() => console.log("tick"));
+51
View File
@@ -0,0 +1,51 @@
import { compose, PassThrough, Readable, Transform, Writable } from "node:stream";
// object-mode tail: Readable.from strings through compose
const objs = compose(Readable.from(["a", "b", "c"]), new PassThrough({ objectMode: false }));
console.log("objs", objs.readableObjectMode, objs.readableHighWaterMark);
objs.on("data", (c: Buffer) => console.log("objs data", c.toString()));
objs.on("end", () => console.log("objs end"));
// an operator stream as the tail: object mode
const mapped = compose(new PassThrough(), new PassThrough()).map((c: Buffer) => c.toString() + "?");
mapped.on("data", (c: string) => console.log("mapped", c));
// backpressure: a slow sink makes the composed writable buffer
const slow = new Writable({
highWaterMark: 2,
write(chunk: Buffer, _enc, cb) {
console.log("slow write", chunk.toString());
setTimeout(cb, 1);
},
});
const bp = compose(new PassThrough({ highWaterMark: 2 }), slow);
console.log("bp write 1", bp.write("aa"));
console.log("bp write 2", bp.write("bb"));
console.log("bp write 3", bp.write("cc"));
bp.on("drain", () => console.log("bp drain"));
bp.on("finish", () => console.log("bp finish"));
bp.end();
// a destroyed stream in the chain
const gone = new PassThrough();
gone.destroy();
try {
compose(gone, new PassThrough());
} catch (err) {
console.log((err as NodeJS.ErrnoException).code, (err as Error).message.slice(0, 44));
}
// upper-casing round trip with a for-await reader
const up = compose(
new Transform({
transform(c: Buffer, _e, cb) {
cb(null, c.toString().toUpperCase());
},
}),
new PassThrough(),
);
(async () => {
up.end("hello");
for await (const c of up) console.log("for await", String(c));
console.log("done");
})();
@@ -0,0 +1,71 @@
import { PassThrough, pipeline, Transform, Writable } from "node:stream";
const a = new PassThrough();
const b = new PassThrough();
b.resume();
b.on("close", () => {
console.log("b close");
process.nextTick(() => console.log("tick after b close"));
queueMicrotask(() => console.log("micro after b close"));
});
pipeline(a, b, (err?: Error | null) => console.log("cb", err ? err.message : "ok"));
a.end("x");
const c = new PassThrough();
const d = new PassThrough();
d.on("close", () => {
console.log("d close");
process.nextTick(() => console.log("tick after d close"));
});
c.on("close", () => {
console.log("c close");
process.nextTick(() => console.log("tick after c close"));
});
pipeline(c, d, (err?: Error | null) => console.log("cb2", err ? err.message : "ok"));
d.destroy(new Error("bad"));
// a synchronous write error reaches 'error' one tick later
let k = 0;
const note = (s: string) => console.log(k++, s);
const src = new PassThrough();
const failing = new Transform({
transform(chunk: Buffer, _enc, cb) {
note("transform " + chunk.toString());
cb(new Error("boom"));
},
});
failing.on("error", (err: Error) => note("failing error " + err.message));
src.on("resume", () => note("src resume"));
src.pipe(failing);
src.write("q");
note("after write");
let u = 0;
const step = () => {
if (u++ < 3) {
note("step " + u);
process.nextTick(step);
}
};
process.nextTick(step);
// 'finish' waits for a pending _final callback, even when end() runs again
const slowFinal = new Writable({
write(_c, _e, cb) {
cb();
},
final(cb) {
console.log("final called");
setTimeout(() => {
console.log("final done");
cb();
}, 5);
},
});
slowFinal.on("finish", () => console.log("slow final finish"));
slowFinal.write("a");
slowFinal.end();
setTimeout(() => {
slowFinal.end();
console.log("second end");
}, 1);
+104
View File
@@ -0,0 +1,104 @@
// Number-keyed Map and Set lookups on hashed storage: small maps that
// scan their entries, bucket tables, keys that need SameValueZero
// normalization, deletes and clears between reads, and maps that move
// between the direct index and hashing.
const special = [0, -0, NaN, Infinity, -Infinity, 0.5, -1, 2 ** 53, 1e300, Number.MIN_VALUE, Number.MAX_VALUE];
function describe(m: Map<number, string>, keys: number[]): string {
return keys.map((k) => `${Object.is(k, -0) ? "-0" : k}:${m.has(k) ? m.get(k) : "-"}`).join(" ");
}
// Small maps keep no bucket table.
const small = new Map<number, string>();
small.set(1e300, "big");
small.set(-0, "zero");
small.set(NaN, "nan");
console.log("small", small.size, describe(small, special));
console.log("small keys", [...small.keys()].map((k) => (Object.is(k, -0) ? "-0" : String(k))).join(","));
small.delete(0);
console.log("small after delete", small.size, describe(small, special));
small.set(0, "again");
console.log("small re-add", describe(small, [0, -0, NaN, 1e300]));
// Bucket tables of sparse ids, with hits, misses and tombstones.
const table = new Map<number, number>();
for (let i = 0; i < 3000; i++) table.set(100000 + i * 7919, i);
let hits = 0,
misses = 0,
sum = 0;
for (let i = 0; i < 6000; i++) {
const k = 100000 + i * 7919;
const v = table.get(k);
if (v === undefined) misses++;
else {
hits++;
sum += v;
}
if (table.has(k + 1)) hits += 1000000;
}
console.log("table", table.size, hits, misses, sum);
for (let i = 0; i < 3000; i += 3) table.delete(100000 + i * 7919);
let live = 0;
for (let i = 0; i < 3000; i++) if (table.has(100000 + i * 7919)) live++;
console.log("table after deletes", table.size, live, table.get(100000), table.get(100000 + 7919));
for (const k of special) table.set(k, -1);
console.log("table specials", table.size, special.map((k) => table.get(k)).join(","), table.get(0.25));
table.clear();
console.log("table cleared", table.size, table.get(100000 + 7919), table.has(NaN));
// A read, a write of the same key, then the read again.
const m = new Map<number, number>();
for (let i = 0; i < 64; i++) m.set(i * 1000003, i);
const key = 5 * 1000003;
const before = m.get(key);
m.delete(key);
const afterDelete = m.get(key);
const hadIt = m.has(key);
m.set(key, 99);
const afterSet = m.get(key);
for (let i = 64; i < 400; i++) m.set(i * 1000003, i); // grows the bucket table
const afterGrowth = m.get(key);
console.log("reread", before, afterDelete, hadIt, afterSet, afterGrowth, m.size);
// Direct index to hashing and back.
const ids = new Map<number, string>();
for (let i = 0; i < 40; i++) ids.set(i, `id${i}`);
console.log("dense", ids.get(0), ids.get(39), ids.get(40), ids.has(-1), ids.get(-0));
ids.set(-5, "neg");
ids.set(1.5, "half");
console.log("hashed", ids.size, ids.get(0), ids.get(39), ids.get(-5), ids.get(1.5), ids.get(2.5));
ids.delete(-5);
ids.delete(1.5);
for (let i = 40; i < 200; i++) ids.set(i, `id${i}`);
console.log("regrown", ids.size, ids.get(0), ids.get(150), ids.get(-5), ids.get(1.5));
// Sets and boolean and reference values.
const seen = new Set<number>();
for (let i = 0; i < 500; i++) seen.add((i * 2654435761) % 4294967296);
let found = 0;
for (let i = 0; i < 1000; i++) if (seen.has((i * 2654435761) % 4294967296)) found++;
console.log("set", seen.size, found, seen.has(NaN), seen.has(-0));
seen.add(-0);
seen.add(NaN);
console.log("set specials", seen.has(0), seen.has(NaN), seen.size);
const flags = new Map<number, boolean>();
for (let i = 0; i < 100; i++) flags.set(i * 37 + 1e6, i % 3 === 0);
let truthy = 0;
for (let i = 0; i < 100; i++) if (flags.get(i * 37 + 1e6)) truthy++;
console.log("flags", truthy, flags.get(1e6), flags.get(1e6 + 37), flags.get(5));
interface Node {
id: number;
label: string;
}
const nodes = new Map<number, Node>();
for (let i = 0; i < 300; i++) nodes.set(50000 + i * 13, { id: i, label: `n${i}` });
let labels = "";
for (let i = 0; i < 300; i += 37) {
const node = nodes.get(50000 + i * 13);
labels += node === undefined ? "?" : node.label;
if (nodes.has(50000 + i * 13 + 1)) labels += "!";
}
console.log("nodes", nodes.size, labels, nodes.get(49999)?.label ?? "none");
@@ -0,0 +1,61 @@
// Number-key Map.set and Set.add of present keys while the map is at its
// bucket-table growth boundary (the next new key grows the index):
// overwrites keep the entry, its position and the size, and release the
// value they replace.
interface Box {
label: string;
}
function sparse(i: number): number {
return 1000 + i * 7919;
}
for (const size of [8, 64, 1024]) {
const nums = new Map<number, number>();
const boxes = new Map<number, Box>();
const names = new Map<number, string>();
const seen = new Set<number>();
for (let i = 0; i < size; i++) {
nums.set(sparse(i), i);
boxes.set(sparse(i), { label: `b${i}` });
names.set(sparse(i), `n${i}`);
seen.add(sparse(i));
}
for (let round = 0; round < 3; round++) {
for (let i = size - 1; i >= 0; i--) {
nums.set(sparse(i), i * 10 + round);
boxes.set(sparse(i), { label: `b${i}.${round}` });
names.set(sparse(i), `n${i}.${round}`.repeat(2));
seen.add(sparse(i));
}
}
let sum = 0;
for (const v of nums.values()) sum += v;
const firstKeys = [...nums.keys()].slice(0, 3).join(",");
console.log(
size,
nums.size,
boxes.size,
names.size,
seen.size,
sum,
firstKeys,
boxes.get(sparse(size - 1))?.label,
names.get(sparse(0)),
[...seen].slice(0, 2).join(","),
);
// One more key grows the index; existing entries are unchanged.
nums.set(-1, -1);
boxes.set(-1, { label: "new" });
console.log(size, "grown", nums.size, nums.get(sparse(size >> 1)), nums.get(-1), boxes.get(-1)?.label, boxes.get(sparse(1))?.label);
}
// At the boundary after deletes: a delete frees no bucket, so re-adding
// the deleted key compacts instead of growing.
const churn = new Map<number, number>();
for (let i = 0; i < 64; i++) churn.set(sparse(i), i);
churn.delete(sparse(10));
churn.set(sparse(11), 111);
churn.set(sparse(10), 110);
console.log("churn", churn.size, churn.get(sparse(10)), churn.get(sparse(11)), [...churn.keys()].indexOf(sparse(10)));
@@ -0,0 +1,28 @@
// @no-node-shims
// Readable.from over parsed JSON strings: the first lazy pull makes the
// stream hold strings, which reach 'data', for await, an operator and a
// pipe as strings; a mixed source decodes once setEncoding() is set.
'use strict';
const { Readable, Writable } = require('node:stream');
async function main() {
const fromJson = () => Readable.from(JSON.parse('["j1", "j2", "j3"]'));
const data = [];
const r = fromJson();
r.on('data', (c) => data.push(typeof c + ':' + c));
await new Promise((resolve) => r.on('end', resolve));
console.log('data', data.join(' '));
for await (const c of fromJson()) console.log('for await', typeof c, c);
console.log('toArray', await fromJson().map((c) => c.toUpperCase()).toArray());
const hex = [];
const sink = new Writable({ write(chunk, enc, cb) { hex.push(chunk.toString('hex')); cb(); } });
fromJson().pipe(sink);
await new Promise((resolve) => sink.on('finish', resolve));
console.log('piped', hex.join(','));
const mixed = Readable.from(['s1', Buffer.from('b1'), 's2']);
mixed.setEncoding('utf8');
for await (const c of mixed) console.log('mixed', typeof c, c);
const switched = fromJson();
switched.setEncoding('latin1');
console.log('switched', await switched.toArray());
}
main();
@@ -0,0 +1,18 @@
// A Readable.from stream over parsed JSON strings read through an any
// receiver: read() answers the strings, as in Node, and setEncoding()
// before the first read keeps a Buffer and string source as strings.
import { Readable } from "node:stream";
const source: any = JSON.parse('["t1", "t2"]');
const r: any = Readable.from(source);
r.once("readable", () => {
let c: any;
while ((c = r.read()) !== null) console.log("read", typeof c, c, Buffer.isBuffer(c));
});
r.on("end", () => console.log("end"));
const values: any = JSON.parse('["s1"]');
values.unshift(Buffer.from("b1"));
const mixed = Readable.from(values);
mixed.setEncoding("utf8");
mixed.on("data", (c: string) => console.log("mixed", typeof c, c));
@@ -0,0 +1,29 @@
// @no-node-shims
// setEncoding on Readable.from streams from CommonJS: Buffer, string and
// Uint8Array entries, a decoder set after the first chunk, and pipe.
'use strict';
const { Readable, Writable } = require('node:stream');
async function main() {
const mixed = Readable.from([Buffer.from('b1'), 's2', new Uint8Array([0x75, 0x33])]);
mixed.setEncoding('utf8');
for await (const c of mixed) console.log('mixed', typeof c, c);
const late = Readable.from([Buffer.from('q1'), Buffer.from('q2')], { objectMode: false });
const rest = [];
await new Promise((resolve) => {
late.once('data', (first) => {
console.log('late first', Buffer.isBuffer(first), String(first));
late.setEncoding('hex');
late.on('data', (c) => rest.push(typeof c + ':' + c));
late.on('end', () => resolve(undefined));
});
});
console.log('late rest', rest.join(' '));
const seen = [];
const w = new Writable({ write(chunk, enc, cb) { seen.push(chunk.toString('hex')); cb(); } });
const src = Readable.from([Buffer.from('w1'), Buffer.from('w2')]);
src.setEncoding('utf8');
src.pipe(w);
await new Promise((resolve) => w.on('finish', resolve));
console.log('piped', seen.join(','));
}
main();
@@ -0,0 +1,71 @@
// setEncoding on Readable.from streams: Buffer and Uint8Array entries
// decode through the stream's decoder whether they were buffered before
// setEncoding or pulled after it, in object and byte mode, and reach
// for-await, pipe, pipeline, an operator's output and a composed stream
// as strings.
import { PassThrough, Readable, Writable, compose, pipeline } from "node:stream";
const log: string[] = [];
const sink = (label: string): Writable =>
new Writable({
write(chunk: Buffer, _enc, cb) {
log.push(`${label} wrote ${chunk.toString("hex")}`);
cb();
},
});
const drain = async (label: string, r: Readable): Promise<void> => {
for await (const c of r) log.push(`${label} got ${typeof c} ${String(c)}`);
};
// setEncoding before the first lazy read
const before = Readable.from([Buffer.from("ab"), Buffer.from("c\\u00e9")]);
before.setEncoding("utf8");
await drain("before", before);
// setEncoding after the first lazy read (byte mode: Node re-counts an
// object-mode buffer's length in characters)
const after = Readable.from([Buffer.from("xy"), Buffer.from("zw"), Buffer.from("!")], {
objectMode: false,
});
await new Promise<void>((resolve) => {
after.once("data", (c: Buffer) => {
log.push(`after first ${c.toString()}`);
after.pause();
after.setEncoding("hex");
after.on("data", (s: string) => log.push(`after rest ${s}`));
after.on("end", () => resolve());
after.resume();
});
});
// Uint8Array entries, a split multi-byte character across entries
const split = Readable.from([new Uint8Array([0xe2, 0x82]), new Uint8Array([0xac, 0x41])]);
split.setEncoding("utf8");
await drain("split", split);
// byte mode
const bytes = Readable.from([Buffer.from("by"), Buffer.from("te")], { objectMode: false });
bytes.setEncoding("base64");
await drain("bytes", bytes);
// pipe and pipeline delivery of an encoded stream
const piped = Readable.from([Buffer.from("p1"), Buffer.from("p2")]);
piped.setEncoding("utf8");
await new Promise<void>((resolve) => piped.pipe(sink("pipe")).on("finish", () => resolve()));
const lined = Readable.from([Buffer.from("l1"), Buffer.from("l2")]);
lined.setEncoding("latin1");
await new Promise<void>((resolve) =>
pipeline(lined, new PassThrough(), sink("pipeline"), () => resolve()),
);
// an operator's output and a composed stream, encoded
const mapped = Readable.from([Buffer.from("m")]).map((b: Buffer) => Buffer.concat([b, b]));
mapped.setEncoding("utf8");
await drain("mapped", mapped);
const composed = compose(Readable.from([Buffer.from("k1"), Buffer.from("k2")]), new PassThrough());
composed.setEncoding("utf8");
const parts: string[] = [];
composed.on("data", (c: string) => parts.push(typeof c + ":" + c));
await new Promise<void>((resolve) => composed.on("end", () => resolve()));
log.push(`composed ${parts.join("").replace(/string:/g, "")} ${parts.every((p) => p.startsWith("string:"))}`);
console.log(log.join("\n"));
+158
View File
@@ -0,0 +1,158 @@
/* Stream chunk kinds Node cannot oracle — scriptc-only.
*
* A typed read() answers `Buffer | null`. When the read itself makes the
* stream hold strings (the first string a lazy Readable.from pull over a
* dynamic array produces, or setEncoding() inside _read), Node returns
* the string, which the typed result cannot hold, so the read throws the
* encoding fence instead (caught here, so the program exits normally).
* These pins keep that refusal exact and, under
* SCRIPTC_SAN=1 (ASan and the RC audit, like every harness lane), prove
* no string reaches Buffer code. The Node-observable cases are corpus
* programs (stream-dynamic-entry-kinds.*).
*/
import { execFile } from "node:child_process";
import { createHash } from "node:crypto";
import { mkdirSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { promisify } from "node:util";
import { describe, expect, test } from "vitest";
import { compile } from "@scriptc/compiler";
const execFileAsync = promisify(execFile);
const repoRoot = join(import.meta.dirname, "../..");
const cacheDir = join(repoRoot, "node_modules/.cache/scriptc-tests");
const sanitize = process.env["SCRIPTC_SAN"] === "1";
interface RunResult {
stdout: string;
stderr: string;
exitCode: number;
}
async function compileAndRun(name: string, source: string, ext = "ts"): Promise<RunResult> {
const key = createHash("sha256")
.update(source)
.update(sanitize ? "san" : "plain")
.digest("hex")
.slice(0, 16);
const outDir = join(cacheDir, `stream-kinds-${key}`);
mkdirSync(outDir, { recursive: true });
const file = join(outDir, `${name}.${ext}`);
writeFileSync(file, source);
const result = await compile(file, {
outPath: join(outDir, name),
outDir,
sanitize,
backend: "llvm",
});
if (!result.ok) {
throw new Error(
"stream program failed to compile:\n" +
result.diagnostics.map((d) => `${d.code}: ${d.message}`).join("\n"),
);
}
try {
const { stdout, stderr } = await execFileAsync(result.binaryPath, [], { encoding: "utf8" });
return { stdout, stderr: stripAsanFiberWarning(stderr), exitCode: 0 };
} catch (err) {
const e = err as { code?: unknown; stdout?: string; stderr?: string };
if (typeof e.code !== "number") throw err;
return {
stdout: e.stdout ?? "",
stderr: stripAsanFiberWarning(e.stderr ?? ""),
exitCode: e.code,
};
}
}
/** Linux ASan's once-per-process swapcontext warning (see errors.test.ts). */
function stripAsanFiberWarning(s: string): string {
return s.replace(
/^==\d+==WARNING: ASan doesn't fully support makecontext\/swapcontext.*\n/gm,
"",
);
}
const READ_FENCE_MESSAGE =
"read() on a stream with an encoding set is not supported yet (consume 'data' events, which deliver strings)";
describe(`typed stream reads of string chunks (scriptc-only${sanitize ? ", sanitized" : ""})`, () => {
test("a typed read() whose lazy pull yields the first string throws the encoding fence", async () => {
const r = await compileAndRun(
"typed-read-json-string",
`import { Readable } from "node:stream";
const r = Readable.from(JSON.parse('["j1", "j2"]'));
try {
const chunk = r.read();
console.log("hex", chunk === null ? "null" : chunk.toString("hex"));
} catch (err) {
console.log("caught", (err as Error).message);
}
`,
);
expect(r.exitCode).toBe(0);
expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`);
});
test("a typed read() whose _read calls setEncoding() throws the encoding fence", async () => {
const r = await compileAndRun(
"typed-read-encoding-in-read",
`import { Readable } from "node:stream";
const r = new Readable({
read() {
this.setEncoding("utf8");
this.push(Buffer.from("ab"));
this.push(null);
},
});
try {
const chunk = r.read();
console.log("hex", chunk === null ? "null" : chunk.toString("hex"));
} catch (err) {
console.log("caught", (err as Error).message);
}
`,
);
expect(r.exitCode).toBe(0);
expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`);
});
test("the JavaScript lane's read() on a JSON source throws the same fence", async () => {
const r = await compileAndRun(
"js-read-json-string",
`'use strict';
const { Readable } = require('node:stream');
const r = Readable.from(JSON.parse('["j1"]'));
r.once('readable', () => {
try {
const chunk = r.read();
console.log('got', chunk === null ? 'null' : chunk.length);
} catch (err) {
console.log('caught', err.message);
}
});
`,
"cjs",
);
expect(r.exitCode).toBe(0);
expect(r.stdout).toBe(`caught ${READ_FENCE_MESSAGE}\n`);
});
test("a dynamic source that mixes Buffers and strings without a decoder is destroyed", async () => {
const r = await compileAndRun(
"mixed-dynamic-source",
`import { Readable } from "node:stream";
const values: any = JSON.parse('["s1"]');
values.unshift(Buffer.from("b1"));
const r = Readable.from(values);
r.on("data", (c: any) => console.log("data", Buffer.isBuffer(c), String(c)));
r.on("error", (err: Error) => console.log("error", err.message));
r.on("close", () => console.log("close"));
`,
);
expect(r.exitCode).toBe(0);
expect(r.stdout).toBe(
"data true b1\nerror a Readable.from source that mixes strings and Buffers is not supported yet\nclose\n",
);
});
});
+2
View File
@@ -269,6 +269,8 @@ describe.skipIf(!zigOnPath())("wasm32-wasi differential", () => {
"5317-map-iterate-while-mutating.ts",
"5318-set-iterate-after-deletes.ts",
"5319-map-iteration-values.ts",
// Hashed number-key lookups probe ScrMap through size_t-width fields.
"5710-map-number-keys-hashed.ts",
// JSON writers read array slots as pointers.
"5305-json-stringify-typed-arrays.ts",
// Comparator sorts move 64-bit slots of 32-bit pointers.