Encode posted records from their fields, and queue a message in one block (#1824)

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7139-message-typed-records.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7139-message-typed-records.ts"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7360-worker-typed-records.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/7360-worker-typed-records.ts"
+ ],
+ "diags": []
+}

packages/runtime/src/scr_dyn.h

@@ -364,6 +364,9 @@ typedef struct ScrDynTypedRecordOps {
    * for-in over a record with every field present (atomic slot). */
   ScrArr *keys;
 } ScrDynTypedRecordOps;
+/* A record capsule's ops while it answers from the native record (no view
+ * created yet), or NULL. */
+const ScrDynTypedRecordOps *scr_dyn_typed_ref_live_record(const ScrDyn *d);
 
 /* Prototype members of a compiled class seen as `unknown`, emitted per
  * class by the compiler: the string names of methods, getters and setters

packages/runtime/src/scr_json.c

@@ -3591,6 +3591,10 @@ static const ScrDynTypedRecordOps *scr_typed_ref_record_ops(const ScrDyn *d) {
   return (const ScrDynTypedRecordOps *)(const void *)d->v.typed_ref.array;
 }
 
+const ScrDynTypedRecordOps *scr_dyn_typed_ref_live_record(const ScrDyn *d) {
+  return d->kind == SCR_DYN_TYPED_REF ? scr_typed_ref_record_ops(d) : NULL;
+}
+
 static bool scr_typed_ref_record_field(const ScrDynTypedRecordOps *ops, const ScrStr *key,
                                        size_t *out) {
   /* Compiled reads pass the interned literal the field names also use. */

packages/runtime/src/scr_mailbox.c

@@ -140,9 +140,11 @@ ScrMailbox *scr_mailbox_retain(ScrMailbox *mailbox) {
 
 void scr_mail_event_free(ScrMailEvent *event) {
   if (!event) return;
-  scr_message_free(event->message);
   free(event->data);
-  free(event);
+  ScrMessage *message = event->message;
+  /* An event in its message's block goes with the message. */
+  if (!event->in_message) free(event);
+  scr_message_free(message);
 }
 
 void scr_mailbox_release(ScrMailbox *mailbox) {
@@ -460,7 +462,7 @@ void scr_channel_set_owner(ScrChannel *channel, int side, ScrMailbox *owner, uns
  * messages as ready, so it does not wait while the queue is non-empty. */
 bool scr_channel_post(ScrChannel *channel, int side, ScrMessage *message, unsigned message_kind,
                       unsigned wake_kind) {
-  ScrMailEvent *event = calloc(1, sizeof(*event));
+  ScrMailEvent *event = message ? scr_message_mail_event(message) : calloc(1, sizeof(*event));
   if (!event) scr_trap("scriptc: out of memory\n");
   event->kind = message_kind;
   event->message = message;

packages/runtime/src/scr_mailbox.h

@@ -12,8 +12,13 @@ typedef struct ScrMailEvent {
   unsigned char *data; /* raw bytes (a worker's captured stdout or stderr), owned */
   size_t length;
   struct ScrMailEvent *next;
+  bool in_message; /* the event is the one message's block holds (scr_message_mail_event) */
 } ScrMailEvent;
 
+/* The event that carries `message`, in the message's own block: a queued
+ * message is then one allocation. Freeing the event frees the message. */
+ScrMailEvent *scr_message_mail_event(ScrMessage *message);
+
 ScrMailbox *scr_mailbox_new(void);
 ScrMailbox *scr_mailbox_retain(ScrMailbox *mailbox);
 void scr_mailbox_release(ScrMailbox *mailbox);

packages/runtime/src/scr_message.c

@@ -1,4 +1,5 @@
 #include "scr_message.h"
+#include "scr_mailbox.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -72,6 +73,7 @@ struct ScrMessage {
   bool storage; /* has STORAGE nodes */
   bool releases; /* has STORAGE, PORT or HANDLE nodes, which own resources */
   bool readable; /* see scr_message_readable */
+  ScrMailEvent event; /* the event that carries the message (scr_message_mail_event) */
 };
 
 typedef struct {
@@ -98,6 +100,10 @@ typedef struct {
  * and properties allocates only its block. */
 #define SCR_MESSAGE_INLINE 8
 
+/* How deep live records nest before the rest of a graph encodes through
+ * the visit stack (scr_message_value recurses for them). */
+#define SCR_MESSAGE_RECORD_DEPTH 32
+
 typedef struct {
   ScrMessageNode *nodes;
   ScrMessageVisit *visits;
@@ -121,6 +127,7 @@ typedef struct {
   bool storage;
   bool releases;
   bool shared; /* a node was reached twice */
+  size_t record_depth; /* live records being encoded from their fields */
   bool adopts; /* has PORT or HANDLE nodes */
   ScrMessageNode inline_nodes[SCR_MESSAGE_INLINE];
   ScrMessageVisit inline_visits[SCR_MESSAGE_INLINE];
@@ -148,6 +155,7 @@ static void scr_message_encoder_init(ScrMessageEncoder *encoder, const ScrDyn *t
   encoder->transfer = transfer;
   encoder->uncaught = uncaught;
   encoder->storage = encoder->releases = encoder->shared = encoder->adopts = false;
+  encoder->record_depth = 0;
 }
 
 static bool scr_message_listed(const ScrMessageEncoder *encoder, const ScrDyn *value) {
@@ -321,6 +329,10 @@ static ScrDyn *scr_message_uncaught_keys(const ScrDyn *view) {
   return keys;
 }
 
+static void scr_message_property(ScrMessageEncoder *encoder, size_t id,
+                                  const ScrStr *key, size_t value, bool hidden);
+static bool scr_message_drain(ScrMessageEncoder *encoder, size_t base);
+
 static size_t scr_message_value(ScrMessageEncoder *encoder, const ScrDyn *input) {
   bool uncaught = encoder->uncaught;
   encoder->uncaught = false;
@@ -382,6 +394,46 @@ static size_t scr_message_value(ScrMessageEncoder *encoder, const ScrDyn *input)
     node->v.published = (ScrDyn *)input;
     return id;
   }
+  /* A live record capsule (scr_dyn_typed_ref_bind_record) whose view was
+   * never created encodes its fields from the record: the view would hold
+   * exactly those keys, in that order, with the values its ops box. Nested
+   * records encode the same way up to a depth, then through their views
+   * and the visit stack. */
+  const ScrDynTypedRecordOps *record =
+      input->kind == SCR_DYN_TYPED_REF && encoder->record_depth < SCR_MESSAGE_RECORD_DEPTH
+          ? scr_dyn_typed_ref_live_record(input) : NULL;
+  if (record) {
+    encoder->visits[id].view = (ScrDyn *)input;
+    encoder->nodes[id].kind = SCR_MESSAGE_OBJECT;
+    void *rec = input->v.typed_ref.ptr;
+    encoder->record_depth++;
+    size_t result = id;
+    /* A field's value encodes completely, its containers' properties
+     * included, before the next field's: getters run in Node's order. */
+    size_t base = encoder->depth;
+    for (size_t i = 0; i < record->count; i++) {
+      if (!record->fixed && !record->has(rec, i)) continue;
+      ScrDyn *child = record->get(rec, i);
+      if (!child || scr_exc_pending()) {
+        scr_dyn_release(child);
+        result = SIZE_MAX;
+        break;
+      }
+      size_t next = scr_message_value(encoder, child);
+      scr_dyn_release(child);
+      if (next == SIZE_MAX) {
+        result = SIZE_MAX;
+        break;
+      }
+      scr_message_property(encoder, id, record->names[i], next, false);
+      if (encoder->depth > base && !scr_message_drain(encoder, base)) {
+        result = SIZE_MAX;
+        break;
+      }
+    }
+    encoder->record_depth--;
+    return result;
+  }
   ScrDyn *value = input->kind == SCR_DYN_TYPED_REF
       ? scr_dyn_typed_ref_materialize(input) : (ScrDyn *)input;
   encoder->visits[id].view = value;
@@ -756,8 +808,8 @@ static ScrMessage *scr_message_pack(ScrMessageEncoder *encoder) {
   ScrMessageProperty *slots = (ScrMessageProperty *)(nodes + count);
   ScrMessageTyped *records = (ScrMessageTyped *)(slots + properties);
   char *text = (char *)(records + typed);
-  *message = (ScrMessage){nodes, count, encoder->storage, encoder->releases,
-                         !encoder->shared && !encoder->adopts};
+  *message = (ScrMessage){.nodes = nodes, .count = count, .storage = encoder->storage,
+                          .releases = encoder->releases, .readable = !encoder->shared && !encoder->adopts};
   memcpy(nodes, encoder->nodes, count * sizeof(*nodes));
   if (typed) memcpy(records, encoder->typed, typed * sizeof(*records));
   size_t first = 0;
@@ -788,10 +840,11 @@ static ScrMessage *scr_message_pack(ScrMessageEncoder *encoder) {
   return message;
 }
 
-static ScrMessage *scr_message_encode_run(ScrMessageEncoder *encoder, const ScrDyn *value) {
-  ScrMessage *message = NULL;
-  if (scr_message_value(encoder, value) == SIZE_MAX) goto done;
-  while (encoder->depth && !scr_exc_pending()) {
+/* Encodes the properties of the containers on the visit stack above
+ * `base`, depth first, as Node's serializer visits them (getters run in
+ * that order). False when a property could not be read or encoded. */
+static bool scr_message_drain(ScrMessageEncoder *encoder, size_t base) {
+  while (encoder->depth > base && !scr_exc_pending()) {
     size_t id = encoder->stack[encoder->depth - 1];
     ScrMessageVisit *visit = &encoder->visits[id];
     if (visit->cursor == visit->keys->v.arr.len) { encoder->depth--; continue; }
@@ -799,7 +852,7 @@ static ScrMessage *scr_message_encode_run(ScrMessageEncoder *encoder, const ScrD
     bool error = encoder->nodes[id].kind == SCR_MESSAGE_ERROR;
     if (key->kind != SCR_DYN_STR || (!error && !scr_dyn_property_is_enumerable(visit->view, key->v.str))) continue;
     ScrDyn *child = scr_dyn_bag_get(visit->view, key->v.str, visit->view);
-    if (!child || scr_exc_pending()) { scr_dyn_release(child); break; }
+    if (!child || scr_exc_pending()) { scr_dyn_release(child); return false; }
     bool hidden = error;
     if (visit->uncaught) {
       /* Node keeps an uncaught Error's data properties except functions
@@ -810,10 +863,15 @@ static ScrMessage *scr_message_encode_run(ScrMessageEncoder *encoder, const ScrD
     }
     size_t next = scr_message_value(encoder, child);
     scr_dyn_release(child);
-    if (next == SIZE_MAX) break;
+    if (next == SIZE_MAX) return false;
     scr_message_property(encoder, id, key->v.str, next, hidden);
   }
-done:
+  return !scr_exc_pending();
+}
+
+static ScrMessage *scr_message_encode_run(ScrMessageEncoder *encoder, const ScrDyn *value) {
+  ScrMessage *message = NULL;
+  if (scr_message_value(encoder, value) != SIZE_MAX) scr_message_drain(encoder, 0);
   if (!scr_exc_pending()) message = scr_message_pack(encoder);
   else
     for (size_t i = 0; i < encoder->count; i++) scr_message_drop(&encoder->nodes[i]);
@@ -1003,6 +1061,11 @@ done:;
 
 bool scr_message_readable(const ScrMessage *message) { return message->readable; }
 
+ScrMailEvent *scr_message_mail_event(ScrMessage *message) {
+  message->event = (ScrMailEvent){.message = message, .in_message = true};
+  return &message->event;
+}
+
 int32_t scr_message_kind(const ScrMessage *message, size_t node) { return message->nodes[node].kind; }
 
 double scr_message_number(const ScrMessage *message, size_t node) { return message->nodes[node].v.number; }

packages/runtime/src/scr_worker.c

@@ -99,9 +99,10 @@ static bool scr_worker_stats(void) {
 
 static ScrMailEvent *scr_worker_event(unsigned kind, uint64_t source,
                                      int code, ScrMessage *message) {
-  ScrMailEvent *event = calloc(1, sizeof(*event));
+  ScrMailEvent *event = message ? scr_message_mail_event(message) : calloc(1, sizeof(*event));
   if (!event) scr_trap("scriptc: out of memory\n");
-  *event = (ScrMailEvent){.kind = kind, .source = source, .code = code, .message = message};
+  *event = (ScrMailEvent){
+      .kind = kind, .source = source, .code = code, .message = message, .in_message = message != NULL};
   return event;
 }
 

tests/corpus/7139-message-typed-records.ts

@@ -0,0 +1,85 @@
+import { MessageChannel } from "node:worker_threads";
+
+// Typed records posted on a MessageChannel arrive as plain objects with
+// the same keys, order and values as in Node: optional fields present and
+// absent, nested records, a record reached twice, a record nested deeper
+// than the encoder recurses, a record parsed with undeclared keys, and a
+// function-valued field, which postMessage rejects.
+type Point = { x: number; y: number };
+type Item = { id: number; name: string; done: boolean; pos: Point; note?: string };
+type Pair = { left: Point; right: Point };
+type Link = { value: number; next?: Link };
+
+const { port1, port2 } = new MessageChannel();
+const lines: string[] = [];
+const expected = 7;
+port1.on("message", (value: unknown) => {
+  let checks = "";
+  if (!Array.isArray(value)) {
+    const object = value as Record<string, unknown>;
+    if ("left" in object) checks += ` same ${object["left"] === object["right"]}`;
+    const pos = object["pos"] as Record<string, unknown> | undefined;
+    if (pos !== undefined) checks += ` negative zero ${Object.is(pos["x"], -0)}`;
+  }
+  lines.push(JSON.stringify(value) + checks);
+  if (lines.length === expected) {
+    for (const line of lines) console.log(line);
+    port1.close();
+  }
+});
+
+const item: Item = { id: 1, name: "first", done: false, pos: { x: 1, y: 2 } };
+port2.postMessage(item);
+const noted: Item = { id: 2, name: "", done: true, pos: { x: -0, y: 0.5 }, note: "n" };
+port2.postMessage(noted);
+
+const shared: Point = { x: 3, y: 4 };
+const pair: Pair = { left: shared, right: shared };
+port2.postMessage(pair);
+
+let chain: Link = { value: 0 };
+for (let i = 1; i < 40; i++) chain = { value: i, next: chain };
+port2.postMessage(chain);
+
+const parsed = JSON.parse('{"y":6,"extra":true,"x":5}') as Point;
+port2.postMessage(parsed);
+port2.postMessage({ x: 7, y: 8 } as Point);
+port2.postMessage([item.pos, noted.pos]);
+
+type WithFunction = { id: number; run: () => number };
+const bad: WithFunction = { id: 9, run: () => 1 };
+try {
+  port2.postMessage(bad);
+} catch (error) {
+  console.log("rejected", (error as Error).name);
+}
+
+// Fields that hold objects with getters: each field's value is read
+// completely, its getters included, before the next field's, as in Node.
+type Holder = { id: number; first: unknown; second: unknown };
+const reads: string[] = [];
+const getters: Holder = {
+  id: 1,
+  first: {
+    get a() {
+      reads.push("first.a");
+      return 1;
+    },
+    nested: {
+      get b() {
+        reads.push("first.nested.b");
+        return 2;
+      },
+    },
+  },
+  second: {
+    get c() {
+      reads.push("second.c");
+      return 3;
+    },
+  },
+};
+const { port1: getterPort, port2: getterSender } = new MessageChannel();
+getterSender.postMessage(getters);
+console.log("getter order", reads.join(","));
+getterPort.close();

tests/corpus/7360-worker-typed-records.ts

@@ -0,0 +1,37 @@
+import { Worker, isMainThread, parentPort, workerData } from "node:worker_threads";
+
+// Typed records posted to a worker and back keep their keys, order and
+// values, as in Node: records with optional fields, nested records and a
+// record in workerData.
+type Point = { x: number; y: number };
+type Job = { id: number; label: string; urgent: boolean; at: Point; note?: string };
+type Reply = { id: number; sum: number; label: string };
+
+if (isMainThread) {
+  const start: Point = { x: 10, y: 20 };
+  const worker = new Worker(new URL(import.meta.url), { workerData: start });
+  const replies: string[] = [];
+  worker.on("message", (reply: Reply) => {
+    replies.push(`${JSON.stringify(reply)} ${Object.keys(reply).join(",")}`);
+    if (replies.length === 3) {
+      for (const line of replies) console.log(line);
+      void worker.terminate();
+    }
+  });
+  const jobs: Job[] = [
+    { id: 1, label: "a", urgent: false, at: { x: 1, y: 2 } },
+    { id: 2, label: "b", urgent: true, at: { x: 3, y: 4 }, note: "n" },
+    { id: 3, label: "", urgent: false, at: { x: -1, y: 0.25 } },
+  ];
+  for (const job of jobs) worker.postMessage(job);
+} else {
+  const origin = workerData as Point;
+  parentPort!.on("message", (job: Job) => {
+    const reply: Reply = {
+      id: job.id,
+      sum: job.at.x + job.at.y + origin.x + origin.y,
+      label: `${job.label}${job.urgent ? "!" : ""}${job.note ?? ""}`,
+    };
+    parentPort!.postMessage(reply);
+  });
+}

tests/harness/corpus-costs/7139-message-typed-records.ts.json

@@ -0,0 +1 @@
+{ "plain": 4300, "san": 4400 }

tests/harness/corpus-costs/7360-worker-typed-records.ts.json

@@ -0,0 +1 @@
+{ "plain": 2700, "san": 2900 }