diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6212-worker-message-batch-ports.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6212-worker-message-batch-ports.ts.json new file mode 100644 index 000000000..76d4c8da7 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6212-worker-message-batch-ports.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/6212-worker-message-batch-ports.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_mailbox.c b/packages/runtime/src/scr_mailbox.c index 39e7801ed..a8dc7e5cc 100644 --- a/packages/runtime/src/scr_mailbox.c +++ b/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; } diff --git a/packages/runtime/src/scr_mailbox.h b/packages/runtime/src/scr_mailbox.h index ac8493057..4e5f253b5 100644 --- a/packages/runtime/src/scr_mailbox.h +++ b/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); diff --git a/packages/runtime/src/scr_worker_events.c b/packages/runtime/src/scr_worker_events.c index f7778f3fa..252230372 100644 --- a/packages/runtime/src/scr_worker_events.c +++ b/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; diff --git a/tests/corpus/6212-worker-message-batch-ports.ts b/tests/corpus/6212-worker-message-batch-ports.ts new file mode 100644 index 000000000..3f5a2a476 --- /dev/null +++ b/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); +} diff --git a/tests/harness/corpus-costs/6211-worker-message-batches.ts.json b/tests/harness/corpus-costs/6211-worker-message-batches.ts.json index 75b17b215..af8e672d4 100644 --- a/tests/harness/corpus-costs/6211-worker-message-batches.ts.json +++ b/tests/harness/corpus-costs/6211-worker-message-batches.ts.json @@ -1 +1 @@ -{ "plain": 2507, "san": 20224 } +{ "plain": 3694, "san": 3271 } diff --git a/tests/harness/corpus-costs/6212-worker-message-batch-ports.ts.json b/tests/harness/corpus-costs/6212-worker-message-batch-ports.ts.json new file mode 100644 index 000000000..fb0e48437 --- /dev/null +++ b/tests/harness/corpus-costs/6212-worker-message-batch-ports.ts.json @@ -0,0 +1 @@ +{ "plain": 5538, "san": 3201 }