Keep small listener snapshots of readline and EventEmitter emits on the stack

readline's emit and EventEmitter's emit copy the listener list before
running it, so that listeners added or removed during the emit don't
change who runs. The copy was a malloc and a free per emitted event: per
line for a readline 'line' listener. A list of at most four listeners is
now copied into a stack buffer; longer lists still allocate.

A new corpus program pins the snapshot semantics for lists below and above
four listeners: a listener added during an emit runs from the next emit, a
removed one still runs in that emit, and a once listener runs once.

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/emit-listener-snapshots.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/emit-listener-snapshots.ts"
+ ],
+ "diags": []
+}

packages/runtime/src/scr_events_emitter.c

@@ -1066,7 +1066,9 @@ static bool scr_ee_emit_core(ScrEmitter *em, ScrStr *name, va_list *ap_template,
   ScrEeBucket *b = em->reg ? scr_ee_bucket_of(em->reg, name) : NULL;
   if (!b || b->n == 0) return false;
   size_t n = b->n;
-  ScrEeEntry **snap = malloc(n * sizeof *snap);
+  /* Most events have a listener or two: their snapshot stays on the stack. */
+  ScrEeEntry *small[4];
+  ScrEeEntry **snap = n <= 4 ? small : malloc(n * sizeof *snap);
   if (!snap) scr_ee_oom();
   for (size_t i = 0; i < n; i++) {
     snap[i] = b->ls[i];
@@ -1130,7 +1132,7 @@ static bool scr_ee_emit_core(ScrEmitter *em, ScrStr *name, va_list *ap_template,
     }
     scr_ee_entry_unref(e);
   }
-  free(snap);
+  if (snap != small) free(snap);
   return true;
 }
 

packages/runtime/src/scr_readline.c

@@ -105,7 +105,9 @@ void scr_rl_emit(ScrRl *rl, int ev, ScrStr *line, ScrError *err) {
   RlList *l = &rl->ev[ev];
   size_t n = l->n;
   if (n == 0) return;
-  RlEntry *snap = malloc(n * sizeof *snap);
+  /* Most events have a listener or two: their snapshot stays on the stack. */
+  RlEntry small[4];
+  RlEntry *snap = n <= 4 ? small : malloc(n * sizeof *snap);
   if (!snap) scr_rl_oom();
   memcpy(snap, l->v, n * sizeof *snap);
   for (size_t i = 0; i < n; i++) {
@@ -139,7 +141,7 @@ void scr_rl_emit(ScrRl *rl, int ev, ScrStr *line, ScrError *err) {
     }
     if (!e.iter) scr_closure_release(e.cb);
   }
-  free(snap);
+  if (snap != small) free(snap);
 }
 
 ScrArr *scr_rl_history_of(ScrRl *rl) {

tests/corpus/emit-listener-snapshots.ts

@@ -0,0 +1,46 @@
+// Listeners run from a snapshot of the list taken when the event is
+// emitted: a listener added during an emit runs from the next emit on, a
+// listener removed during an emit still runs in that emit, and a once
+// listener runs once. Covers readline 'line' events and EventEmitter
+// events with fewer and more listeners than a few.
+import { EventEmitter } from "node:events";
+import { createInterface } from "node:readline";
+import { Readable } from "node:stream";
+
+function emitter(count: number): void {
+  const em = new EventEmitter();
+  const log: string[] = [];
+  const listeners: ((n: number) => void)[] = [];
+  for (let i = 0; i < count; i++) {
+    const listener = (n: number): void => {
+      log.push(`${i}:${n}`);
+      if (n === 1 && i === 0) em.on("tick", (m: number) => log.push(`late:${m}`));
+      if (n === 1 && i === 1 && listeners[count - 1]) em.off("tick", listeners[count - 1]!);
+    };
+    listeners.push(listener);
+    em.on("tick", listener);
+  }
+  em.once("tick", (n: number) => log.push(`once:${n}`));
+  em.emit("tick", 1);
+  em.emit("tick", 2);
+  console.log(count, em.listenerCount("tick"), log.join(" "));
+}
+
+for (const count of [1, 2, 4, 5, 7]) emitter(count);
+
+async function lines(count: number): Promise<void> {
+  const input = Readable.from(["alpha\nbeta\n", "gamma\ndelta\n"]);
+  const rl = createInterface({ input, terminal: false });
+  const log: string[] = [];
+  for (let i = 0; i < count; i++) {
+    rl.on("line", (line: string) => {
+      log.push(`${i}:${line}`);
+      if (line === "beta" && i === 0) rl.on("line", (next: string) => log.push(`late:${next}`));
+    });
+  }
+  rl.once("line", (line: string) => log.push(`once:${line}`));
+  await new Promise<void>((resolve) => rl.on("close", () => resolve()));
+  console.log(count, log.join(" "));
+}
+
+for (const count of [1, 3, 6]) await lines(count);

tests/harness/corpus-costs/emit-listener-snapshots.ts.json

@@ -0,0 +1 @@
+{ "plain": 5463, "san": 3430 }