diff --git a/benchmarks/runtime/README.md b/benchmarks/runtime/README.md index 83cfcb831..520964cf3 100644 --- a/benchmarks/runtime/README.md +++ b/benchmarks/runtime/README.md @@ -20,6 +20,7 @@ Application-shaped workloads for measuring the speed of compiled executables. Ea | `json-records` | cpu | `JSON.stringify`, typed `JSON.parse`, Map aggregation | | `json-unions` | cpu | `JSON.parse` into discriminated unions: an event log with eight kinds and nested expression trees with nine, evaluated after the cast | | `keyed-option-writes` | cpu | class methods whose names are also keys of option records written through element accesses: `enabled[name] = true` over a union of literal names and `totals["count"] = n` | +| `line-reader` | cpu | `readline` over a file stream: 'line' events for a 67 MB access log (every 50th line not ASCII), with a status count, a path tally and byte totals | | `log-lines` | cpu | 200k immediate `console.log`/`process.stdout.write` lines into the runner's pipe | | `log-summary`, `inventory-report` | cpu | the multi-module applications from `benchmarks/builds`, on 300k-line inputs | | `number-maps` | cpu | number-keyed Map/Set: a dense-id table, sparse ids in many small maps, get/has/set mixes | diff --git a/benchmarks/runtime/workloads/line-reader.json b/benchmarks/runtime/workloads/line-reader.json new file mode 100644 index 000000000..85194f70c --- /dev/null +++ b/benchmarks/runtime/workloads/line-reader.json @@ -0,0 +1,7 @@ +{ + "entry": "workloads/line-reader.ts", + "input": "line-reader", + "args": [ + "{input}" + ] +} diff --git a/benchmarks/runtime/workloads/line-reader.ts b/benchmarks/runtime/workloads/line-reader.ts new file mode 100644 index 000000000..cbdf384bb --- /dev/null +++ b/benchmarks/runtime/workloads/line-reader.ts @@ -0,0 +1,34 @@ +// Line-by-line processing of a log file with readline over a file stream: +// 'line' events for about 67 MB of access-log lines, with light work per +// line (a status count, a path tally, byte totals). +import { createReadStream } from "node:fs"; +import { createInterface } from "node:readline"; + +const path = process.argv[2]!; +const statuses = new Map(); +const paths = new Map(); +let lines = 0; +let chars = 0; +let bytes = 0; +let longest = 0; + +const input = createInterface({ input: createReadStream(path), crlfDelay: Infinity }); +input.on("line", (line: string) => { + lines++; + chars += line.length; + if (line.length > longest) longest = line.length; + const quote = line.indexOf('"'); + const space = line.indexOf(" ", quote); + const path = line.slice(space + 1, line.indexOf("/", line.indexOf("/", space) + 1)); + paths.set(path, (paths.get(path) ?? 0) + 1); + const close = line.indexOf('"', space); + const status = line.slice(close + 2, close + 5); + statuses.set(status, (statuses.get(status) ?? 0) + 1); + bytes += Number(line.slice(close + 6, line.indexOf(" ", close + 6))); +}); +input.on("close", () => { + console.log(`${lines} lines, ${chars} characters, longest ${longest}`); + console.log(`bytes ${bytes}`); + for (const status of [...statuses.keys()].sort()) console.log(`status ${status}: ${statuses.get(status)}`); + for (const prefix of [...paths.keys()].sort()) console.log(`path ${prefix}: ${paths.get(prefix)}`); +}); diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8344-readline-chunk-splitting.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8344-readline-chunk-splitting.ts.json new file mode 100644 index 000000000..8c01e6b47 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8344-readline-chunk-splitting.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/8344-readline-chunk-splitting.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_bytes.c b/packages/runtime/src/scr_bytes.c index e6df837c8..6b8594c08 100644 --- a/packages/runtime/src/scr_bytes.c +++ b/packages/runtime/src/scr_bytes.c @@ -2722,7 +2722,9 @@ static ScrBytes *scr_bytes_encode_str(ScrBytesSink *sink, const ScrStr *s, const /* Node writes charCodeAt(i) & 0xFF for BOTH spellings — an astral * code point contributes its two surrogates' low bytes. */ ScrBytes *b = scr_bytes_conversion_storage_in(sink, n); - size_t o = 0, i = 0; + /* An ASCII prefix (all of an ASCII string) is its own encoding. */ + size_t o = s->ascii ? n : scr_bytes_ascii_prefix(in, n), i = o; + memcpy(b->data, in, o); while (i < n) { uint32_t cp = scr_bytes_next_cp(in, &i); if (cp > 0xffff) { diff --git a/packages/runtime/src/scr_readline.c b/packages/runtime/src/scr_readline.c index 1bc48bbe7..178d620c6 100644 --- a/packages/runtime/src/scr_readline.c +++ b/packages/runtime/src/scr_readline.c @@ -283,7 +283,23 @@ static void scr_rl_buf_append(ScrRl *rl, const char *p, size_t n) { rl->line_len += n; } -/* kNormalWrite over decoded UTF-8 text. */ +/* The end of the line starting at `i` in text without a CR: the next LF, + * or n when the text ends first. */ +static size_t scr_rl_next_lf(const char *p, size_t i, size_t n) { + const char *lf = memchr(p + i, '\n', n - i); + return lf ? (size_t)(lf - p) : n; +} + +/* The end of the line starting at `i`: the next LF or CR, or n. */ +static size_t scr_rl_next_break(const char *p, size_t i, size_t n) { + while (i < n && p[i] != '\n' && p[i] != '\r') i++; + return i; +} + +/* kNormalWrite over decoded UTF-8 text. The text is scanned once: LF-only + * text (no CR anywhere in it) moves from LF to LF with memchr, other text + * splits at LF, CR and CRLF. The unfinished line before the text only + * joins the first line, so the rest of the text is never copied whole. */ static void scr_rl_feed_text(ScrRl *rl, const char *p, size_t n) { if (rl->tty) { scr_rl_tty_ops->feed(rl, p, n); @@ -296,43 +312,31 @@ static void scr_rl_feed_text(ScrRl *rl, const char *p, size_t n) { } rl->saw_return_at = 0; } - bool ending = false; - for (size_t i = 0; i < n && !ending; i++) ending = p[i] == '\n' || p[i] == '\r'; - if (!ending) { + bool cr = n > 0 && memchr(p, '\r', n) != NULL; + size_t end = n == 0 ? 0 : cr ? scr_rl_next_break(p, 0, n) : scr_rl_next_lf(p, 0, n); + if (end == n) { if (n > 0) scr_rl_buf_append(rl, p, n); return; } - /* The unfinished line joins the new text; the scan runs over both. */ - char *text = p ? (char *)p : (char *)""; - size_t len = n; - char *joined = NULL; - if (rl->line_len > 0) { - joined = malloc(rl->line_len + n); - if (!joined) scr_rl_oom(); - memcpy(joined, rl->line_buf, rl->line_len); - memcpy(joined + rl->line_len, p, n); - text = joined; - len = rl->line_len + n; - } - rl->line_len = 0; - rl->saw_return_at = len > 0 && text[len - 1] == '\r' ? scr_now_ms() : 0; + rl->saw_return_at = p[n - 1] == '\r' ? scr_now_ms() : 0; /* Split first (the remainder becomes the new unfinished line before any * line is delivered, as in Node), then deliver in order. */ size_t n_lines = 0, cap_lines = 0; ScrStr **lines = NULL; size_t start = 0; - for (size_t i = 0; i < len; i++) { - size_t adv; - if (text[i] == '\n') adv = 1; - else if (text[i] == '\r') adv = i + 1 < len && text[i + 1] == '\n' ? 2 : 1; - else continue; + while (end < n) { lines = scr_rl_grow(lines, &cap_lines, n_lines + 1, sizeof *lines); - lines[n_lines++] = scr_str_new(text + start, i - start); - i += adv - 1; - start = i + 1; + if (n_lines == 0 && rl->line_len > 0) { + scr_rl_buf_append(rl, p, end); + lines[n_lines++] = scr_str_new(rl->line_buf, rl->line_len); + } else { + lines[n_lines++] = scr_str_new(p + start, end - start); + } + start = end + (p[end] == '\r' && end + 1 < n && p[end + 1] == '\n' ? 2 : 1); + end = start >= n ? n : cr ? scr_rl_next_break(p, start, n) : scr_rl_next_lf(p, start, n); } - if (start < len) scr_rl_buf_append(rl, text + start, len - start); - free(joined); + rl->line_len = 0; + if (start < n) scr_rl_buf_append(rl, p + start, n - start); for (size_t i = 0; i < n_lines; i++) { if (rl->closed || scr_exc_pending()) { scr_str_release(lines[i]); @@ -366,6 +370,11 @@ static void scr_rl_feed_bytes(ScrRl *rl, const char *p, size_t n) { rl->parked_len += n; return; } + /* ASCII with no held bytes decodes to itself: split it in place. */ + if (rl->n_held == 0 && scr_bytes_ascii_span(p, n) == n) { + scr_rl_feed_text(rl, p, n); + return; + } size_t total = rl->n_held + n; unsigned char *all = malloc(total ? total : 1); if (!all) scr_rl_oom(); @@ -401,6 +410,12 @@ static void scr_rl_feed_end(ScrRl *rl) { void scr_rl_input_bytes(double id, ScrBytes *chunk) { ScrRl *rl = scr_rl_find(id); if (!rl || rl->closed) return; + /* The splitter copies every line it keeps before it delivers one, so it + * can read a chunk's own storage; shared memory is copied out first. */ + if (!chunk->shared) { + scr_rl_feed_bytes(rl, (const char *)chunk->data, chunk->len); + return; + } char *p = malloc(chunk->len ? chunk->len : 1); if (!p) scr_rl_oom(); scr_bytes_read(chunk, 0, p, chunk->len); diff --git a/scripts/bench-runtime.mjs b/scripts/bench-runtime.mjs index b45e3ebdf..d89a4e759 100644 --- a/scripts/bench-runtime.mjs +++ b/scripts/bench-runtime.mjs @@ -180,6 +180,16 @@ function generateInput(kind) { ["sku-" + i, "category-" + (i % 17), "product " + i, i % 40, 199 + (i % 5000)].join("\t"), ).join("\n") + "\n", ); + } else if (kind === "line-reader") { + // About 67 MB of access-log lines; every 50th path is not ASCII. + writeFileSync( + path, + Array.from( + { length: 600_000 }, + (_, i) => + `10.${i % 251}.${i % 7}.${i % 13} - - [24/Oct/2026:${String(i % 24).padStart(2, "0")}:${String(i % 60).padStart(2, "0")}:00 +0000] "${i % 9 === 0 ? "POST" : "GET"} /${i % 50 === 0 ? "caf\u00e9" : "api"}/items/${i % 997}?page=${i % 31} HTTP/1.1" ${i % 23 === 0 ? 404 : 200} ${(i * 7919) % 50000} "scriptc-bench/1.0"`, + ).join("\n") + "\n", + ); } else if (kind === "cli-config") { writeFileSync(path, JSON.stringify(deployManifest(), null, 2) + "\n"); } else throw new Error(`unknown input kind ${kind}`); diff --git a/scripts/zero-cost-compare.mjs b/scripts/zero-cost-compare.mjs index b23f61a0a..7ae33d64a 100644 --- a/scripts/zero-cost-compare.mjs +++ b/scripts/zero-cost-compare.mjs @@ -418,6 +418,7 @@ export const DYNAMIC_WORKLOADS = new Set([ "bench/typed-array-scan-workers", "bench/log-summary", "bench/inventory-report", + "bench/line-reader", "bench/json-unions", "bench/throw-chains", ]); diff --git a/tests/corpus/8344-readline-chunk-splitting.ts b/tests/corpus/8344-readline-chunk-splitting.ts new file mode 100644 index 000000000..85eff8cdc --- /dev/null +++ b/tests/corpus/8344-readline-chunk-splitting.ts @@ -0,0 +1,41 @@ +// Line splitting over stream chunks: an unfinished line joined across many +// chunks, a line longer than a chunk, empty lines, LF-only +// chunks next to chunks with CR and CRLF, an ASCII chunk after a chunk that +// ends inside a UTF-8 sequence, and text written with rl.write(). Each line +// event shows the line's length and its first and last characters. +import { createInterface } from "node:readline"; +import { Readable } from "node:stream"; + +function show(tag: string, line: string): void { + console.log(tag, line.length, JSON.stringify(line.slice(0, 12)), JSON.stringify(line.slice(-6))); +} + +function feed(tag: string, chunks: (string | Buffer)[]): Promise { + return new Promise((resolve) => { + const src = new Readable({ read() {} }); + const rl = createInterface({ input: src, crlfDelay: Infinity }); + rl.on("line", (line: string) => show(tag, line)); + rl.on("close", () => { + console.log(tag, "closed"); + resolve(); + }); + for (const chunk of chunks) src.push(typeof chunk === "string" ? Buffer.from(chunk, "utf8") : chunk); + src.push(null); + }); +} + +const long = "x".repeat(70000) + "y"; +await feed("joined", ["par", "tial ", "line", " continues", "\nnext\n", "tail"]); +await feed("long", [long.slice(0, 30000), long.slice(30000, 65536), long.slice(65536) + "\nshort\n"]); +await feed("empty", ["\n\n", "a\n\n", "\n", "b"]); +await feed("mixed", ["lf only\nstill lf\n", "cr\rcrlf\r\nlf\n", "end\r", "\nafter\r", "\r\n", "done"]); +const euro = Buffer.from("€uro\n", "utf8"); +await feed("utf8", [euro.subarray(0, 2), Buffer.concat([euro.subarray(2), Buffer.from("plain ascii\nmore")])]); +await feed("utf8-lines", ["café one\nnaïve two\n", "ascii three\n", "中文 four"]); + +const writer = createInterface({ input: new Readable({ read() {} }), crlfDelay: Infinity }); +writer.on("line", (line: string) => show("write", line)); +writer.write("first part "); +writer.write("second part\nthird\r\nfourth\rfifth"); +writer.write("\n"); +writer.close();