Split readline input in one scan and copy ASCII strings into buffers whole (#2192)

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 |

benchmarks/runtime/workloads/line-reader.json

@@ -0,0 +1,7 @@
+{
+  "entry": "workloads/line-reader.ts",
+  "input": "line-reader",
+  "args": [
+    "{input}"
+  ]
+}

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<string, number>();
+const paths = new Map<string, number>();
+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)}`);
+});

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8344-readline-chunk-splitting.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/8344-readline-chunk-splitting.ts"
+ ],
+ "diags": []
+}

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) {

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);

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}`);

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",
 ]);

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<void> {
+  return new Promise<void>((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();