Finish a worker message batch before port messages, and count queued messages as they arrive

- Port messages that a 'message' listener posts while more worker
  messages are queued are delivered after those messages and before
  setImmediate callbacks, as in Node. The ports' poll phase waits while a
  batch's next message is queued, and the loop pass continues once more
  after the batch when a port has something to deliver.
- A mailbox counts its queued events per kind as they are posted and
  taken, so starting a batch reads the count instead of scanning the
  queue under the lock. Messages that alternate with a worker's captured
  stdout or stderr start a batch each, which made delivery quadratic:
  20000 such messages took 5.6 s to drain, and now take 0.5 s (Node 0.57 s).
- The continue hook passes on that the batch's next message is at the
  head, so the dispatch that follows does not lock the mailbox again.
- Corpus 6212 checks the order of worker messages, a MessageChannel
  message posted by the first listener, and setImmediate against Node.
  6211's costs are now taken from its CI timings.

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6212-worker-message-batch-ports.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/6212-worker-message-batch-ports.ts"
+ ],
+ "diags": []
+}

packages/runtime/src/scr_mailbox.c

@@ -25,11 +25,15 @@
 #endif
 #endif
 
+/* Event kinds below this have their queued events counted (scr_mailbox_count_kind). */
+#define SCR_MAILBOX_COUNTED_KINDS 16u
+
 struct ScrMailbox {
   atomic_size_t references;
   ScrMailEvent *head;
   ScrMailEvent **tail;
   bool closed;
+  size_t queued[SCR_MAILBOX_COUNTED_KINDS]; /* under the lock */
   /* head != NULL || closed, written under the lock, read without it by
    * scr_mailbox_wait's spin. */
   atomic_bool ready;
@@ -167,6 +171,7 @@ bool scr_mailbox_post(ScrMailbox *mailbox, ScrMailEvent *event) {
   event->next = NULL;
   *mailbox->tail = event;
   mailbox->tail = &event->next;
+  if (event->kind < SCR_MAILBOX_COUNTED_KINDS) mailbox->queued[event->kind]++;
   atomic_store_explicit(&mailbox->ready, true, memory_order_release);
   scr_mailbox_unlock(mailbox);
   /* Signaled after the unlock: a woken or spinning taker then finds the
@@ -182,6 +187,7 @@ ScrMailEvent *scr_mailbox_take(ScrMailbox *mailbox) {
   if (event) {
     mailbox->head = event->next;
     event->next = NULL;
+    if (event->kind < SCR_MAILBOX_COUNTED_KINDS) mailbox->queued[event->kind]--;
   }
   if (!mailbox->head) {
     mailbox->tail = &mailbox->head;
@@ -201,6 +207,7 @@ ScrMailEvent *scr_mailbox_take_kind(ScrMailbox *mailbox, unsigned kind, uint64_t
     *slot = event->next;
     if (mailbox->tail == &event->next) mailbox->tail = slot;
     event->next = NULL;
+    if (event->kind < SCR_MAILBOX_COUNTED_KINDS) mailbox->queued[event->kind]--;
   }
   if (!mailbox->head) {
     mailbox->tail = &mailbox->head;
@@ -235,8 +242,7 @@ unsigned scr_mailbox_head_kind(ScrMailbox *mailbox) {
 
 size_t scr_mailbox_count_kind(ScrMailbox *mailbox, unsigned kind) {
   scr_mailbox_lock(mailbox);
-  size_t count = 0;
-  for (ScrMailEvent *event = mailbox->head; event; event = event->next) count += event->kind == kind;
+  size_t count = kind < SCR_MAILBOX_COUNTED_KINDS ? mailbox->queued[kind] : 0;
   scr_mailbox_unlock(mailbox);
   return count;
 }

packages/runtime/src/scr_mailbox.h

@@ -28,7 +28,8 @@ void scr_mailbox_close(ScrMailbox *mailbox);
 bool scr_mailbox_pending(ScrMailbox *mailbox);
 /* The kind of the first queued event, or 0 when none is queued. */
 unsigned scr_mailbox_head_kind(ScrMailbox *mailbox);
-/* The number of queued events of this kind. */
+/* The number of queued events of this kind, kept as events are posted and
+ * taken; 0 for kinds of 16 and above. */
 size_t scr_mailbox_count_kind(ScrMailbox *mailbox, unsigned kind);
 void scr_mailbox_wait(ScrMailbox *mailbox, double milliseconds);
 int scr_mailbox_pollfd(const ScrMailbox *mailbox);

packages/runtime/src/scr_worker_events.c

@@ -63,9 +63,22 @@ static SCR_TL bool scr_channel_closing_phase;
  * has taken SCR_WORKER_BATCH_MIN, with a microtask checkpoint after each,
  * before the loop goes on to timers and immediates. A delivered message
  * starts a batch; the loop pass continues it (scr_workers_more) while the
- * next event is a message and the batch has room. */
+ * next event is a message and the batch has room. Port messages that a
+ * listener posts meanwhile wait for the batch to end, and the pass then
+ * continues once more so they are delivered before immediates, as in Node. */
 #define SCR_WORKER_BATCH_MIN 1000
 static SCR_TL size_t scr_worker_batch_left;
+static SCR_TL bool scr_worker_batch_ended;
+/* scr_workers_more found the batch's next message at the head of the
+ * mailbox. Only this thread takes events and others append them, so it is
+ * still there when the pass dispatches. */
+static SCR_TL bool scr_worker_batch_queued;
+
+/* Whether a batch is open and its next message is queued. */
+static bool scr_worker_batch_next(void) {
+  return scr_worker_batch_left > 0 && scr_worker_events &&
+         scr_mailbox_head_kind(scr_worker_events) == SCR_WORKER_MESSAGE;
+}
 static const char *const scr_worker_method_names[] = {
   "ref", "unref", "hasRef", "terminate", "postMessage", "on", "once", "addListener",
   "prependListener", "prependOnceListener", "off", "removeListener", "removeAllListeners",
@@ -710,15 +723,29 @@ static bool scr_channel_more(void) {
 }
 
 static bool scr_workers_more(void) {
+  if (scr_worker_batch_left > 0) {
+    if (scr_worker_batch_next()) {
+      scr_worker_batch_queued = true;
+      return true;
+    }
+    scr_worker_batch_left = 0;
+    scr_worker_batch_ended = true;
+  }
   if (scr_channel_more()) return true;
-  if (scr_worker_batch_left == 0) return false;
-  if (scr_worker_events && scr_mailbox_head_kind(scr_worker_events) == SCR_WORKER_MESSAGE) return true;
-  scr_worker_batch_left = 0;
+  /* One more pass after a batch when a port has something to deliver: the
+   * ports' poll phase for what its listeners posted. */
+  bool ended = scr_worker_batch_ended;
+  scr_worker_batch_ended = false;
+  if (ended)
+    for (ScrWorkerView *view = scr_channel_ports; view; view = view->next)
+      if (scr_channel_port_ready(view)) return true;
   return false;
 }
 
 static void scr_worker_dispatch(void) {
-  if (scr_channel_dispatch()) return;
+  bool batch = scr_worker_batch_queued;
+  scr_worker_batch_queued = false;
+  if (!batch && scr_channel_dispatch()) return;
   ScrWorkerView *port = scr_worker_port;
   if (port && port->closing) {
     port->closing = false;

tests/corpus/6212-worker-message-batch-ports.ts

@@ -0,0 +1,47 @@
+import { MessageChannel, Worker, isMainThread, parentPort, workerData } from "node:worker_threads";
+
+// A MessageChannel message that a worker 'message' listener posts while
+// more worker messages are queued arrives after those messages and before
+// setImmediate callbacks, as in Node. The listener waits on a shared flag
+// until the worker has posted the rest.
+function waitFor(flag: Int32Array, index: number): void {
+  while (Atomics.load(flag, index) === 0) Atomics.wait(flag, index, 0, 50);
+}
+function signal(flag: Int32Array, index: number): void {
+  Atomics.store(flag, index, 1);
+  Atomics.notify(flag, index);
+}
+
+if (isMainThread) {
+  const flag = new Int32Array(new SharedArrayBuffer(8));
+  const worker = new Worker(new URL(import.meta.url), { workerData: flag });
+  const channel = new MessageChannel();
+  const log: string[] = [];
+  channel.port2.on("message", (m: unknown) => {
+    log.push(`channel ${String(m)}`);
+  });
+  let count = 0;
+  worker.on("message", (m: unknown) => {
+    count++;
+    log.push(`worker ${String(m)}`);
+    if (count === 1) {
+      signal(flag, 0);
+      waitFor(flag, 1);
+      channel.port1.postMessage("c1");
+      setImmediate(() => {
+        log.push("immediate");
+        setImmediate(() => {
+          log.push("second immediate");
+          for (const line of log) console.log(line);
+          channel.port1.close();
+        });
+      });
+    }
+  });
+} else {
+  const flag = workerData as Int32Array;
+  parentPort!.postMessage("w1");
+  waitFor(flag, 0);
+  for (let i = 2; i <= 6; i++) parentPort!.postMessage(`w${i}`);
+  signal(flag, 1);
+}

tests/harness/corpus-costs/6211-worker-message-batches.ts.json

@@ -1 +1 @@
-{ "plain": 2507, "san": 20224 }
+{ "plain": 3694, "san": 3271 }

tests/harness/corpus-costs/6212-worker-message-batch-ports.ts.json

@@ -0,0 +1 @@
+{ "plain": 5538, "san": 3201 }