Merge pull request #1108 from vercel-labs/ctate/add-messagechannel-receivemessageonport-5d9f36

Add MessageChannel, receiveMessageOnPort and MessagePort transfer

packages/runtime/src/scr_async.c

@@ -3755,6 +3755,7 @@ void scr_loop_set_stream(bool (*pending)(void), void (*dispatch)(void)) {
 #ifdef SCR_WORKERS
 static SCR_TL bool (*scr_workers_pending_fn)(void);
 static SCR_TL void (*scr_workers_dispatch_fn)(void);
+static SCR_TL bool (*scr_workers_more_fn)(void);
 static SCR_TL int (*scr_workers_pollfd_fn)(void);
 static SCR_TL bool (*scr_workers_ready_fn)(void);
 static SCR_TL void (*scr_workers_wait_fn)(double);
@@ -3767,6 +3768,8 @@ void scr_loop_set_workers(bool (*pending)(void), bool (*ready)(void), void (*dis
   scr_workers_ready_fn = ready;
   scr_workers_wait_fn = wait;
 }
+
+void scr_loop_set_workers_more(bool (*more)(void)) { scr_workers_more_fn = more; }
 #endif
 
 static bool scr_workers_pending(void) {
@@ -3925,6 +3928,9 @@ static bool scr_loop_run_pass(ScrPromise *top_level, bool first_checkpoint) {
     scr_workers_dispatch();
     if (scr_exc_pending()) return false;
     if (scr_ready_len > 0 || scr_nt_head != NULL) continue;
+#ifdef SCR_WORKERS
+    if (scr_workers_more_fn != NULL && scr_workers_more_fn()) continue;
+#endif
     /* Event dispatch (scr_events.c, when linked): watched signals
      * delivered since the last turn fire their listeners now (macrotasks,
      * like timers), then stdin — while a consumer exists, probe fd 0 and

packages/runtime/src/scr_bytes.h

@@ -90,6 +90,10 @@ void scr_bytes_require_unshared(const ScrBytes *);
 ScrDyn *scr_shared_array_buffer_new(ScrDyn *);
 ScrDyn *scr_worker_data(void);
 ScrDyn *scr_worker_parent_port(void);
+/* new MessageChannel() ({ port1, port2 }) and receiveMessageOnPort(port)
+ * (scr_worker_events.c). */
+ScrDyn *scr_worker_message_channel(void);
+ScrDyn *scr_worker_receive_message_on_port(ScrDyn *port);
 ScrDyn *scr_worker_new(double root, ScrStr *filename, ScrDyn *options);
 bool scr_shared_array_buffer_is(const ScrDyn *);
 bool scr_buffer_storage_is(const ScrDyn *);

packages/runtime/src/scr_mailbox.c

@@ -231,3 +231,192 @@ int scr_mailbox_pollfd(const ScrMailbox *mailbox) {
   return mailbox->read_fd;
 #endif
 }
+
+/* ── MessageChannel ──────────────────────────────────────────────────── */
+
+typedef struct {
+  ScrMailEvent *head; /* incoming events for this end, in order */
+  ScrMailEvent **tail;
+  size_t count;
+  ScrMailbox *owner; /* the owning thread's inbox (+1), or NULL in transit */
+  bool closed;       /* this end called close() */
+  bool peer_closed;  /* the other end's close reached this end */
+  uint64_t peer_closed_seq; /* when: notices are delivered in this order */
+} ScrChannelEnd;
+
+static atomic_uint_fast64_t scr_channel_notice_seq;
+
+struct ScrChannel {
+  atomic_size_t references;
+  ScrChannelEnd end[2];
+#ifdef _WIN32
+  SRWLOCK lock;
+#else
+  pthread_mutex_t lock;
+#endif
+};
+
+static void scr_channel_lock(ScrChannel *channel) {
+#ifdef _WIN32
+  AcquireSRWLockExclusive(&channel->lock);
+#else
+  (void)pthread_mutex_lock(&channel->lock);
+#endif
+}
+
+static void scr_channel_unlock(ScrChannel *channel) {
+#ifdef _WIN32
+  ReleaseSRWLockExclusive(&channel->lock);
+#else
+  (void)pthread_mutex_unlock(&channel->lock);
+#endif
+}
+
+ScrChannel *scr_channel_new(void) {
+  ScrChannel *channel = calloc(1, sizeof(*channel));
+  if (!channel) scr_trap("scriptc: out of memory\n");
+  atomic_init(&channel->references, 1);
+  for (int side = 0; side < 2; side++) channel->end[side].tail = &channel->end[side].head;
+#ifdef _WIN32
+  InitializeSRWLock(&channel->lock);
+#else
+  if (pthread_mutex_init(&channel->lock, NULL) != 0) scr_trap("scriptc: out of memory\n");
+#endif
+  return channel;
+}
+
+ScrChannel *scr_channel_retain(ScrChannel *channel) {
+  atomic_fetch_add_explicit(&channel->references, 1, memory_order_relaxed);
+  return channel;
+}
+
+void scr_channel_release(ScrChannel *channel) {
+  if (!channel || atomic_fetch_sub_explicit(&channel->references, 1, memory_order_acq_rel) != 1)
+    return;
+  for (int side = 0; side < 2; side++) {
+    ScrChannelEnd *end = &channel->end[side];
+    while (end->head) {
+      ScrMailEvent *event = end->head;
+      end->head = event->next;
+      scr_mail_event_free(event);
+    }
+    scr_mailbox_release(end->owner);
+  }
+#ifndef _WIN32
+  pthread_mutex_destroy(&channel->lock);
+#endif
+  free(channel);
+}
+
+/* Wakes the owner's loop with a wake_kind event. */
+static void scr_channel_wake(ScrMailbox *owner, unsigned wake_kind) {
+  if (!owner) return;
+  ScrMailEvent *wake = calloc(1, sizeof(*wake));
+  if (!wake) scr_trap("scriptc: out of memory\n");
+  wake->kind = wake_kind;
+  scr_mailbox_post(owner, wake);
+  scr_mailbox_release(owner);
+}
+
+void scr_channel_set_owner(ScrChannel *channel, int side, ScrMailbox *owner, unsigned wake_kind) {
+  scr_channel_lock(channel);
+  ScrChannelEnd *end = &channel->end[side];
+  ScrMailbox *old = end->owner;
+  end->owner = owner ? scr_mailbox_retain(owner) : NULL;
+  ScrMailbox *wake = end->owner && (end->head || end->peer_closed) ? scr_mailbox_retain(end->owner) : NULL;
+  scr_channel_unlock(channel);
+  scr_mailbox_release(old);
+  scr_channel_wake(wake, wake_kind);
+}
+
+/* postMessage from `side`: the message (consumed) queues on the other end
+ * unless either end has closed. */
+bool scr_channel_post(ScrChannel *channel, int side, ScrMessage *message, unsigned message_kind,
+                      unsigned wake_kind) {
+  ScrMailEvent *event = calloc(1, sizeof(*event));
+  if (!event) scr_trap("scriptc: out of memory\n");
+  event->kind = message_kind;
+  event->message = message;
+  scr_channel_lock(channel);
+  ScrChannelEnd *peer = &channel->end[1 - side];
+  if (channel->end[side].closed || peer->closed) {
+    scr_channel_unlock(channel);
+    scr_mail_event_free(event);
+    return false;
+  }
+  *peer->tail = event;
+  peer->tail = &event->next;
+  peer->count++;
+  ScrMailbox *wake = peer->owner ? scr_mailbox_retain(peer->owner) : NULL;
+  scr_channel_unlock(channel);
+  scr_channel_wake(wake, wake_kind);
+  return true;
+}
+
+/* close() on `side`: nothing more is posted in either direction. */
+void scr_channel_close(ScrChannel *channel, int side) {
+  scr_channel_lock(channel);
+  channel->end[side].closed = true;
+  scr_channel_unlock(channel);
+}
+
+/* The closed port's 'close' ran: the other end learns it, after its queued
+ * messages. */
+void scr_channel_notify_close(ScrChannel *channel, int side, unsigned wake_kind) {
+  scr_channel_lock(channel);
+  ScrChannelEnd *peer = &channel->end[1 - side];
+  ScrMailbox *wake = NULL;
+  if (!peer->peer_closed && !peer->closed) {
+    peer->peer_closed = true;
+    peer->peer_closed_seq = atomic_fetch_add(&scr_channel_notice_seq, 1) + 1;
+    wake = peer->owner ? scr_mailbox_retain(peer->owner) : NULL;
+  }
+  scr_channel_unlock(channel);
+  scr_channel_wake(wake, wake_kind);
+}
+
+/* The next queued message for `side`, or NULL. */
+ScrMailEvent *scr_channel_take(ScrChannel *channel, int side) {
+  scr_channel_lock(channel);
+  ScrChannelEnd *end = &channel->end[side];
+  ScrMailEvent *event = end->head;
+  if (event) {
+    end->head = event->next;
+    if (!end->head) end->tail = &end->head;
+    end->count--;
+    event->next = NULL;
+  }
+  scr_channel_unlock(channel);
+  return event;
+}
+
+size_t scr_channel_count(ScrChannel *channel, int side) {
+  scr_channel_lock(channel);
+  size_t count = channel->end[side].count;
+  scr_channel_unlock(channel);
+  return count;
+}
+
+/* The other end's close notice order (0: none). */
+uint64_t scr_channel_peer_closed(ScrChannel *channel, int side) {
+  scr_channel_lock(channel);
+  uint64_t seq = channel->end[side].peer_closed ? channel->end[side].peer_closed_seq : 0;
+  scr_channel_unlock(channel);
+  return seq;
+}
+
+/* Drops the queued messages for `side` (its port closed). */
+void scr_channel_drop(ScrChannel *channel, int side) {
+  scr_channel_lock(channel);
+  ScrChannelEnd *end = &channel->end[side];
+  ScrMailEvent *head = end->head;
+  end->head = NULL;
+  end->tail = &end->head;
+  end->count = 0;
+  scr_channel_unlock(channel);
+  while (head) {
+    ScrMailEvent *event = head;
+    head = event->next;
+    scr_mail_event_free(event);
+  }
+}

packages/runtime/src/scr_mailbox.h

@@ -25,4 +25,25 @@ bool scr_mailbox_pending(ScrMailbox *mailbox);
 void scr_mailbox_wait(ScrMailbox *mailbox, double milliseconds);
 int scr_mailbox_pollfd(const ScrMailbox *mailbox);
 
+/* A MessageChannel: two ends, each with its queue of incoming messages and
+ * the inbox of the thread that owns its port (NULL while the port is in
+ * transit). Posting and closing wake the receiving owner with a wake_kind
+ * event. Ends are thread-neutral, so a port can move between threads. */
+typedef struct ScrChannel ScrChannel;
+ScrChannel *scr_channel_new(void);
+ScrChannel *scr_channel_retain(ScrChannel *channel);
+void scr_channel_release(ScrChannel *channel);
+void scr_channel_set_owner(ScrChannel *channel, int side, ScrMailbox *owner, unsigned wake_kind);
+/* Consumes message; false when either end has closed (the message is dropped). */
+bool scr_channel_post(ScrChannel *channel, int side, ScrMessage *message, unsigned message_kind,
+                      unsigned wake_kind);
+/* close() stops posting both ways; notify_close, once the closed port's
+ * 'close' ran, tells the other end (after its queued messages). */
+void scr_channel_close(ScrChannel *channel, int side);
+void scr_channel_notify_close(ScrChannel *channel, int side, unsigned wake_kind);
+ScrMailEvent *scr_channel_take(ScrChannel *channel, int side);
+size_t scr_channel_count(ScrChannel *channel, int side);
+uint64_t scr_channel_peer_closed(ScrChannel *channel, int side); /* the notice's order, 0: none */
+void scr_channel_drop(ScrChannel *channel, int side);
+
 #endif

packages/runtime/src/scr_message.c

@@ -21,8 +21,12 @@ typedef enum {
   /* A typed value published with @scriptc/threads (immortal), delivered as
    * a fresh typed-ref capsule over the same pointer. */
   SCR_MESSAGE_TYPED_REF,
+  /* A transferred MessagePort: its channel end (+1), the side in number. */
+  SCR_MESSAGE_PORT,
 } ScrMessageKind;
 
+const ScrMessagePortOps *scr_message_port_ops = NULL;
+
 typedef struct {
   char *key;
   size_t length;
@@ -43,6 +47,7 @@ typedef struct {
   ScrBytesElem element;
   bool data_view;
   ScrDyn *published; /* SCR_MESSAGE_PUBLISHED: immortal root, owns nothing */
+  ScrChannel *channel; /* SCR_MESSAGE_PORT */
   /* SCR_MESSAGE_TYPED_REF: the capsule's process-wide adapters and type key
    * (vtables, adapters and type keys are shared by every thread). */
   struct {
@@ -85,8 +90,18 @@ typedef struct {
   size_t *stack;
   size_t depth;
   size_t stack_capacity;
+  const ScrDyn *transfer; /* the transfer list (a dyn array), or NULL */
 } ScrMessageEncoder;
 
+static bool scr_message_listed(const ScrMessageEncoder *encoder, const ScrDyn *value) {
+  if (!encoder->transfer) return false;
+  for (size_t i = 0; i < encoder->transfer->v.arr.len; i++) {
+    const ScrDyn *item = encoder->transfer->v.arr.items[i];
+    if (item && item->kind == SCR_DYN_HANDLE && item->v.handle.ptr == value->v.handle.ptr) return true;
+  }
+  return false;
+}
+
 static void *scr_message_alloc(size_t count, size_t size) {
   if (count > SIZE_MAX / size) scr_trap("scriptc: message too large\n");
   void *data = calloc(count ? count : 1, size);
@@ -303,6 +318,20 @@ static size_t scr_message_value(ScrMessageEncoder *encoder, const ScrDyn *input)
     break;
   case SCR_DYN_BYTES:
   case SCR_DYN_HANDLE: {
+    if (value->kind == SCR_DYN_HANDLE && scr_message_port_ops && scr_message_port_ops->is_port(value)) {
+      int side = 0;
+      ScrChannel *channel = scr_message_listed(encoder, value)
+          ? scr_message_port_ops->channel_of(value, &side) : NULL;
+      if (!channel) {
+        scr_throw_domex("DataCloneError",
+                        "Object that needs transfer was found in message but not listed in transferList");
+        return SIZE_MAX;
+      }
+      node->kind = SCR_MESSAGE_PORT;
+      node->channel = channel;
+      node->number = side;
+      break;
+    }
     bool buffer = scr_buffer_storage_is(value);
     if (value->kind == SCR_DYN_HANDLE && !buffer) goto unsupported;
     const ScrBytes *bytes = buffer ? value->v.handle.ptr : value->v.bytes;
@@ -491,11 +520,39 @@ static ScrMessage *scr_message_try_publish(const ScrDyn *value) {
   return message;
 }
 
-ScrMessage *scr_message_encode(const ScrDyn *value) {
-  ScrMessage *published = scr_message_try_publish(value);
+static ScrMessage *scr_message_encode_with(const ScrDyn *value, const ScrDyn *transfer);
+
+ScrMessage *scr_message_encode(const ScrDyn *value) { return scr_message_encode_with(value, NULL); }
+
+ScrMessage *scr_message_encode_transfer(const ScrDyn *value, const ScrDyn *transfer) {
+  if (!transfer || transfer->kind != SCR_DYN_ARR || transfer->v.arr.len == 0)
+    return scr_message_encode_with(value, NULL);
+  for (size_t i = 0; i < transfer->v.arr.len; i++) {
+    const ScrDyn *item = transfer->v.arr.items[i];
+    int side = 0;
+    ScrChannel *channel = item && item->kind == SCR_DYN_HANDLE && scr_message_port_ops
+        ? scr_message_port_ops->channel_of(item, &side) : NULL;
+    if (!channel) {
+      static const char text[] = "Transfer lists with values other than MessagePorts are not supported yet";
+      ScrStr *m = scr_str_new(text, sizeof text - 1);
+      scr_throw_lowering_fence(m);
+      scr_str_release(m);
+      return NULL;
+    }
+    scr_message_port_ops->release(channel);
+  }
+  ScrMessage *message = scr_message_encode_with(value, transfer);
+  if (!message) return NULL;
+  /* the listed ports now belong to the message */
+  for (size_t i = 0; i < transfer->v.arr.len; i++) scr_message_port_ops->detach(transfer->v.arr.items[i]);
+  return message;
+}
+
+static ScrMessage *scr_message_encode_with(const ScrDyn *value, const ScrDyn *transfer) {
+  ScrMessage *published = transfer ? NULL : scr_message_try_publish(value);
   if (published) return published;
   ScrMessage *message = scr_message_alloc(1, sizeof(*message));
-  ScrMessageEncoder encoder = {.message = message};
+  ScrMessageEncoder encoder = {.message = message, .transfer = transfer};
   if (scr_message_value(&encoder, value) == SIZE_MAX) goto done;
   while (encoder.depth && !scr_exc_pending()) {
     size_t id = encoder.stack[encoder.depth - 1];
@@ -630,6 +687,9 @@ ScrDyn *scr_message_decode(const ScrMessage *message) {
           node->typed.ptr, node->typed.retain, node->typed.release, node->typed.type_key,
           node->typed.type_key_len, node->typed.materialize, node->typed.commit);
       break;
+    case SCR_MESSAGE_PORT:
+      values[i] = scr_message_port_ops->adopt(node->channel, (int)node->number);
+      break;
     }
     if (!values[i] || scr_exc_pending()) goto done;
   }
@@ -676,6 +736,7 @@ void scr_message_free(ScrMessage *message) {
     free(node->properties);
     free(node->data);
     scr_shared_release(node->shared);
+    if (node->kind == SCR_MESSAGE_PORT) scr_message_port_ops->release(node->channel);
   }
   free(message->nodes);
   free(message);

packages/runtime/src/scr_message.h

@@ -8,6 +8,24 @@
  * values; release the message separately after delivery or cancellation. */
 typedef struct ScrMessage ScrMessage;
 ScrMessage *scr_message_encode(const ScrDyn *value);
+/* postMessage(value, transferList): the listed MessagePorts move with the
+ * message (a port in the value that is not listed throws DataCloneError);
+ * transfer is NULL or a dyn array. */
+ScrMessage *scr_message_encode_transfer(const ScrDyn *value, const ScrDyn *transfer);
+
+/* MessagePort transfer, installed by the worker events unit: channel_of
+ * answers a channel port's end (+1 channel; NULL for other values, side
+ * set), is_port any MessagePort, detach leaves the sender's port unusable,
+ * adopt makes a port in the decoding thread (the channel borrowed). */
+typedef struct ScrChannel ScrChannel;
+typedef struct {
+  ScrChannel *(*channel_of)(const ScrDyn *value, int *side);
+  bool (*is_port)(const ScrDyn *value);
+  void (*detach)(const ScrDyn *value);
+  ScrDyn *(*adopt)(ScrChannel *channel, int side);
+  void (*release)(ScrChannel *channel);
+} ScrMessagePortOps;
+extern const ScrMessagePortOps *scr_message_port_ops;
 ScrDyn *scr_message_decode(const ScrMessage *message);
 void scr_message_free(ScrMessage *message);
 

packages/runtime/src/scr_worker.h

@@ -38,5 +38,8 @@ void scr_worker_port_close(void);
 ScrMailbox *scr_worker_inbox(void);
 void scr_loop_set_workers(bool (*pending)(void), bool (*ready)(void), void (*dispatch)(void),
                           int (*pollfd)(void), void (*wait)(double));
+/* After a dispatch: true keeps the loop pass delivering (a MessagePort poll
+ * phase in progress), with microtasks drained between events. */
+void scr_loop_set_workers_more(bool (*more)(void));
 
 #endif

packages/runtime/src/scr_worker_events.c

@@ -17,12 +17,30 @@ typedef struct ScrWorkerView {
   bool closing;
   ScrMailEvent *queued_head;
   ScrMailEvent *queued_tail;
+  /* A MessageChannel port: its channel (+1) and end; NULL for parentPort. */
+  ScrChannel *channel;
+  int side;
+  bool detached; /* transferred to another thread */
+  uint64_t close_seq; /* when close() ran: the latest close emits first */
+  uint64_t close_gen; /* the poll generation close() ran in */
+  size_t phase_left; /* messages this poll phase still delivers */
+  bool phase_close;  /* the other end's close, delivered this phase */
   struct ScrWorkerView *next;
 } ScrWorkerView;
 
 static SCR_TL ScrMailbox *scr_worker_events;
 static SCR_TL ScrWorkerView *scr_worker_views;
 static SCR_TL ScrWorkerView *scr_worker_port;
+/* this thread's MessageChannel ports (registry references) */
+static SCR_TL ScrWorkerView *scr_channel_ports;
+static SCR_TL uint64_t scr_channel_close_seq;
+static SCR_TL bool scr_channel_phase; /* a poll phase is being delivered */
+/* completed poll phases: a port's 'close' runs in the close phase after
+ * the poll phase its close() came before or during */
+static SCR_TL uint64_t scr_channel_poll_gen;
+/* the loop pass continues a phase (scr_channel_more answered true) */
+static SCR_TL bool scr_channel_continuing;
+static SCR_TL bool scr_channel_closing_phase;
 static const char *const scr_worker_method_names[] = {
   "ref", "unref", "hasRef", "terminate", "postMessage", "on", "once", "addListener",
   "prependListener", "prependOnceListener", "off", "removeListener", "removeAllListeners",
@@ -38,6 +56,7 @@ static void scr_worker_view_trace(void *ptr, ScrTraceVisit visit, void *ctx) {
 static void scr_worker_view_gcfree(void *ptr) {
   ScrWorkerView *view = ptr;
   scr_worker_thread_release(view->thread);
+  scr_channel_release(view->channel);
   while (view->queued_head) {
     ScrMailEvent *event = view->queued_head;
     view->queued_head = event->next;
@@ -106,6 +125,44 @@ static bool scr_worker_empty_transfer(const ScrDyn *value) {
   return value->kind == SCR_DYN_UNDEF || (value->kind == SCR_DYN_ARR && value->v.arr.len == 0);
 }
 
+/* postMessage's transfer argument: an array, or { transfer } (+1, or NULL
+ * when empty) */
+static ScrDyn *scr_worker_transfer_list(ScrDyn *arg) {
+  ScrDyn *value = arg;
+  ScrDyn *own = NULL;
+  if (value->kind == SCR_DYN_TYPED_REF) {
+    own = scr_dyn_typed_ref_materialize(value);
+    if (!own || scr_exc_pending()) {
+      scr_dyn_release(own);
+      return NULL;
+    }
+    value = own;
+  }
+  if (value->kind == SCR_DYN_OBJ) {
+    ScrDyn *list = scr_dyn_obj_get(value, "transfer", 8);
+    ScrDyn *result = list ? scr_worker_transfer_list(list) : NULL;
+    scr_dyn_release(own);
+    return result;
+  }
+  if (value->kind == SCR_DYN_ARR && value->v.arr.len > 0) {
+    ScrDyn *result = scr_dyn_retain(value);
+    scr_dyn_release(own);
+    return result;
+  }
+  bool empty = scr_worker_empty_transfer(value);
+  scr_dyn_release(own);
+  if (!empty) scr_worker_refusal("This postMessage transfer argument is not supported yet");
+  return NULL;
+}
+
+static bool scr_worker_transfer_has(const ScrDyn *transfer, const ScrWorkerView *view) {
+  for (size_t i = 0; i < transfer->v.arr.len; i++) {
+    const ScrDyn *item = transfer->v.arr.items[i];
+    if (item && item->kind == SCR_DYN_HANDLE && item->v.handle.ptr == view) return true;
+  }
+  return false;
+}
+
 static ScrDyn *scr_worker_view_invoke(void *ptr, ScrDyn *self, const char *method,
                                      ScrDyn *const *args, size_t argc, const char *what) {
   (void)what;
@@ -115,7 +172,7 @@ static ScrDyn *scr_worker_view_invoke(void *ptr, ScrDyn *self, const char *metho
     return scr_dyn_retain(scr_dyn_undefined());
   }
   if (view->port && !strcmp(method, "hasRef"))
-    return scr_dyn_new_bool(view->referenced && !view->closed);
+    return scr_dyn_new_bool(view->referenced && (!view->closed || (view->channel && view->closing)));
   if (!view->port && !strcmp(method, "terminate")) {
     if (!view->termination) {
       view->termination = scr_promise_new();
@@ -129,16 +186,30 @@ static ScrDyn *scr_worker_view_invoke(void *ptr, ScrDyn *self, const char *metho
     /* Worker drops messages after exit before validating or cloning. A
      * closed MessagePort still validates its payload, as Node does. */
     if (!view->port && view->closed) return scr_dyn_retain(scr_dyn_undefined());
+    if (view->detached) return scr_dyn_retain(scr_dyn_undefined());
     if (!argc) {
-      static const char message[] = "The \"value\" argument must be specified";
-      scr_throw_error_msg_code(SCR_ERR_TYPE, message, sizeof message - 1, "ERR_MISSING_ARGS");
+      if (view->channel) {
+        static const char message[] = "Not enough arguments to MessagePort.postMessage";
+        scr_throw_error_msg_code(SCR_ERR_TYPE, message, sizeof message - 1, "ERR_MISSING_ARGS");
+      } else {
+        static const char message[] = "The \"value\" argument must be specified";
+        scr_throw_error_msg_code(SCR_ERR_TYPE, message, sizeof message - 1, "ERR_MISSING_ARGS");
+      }
       return NULL;
     }
-    if (argc > 1 && !scr_worker_empty_transfer(args[1]))
-      return scr_exc_pending() ? NULL : scr_worker_refusal("Worker postMessage transfer lists are not supported yet");
-    ScrMessage *message = scr_message_encode(args[0]);
+    ScrDyn *transfer = argc > 1 ? scr_worker_transfer_list(args[1]) : NULL;
+    if (scr_exc_pending()) return NULL;
+    if (transfer && scr_worker_transfer_has(transfer, view)) {
+      scr_dyn_release(transfer);
+      scr_throw_domex("DataCloneError", "Transfer list contains source port");
+      return NULL;
+    }
+    ScrMessage *message = scr_message_encode_transfer(args[0], transfer);
+    scr_dyn_release(transfer);
     if (!message) return NULL;
     if (view->closed) scr_message_free(message);
+    else if (view->channel)
+      scr_channel_post(view->channel, view->side, message, SCR_WORKER_MESSAGE, SCR_WORKER_WAKE);
     else if (view->port) scr_worker_post_parent(message);
     else scr_worker_thread_post(view->thread, message);
     return scr_dyn_retain(scr_dyn_undefined());
@@ -185,9 +256,17 @@ static ScrDyn *scr_worker_view_invoke(void *ptr, ScrDyn *self, const char *metho
       scr_emitter_release(scr_emitter_on_flex(view->events, name, args[0], true, false));
       scr_str_release(name);
     }
-    if (!view->closed) view->closing = true;
-    view->closed = true;
-    scr_worker_port_close();
+    if (!view->closed && !view->detached) {
+      view->closing = true;
+      view->closed = true;
+      if (view->channel) {
+        view->close_seq = ++scr_channel_close_seq;
+        view->close_gen = scr_channel_poll_gen;
+        scr_channel_close(view->channel, view->side);
+      } else {
+        scr_worker_port_close();
+      }
+    }
     return scr_dyn_retain(scr_dyn_undefined());
   }
   if (view->port && !strcmp(method, "start")) {
@@ -235,6 +314,9 @@ static ScrDyn *scr_worker_view_get(void *ptr, const char *key, size_t length) {
 static bool scr_workers_pending(void) {
   for (ScrWorkerView *view = scr_worker_views; view; view = view->next)
     if (view->referenced) return true;
+  for (ScrWorkerView *view = scr_channel_ports; view; view = view->next)
+    if (view->closing || (view->started && view->referenced && !view->closed && !view->detached))
+      return true;
   return scr_worker_port && (scr_worker_port->closing ||
     (scr_worker_port->referenced && !scr_worker_port->closed));
 }
@@ -243,7 +325,19 @@ static int scr_worker_events_pollfd(void) {
   return scr_worker_events ? scr_mailbox_pollfd(scr_worker_events) : -1;
 }
 
+/* a channel port with something to deliver: its 'close', the other end's
+ * close, or messages once started */
+static bool scr_channel_port_ready(ScrWorkerView *view) {
+  if (view->closing || view->phase_left || view->phase_close) return true;
+  if (view->closed || view->detached) return false;
+  if (view->started && scr_channel_count(view->channel, view->side) > 0) return true;
+  return scr_channel_peer_closed(view->channel, view->side) &&
+         scr_channel_count(view->channel, view->side) == 0;
+}
+
 static bool scr_worker_events_ready(void) {
+  for (ScrWorkerView *view = scr_channel_ports; view; view = view->next)
+    if (scr_channel_port_ready(view)) return true;
   return (scr_worker_port && (scr_worker_port->closing ||
     (scr_worker_port->started && scr_worker_port->queued_head))) ||
     (scr_worker_events && scr_mailbox_pending(scr_worker_events));
@@ -253,7 +347,123 @@ static void scr_worker_events_wait(double milliseconds) {
   if (!scr_worker_events_ready()) scr_mailbox_wait(scr_worker_events, milliseconds);
 }
 
+static void scr_worker_port_emit(ScrWorkerView *view, const char *event, ScrDyn *argument) {
+  ScrStr *name = scr_str_new(event, strlen(event));
+  scr_dyn_this_push(view, SCR_DYNH_MESSAGE_PORT);
+  scr_emitter_emit_flex(view->events, name, argument ? &argument : NULL, argument ? 1 : 0);
+  scr_dyn_this_pop();
+  scr_str_release(name);
+}
+
+/* The channel ports' turn, one event per call, in Node's phase order:
+ * - the close phase: ports this thread closed, the latest close() first
+ *   (libuv's close callbacks), each telling its other end;
+ * - the poll phase: a snapshot of each port's queued messages (the batch
+ *   Node delivers), ports in creation order, then the other ends' close
+ *   notices in the order they were sent. scr_channel_more keeps the loop
+ *   pass going until the phase is delivered, with microtasks between.
+ * False when nothing was ready. */
+static bool scr_channel_close_due(const ScrWorkerView *view) {
+  return view->closing && view->close_gen < scr_channel_poll_gen;
+}
+
+static bool scr_channel_dispatch(void) {
+  bool continuing = scr_channel_continuing;
+  scr_channel_continuing = false;
+  /* a loop pass runs one phase: the close phase, or the poll phase */
+  if (scr_channel_closing_phase && !continuing) scr_channel_closing_phase = false;
+  if (!scr_channel_phase && (!continuing || scr_channel_closing_phase)) {
+    ScrWorkerView *latest = NULL;
+    for (ScrWorkerView *view = scr_channel_ports; view; view = view->next)
+      if (scr_channel_close_due(view) && (!latest || view->close_seq > latest->close_seq))
+        latest = view;
+    if (latest) {
+      scr_worker_view_retain(latest);
+      latest->closing = false;
+      scr_channel_closing_phase = true;
+      scr_channel_drop(latest->channel, latest->side);
+      scr_channel_notify_close(latest->channel, latest->side, SCR_WORKER_WAKE);
+      scr_worker_port_emit(latest, "close", NULL);
+      scr_worker_view_release(latest);
+      return true;
+    }
+    if (scr_channel_closing_phase) return false;
+    bool any = false;
+    for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) {
+      if (view->closed || view->detached) continue;
+      size_t count = scr_channel_count(view->channel, view->side);
+      view->phase_left = view->started ? count : 0;
+      view->phase_close = count == 0 && scr_channel_peer_closed(view->channel, view->side) != 0;
+      any = any || view->phase_left || view->phase_close;
+    }
+    if (!any) {
+      scr_channel_poll_gen++;
+      return false;
+    }
+    scr_channel_phase = true;
+  }
+  for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) {
+    if (!view->phase_left) continue;
+    view->phase_left--;
+    ScrMailEvent *event = scr_channel_take(view->channel, view->side);
+    if (!event) {
+      view->phase_left = 0;
+      continue;
+    }
+    scr_worker_view_retain(view);
+    ScrDyn *argument = event->message ? scr_message_decode(event->message) : NULL;
+    if (argument) scr_worker_port_emit(view, "message", argument);
+    scr_dyn_release(argument);
+    scr_mail_event_free(event);
+    scr_worker_view_release(view);
+    return true;
+  }
+  ScrWorkerView *first = NULL;
+  uint64_t first_seq = 0;
+  for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) {
+    if (!view->phase_close) continue;
+    uint64_t seq = scr_channel_peer_closed(view->channel, view->side);
+    if (!first || seq < first_seq) {
+      first = view;
+      first_seq = seq;
+    }
+  }
+  if (!first) {
+    scr_channel_phase = false;
+    scr_channel_poll_gen++;
+    return false;
+  }
+  first->phase_close = false;
+  if (first->closed) return true;
+  scr_worker_view_retain(first);
+  first->closed = true;
+  scr_channel_drop(first->channel, first->side);
+  scr_worker_port_emit(first, "close", NULL);
+  scr_worker_view_release(first);
+  return true;
+}
+
+/* more of the current poll phase to deliver in this loop pass */
+static bool scr_channel_more(void) {
+  bool more = false;
+  if (scr_channel_closing_phase) {
+    /* the close phase runs its callbacks back to back */
+    for (ScrWorkerView *view = scr_channel_ports; view; view = view->next)
+      if (scr_channel_close_due(view)) more = true;
+  } else if (scr_channel_phase) {
+    for (ScrWorkerView *view = scr_channel_ports; view; view = view->next)
+      if (view->phase_left || view->phase_close) more = true;
+    if (!more) {
+      scr_channel_phase = false;
+      scr_channel_poll_gen++;
+    }
+  }
+  scr_channel_continuing = more;
+  return more;
+}
+
 static void scr_worker_dispatch(void) {
+  if (scr_channel_dispatch()) return;
   ScrWorkerView *port = scr_worker_port;
   if (port && port->closing) {
     port->closing = false;
@@ -319,6 +529,7 @@ static void scr_worker_dispatch(void) {
 
 static void scr_worker_events_cleanup(void) {
   scr_loop_set_workers(NULL, NULL, NULL, NULL, NULL);
+  scr_loop_set_workers_more(NULL);
   scr_mailbox_close(scr_worker_events);
   for (ScrWorkerView *view = scr_worker_views; view; view = view->next)
     scr_worker_thread_terminate(view->thread);
@@ -331,6 +542,13 @@ static void scr_worker_events_cleanup(void) {
   }
   scr_worker_view_release(scr_worker_port);
   scr_worker_port = NULL;
+  while (scr_channel_ports) {
+    ScrWorkerView *view = scr_channel_ports;
+    scr_channel_ports = view->next;
+    view->next = NULL;
+    if (view->channel) scr_channel_set_owner(view->channel, view->side, NULL, SCR_WORKER_WAKE);
+    scr_worker_view_release(view);
+  }
   for (size_t i = 0; i < sizeof scr_worker_methods / sizeof scr_worker_methods[0]; i++) {
     scr_dyn_release(scr_worker_methods[i]);
     scr_worker_methods[i] = NULL;
@@ -339,6 +557,8 @@ static void scr_worker_events_cleanup(void) {
   scr_worker_events = NULL;
 }
 
+static void scr_port_ops_install(void);
+
 static bool scr_worker_events_install(void) {
   if (scr_worker_events) return true;
   ScrMailbox *inbox = scr_worker_inbox();
@@ -355,12 +575,98 @@ static bool scr_worker_events_install(void) {
   };
   scr_dyn_handle_install(SCR_DYNH_WORKER, &worker_ops);
   scr_dyn_handle_install(SCR_DYNH_MESSAGE_PORT, &port_ops);
+  scr_port_ops_install();
   if (scr_context_atexit(scr_worker_events_cleanup) != 0) scr_trap("scriptc: out of memory\n");
   scr_loop_set_workers(scr_workers_pending, scr_worker_events_ready, scr_worker_dispatch,
     scr_worker_events_pollfd, scr_worker_events_wait);
+  scr_loop_set_workers_more(scr_channel_more);
   return true;
 }
 
+/* ── MessageChannel ports ─────────────────────────────────────────────── */
+
+static ScrWorkerView *scr_channel_port_new(ScrChannel *channel /*borrowed*/, int side) {
+  ScrWorkerView *view = scr_worker_view_new(true);
+  view->channel = scr_channel_retain(channel);
+  view->side = side;
+  /* the registry reference, in creation order (Node delivers ports in the
+   * order their handles were made) */
+  ScrWorkerView **tail = &scr_channel_ports;
+  while (*tail) tail = &(*tail)->next;
+  *tail = view;
+  scr_channel_set_owner(channel, side, scr_worker_events, SCR_WORKER_WAKE);
+  return view;
+}
+
+static ScrWorkerView *scr_port_view_of(const ScrDyn *value) {
+  if (!value || value->kind != SCR_DYN_HANDLE || value->v.handle.tag != SCR_DYNH_MESSAGE_PORT)
+    return NULL;
+  return value->v.handle.ptr;
+}
+
+static ScrChannel *scr_port_channel_of(const ScrDyn *value, int *side) {
+  ScrWorkerView *view = scr_port_view_of(value);
+  if (!view || !view->channel || view->detached) return NULL;
+  *side = view->side;
+  return scr_channel_retain(view->channel);
+}
+
+static bool scr_port_is(const ScrDyn *value) { return scr_port_view_of(value) != NULL; }
+
+/* the sender's port after a transfer: inert, and no longer this thread's */
+static void scr_port_detach(const ScrDyn *value) {
+  ScrWorkerView *view = scr_port_view_of(value);
+  if (!view || !view->channel || view->detached) return;
+  view->detached = true;
+  view->closing = false;
+  scr_channel_set_owner(view->channel, view->side, NULL, SCR_WORKER_WAKE);
+}
+
+static ScrDyn *scr_port_adopt(ScrChannel *channel, int side) {
+  if (!scr_worker_events_install()) return NULL;
+  ScrWorkerView *view = scr_channel_port_new(channel, side);
+  return scr_dyn_new_handle(view, SCR_DYNH_MESSAGE_PORT);
+}
+
+static const ScrMessagePortOps scr_port_ops = {
+    &scr_port_channel_of, &scr_port_is, &scr_port_detach, &scr_port_adopt, &scr_channel_release,
+};
+
+static void scr_port_ops_install(void) { scr_message_port_ops = &scr_port_ops; }
+
+/* new MessageChannel(): { port1, port2 } */
+ScrDyn *scr_worker_message_channel(void) {
+  if (!scr_worker_events_install()) return NULL;
+  ScrChannel *channel = scr_channel_new();
+  ScrWorkerView *a = scr_channel_port_new(channel, 0);
+  ScrWorkerView *b = scr_channel_port_new(channel, 1);
+  scr_channel_release(channel);
+  ScrDyn *object = scr_dyn_new_obj();
+  scr_dyn_obj_set(object, "port1", 5, scr_dyn_new_handle(a, SCR_DYNH_MESSAGE_PORT));
+  scr_dyn_obj_set(object, "port2", 5, scr_dyn_new_handle(b, SCR_DYNH_MESSAGE_PORT));
+  return object;
+}
+
+/* receiveMessageOnPort(port): { message } of the next queued message, or
+ * undefined */
+ScrDyn *scr_worker_receive_message_on_port(ScrDyn *port) {
+  ScrWorkerView *view = scr_port_view_of(port);
+  if (!view) {
+    static const char message[] = "The \"port\" argument must be a MessagePort instance";
+    scr_throw_error_msg_code(SCR_ERR_TYPE, message, sizeof message - 1, "ERR_INVALID_ARG_TYPE");
+    return NULL;
+  }
+  if (!view->channel) return scr_worker_refusal("receiveMessageOnPort on parentPort is not supported yet");
+  ScrMailEvent *event = view->closed || view->detached ? NULL : scr_channel_take(view->channel, view->side);
+  if (!event) return scr_dyn_retain(scr_dyn_undefined());
+  ScrDyn *value = event->message ? scr_message_decode(event->message) : NULL;
+  scr_mail_event_free(event);
+  if (!value) return NULL;
+  ScrDyn *object = scr_dyn_new_obj();
+  scr_dyn_obj_set(object, "message", 7, value);
+  return object;
+}
+
 ScrDyn *scr_worker_parent_port(void) {
   if (scr_context_is_main()) return scr_dyn_new_null();
   if (!scr_worker_events_install()) return NULL;
@@ -407,6 +713,7 @@ static ScrDyn *scr_worker_new_options(double root, ScrStr *filename, ScrDyn *opt
   if (!scr_worker_events_install()) return NULL;
   ScrDyn *data = scr_dyn_undefined();
   ScrDyn *argv = scr_dyn_undefined();
+  ScrDyn *transfer_list = scr_dyn_undefined();
   if (options->kind != SCR_DYN_UNDEF) {
     if (options->kind != SCR_DYN_OBJ) {
       scr_dyn_arg_type_fail("options", "object", options);
@@ -417,12 +724,16 @@ static ScrDyn *scr_worker_new_options(double root, ScrStr *filename, ScrDyn *opt
       if (entry->accessor) return scr_worker_refusal("Worker option accessors are not supported yet");
       if (entry->key_len == 10 && !memcmp(entry->key, "workerData", 10)) data = entry->value;
       else if (entry->key_len == 4 && !memcmp(entry->key, "argv", 4)) argv = entry->value;
+      else if (entry->key_len == 12 && !memcmp(entry->key, "transferList", 12)) transfer_list = entry->value;
       else return scr_worker_refusal("This Worker constructor option is not supported yet");
     }
   }
   ScrArr *arguments = scr_worker_arguments(filename, argv);
   if (!arguments) return NULL;
-  ScrMessage *wire = scr_message_encode(data);
+  ScrDyn *transfer = scr_worker_transfer_list(transfer_list);
+  if (scr_exc_pending()) { scr_arr_release(arguments); return NULL; }
+  ScrMessage *wire = scr_message_encode_transfer(data, transfer);
+  scr_dyn_release(transfer);
   if (!wire) { scr_arr_release(arguments); return NULL; }
   ScrWorkerThread *thread = scr_worker_thread_start(sc_worker_entry, (uint32_t)root, wire, scr_worker_events, arguments);
   scr_arr_release(arguments);