fix: load native addon in worker_threads

src/msgpack.cc

@@ -317,8 +317,11 @@ static v8::Local<v8::Value> Error(const char* msg) {
   return Nan::Error(msg);
 }
 
-/* Persistent identity flag for cycle detection (not enumerable). */
-static Nan::Persistent<v8::String> stack_key;
+/* Persistent identity flag for cycle detection (not enumerable).
+ * thread_local because a v8::Persistent belongs to the isolate that created
+ * it: with a process-global handle, a worker's Init() would dispose the main
+ * isolate's string and then hand its own back to main-thread pack(). */
+static thread_local Nan::Persistent<v8::String> stack_key;
 
 static v8::Local<v8::String> StackKey() {
   return Nan::New(stack_key);
@@ -607,9 +610,25 @@ static v8::Local<v8::Value> MsgpackToJs(const msgpack_object* mo) {
 struct SbufPool {
   msgpack_sbuffer* list[kSbufferPoolMax];
   size_t length;
+
+  SbufPool() : list(), length(0) {}
+
+  /* Each thread owns its pool, so release the cached sbuffers when the
+   * thread goes away instead of leaking them per worker. */
+  ~SbufPool() {
+    while (length > 0) {
+      msgpack_sbuffer_free(list[--length]);
+    }
+  }
+
+ private:
+  SbufPool(const SbufPool&);
+  SbufPool& operator=(const SbufPool&);
 };
 
-static SbufPool sbuf_pool = {{0}, 0};
+/* thread_local, not process-global: a worker thread packing concurrently with
+ * the main thread would otherwise hand the same sbuffer to both. */
+static thread_local SbufPool sbuf_pool;
 
 class PackBuffer {
  public:
@@ -627,16 +646,15 @@ class PackBuffer {
     }
   }
 
+  /* Offer the sbuffer back to this thread's pool whether or not it came from
+   * there: only handing back pooled buffers would leave the pool permanently
+   * empty, so every pack would malloc and every dtor would free. */
   ~PackBuffer() {
     if (sb_ == NULL) return;
-    if (from_pool_) {
-      if (sbuf_pool.length == kSbufferPoolMax) {
-        msgpack_sbuffer_free(sb_);
-      } else {
-        sbuf_pool.list[sbuf_pool.length++] = sb_;
-      }
-    } else {
+    if (sbuf_pool.length == kSbufferPoolMax) {
       msgpack_sbuffer_free(sb_);
+    } else {
+      sbuf_pool.list[sbuf_pool.length++] = sb_;
     }
     sb_ = NULL;
   }
@@ -662,7 +680,9 @@ static void MsgpackFree(char* data, void* hint) {
   free(data);
 }
 
-static int remaining_bytes_in_buffer = 0;
+/* thread_local for the same reason as sbuf_pool: unpack() on a worker must
+ * not clobber the value the main thread's unpack.bytes_remaining reads. */
+static thread_local int remaining_bytes_in_buffer = 0;
 
 NAN_METHOD(BytesRemaining) {
   info.GetReturnValue().Set(Nan::New<v8::Number>(remaining_bytes_in_buffer));
@@ -761,6 +781,12 @@ NAN_MODULE_INIT(Init) {
            Nan::GetFunction(Nan::New<v8::FunctionTemplate>(BytesRemaining)).ToLocalChecked());
 }
 
-NODE_MODULE(msgpackBinding, Init)
+/* Context-aware: without this the addon refuses to load in a worker_threads
+ * Worker ("Module did not self-register"). NAN_MODULE_WORKER_ENABLED is the
+ * right wrapper here -- Init is a NAN_MODULE_INIT, i.e. a one-argument
+ * function, while NODE_MODULE_CONTEXT_AWARE expects a four-argument
+ * addon_context_register_func and casts between the mismatched function
+ * pointer types. */
+NAN_MODULE_WORKER_ENABLED(msgpackBinding, Init)
 
 }  // namespace

test/fixtures/msgpack-worker.js

@@ -0,0 +1,93 @@
+'use strict';
+
+/* Loaded inside a worker_threads Worker. Requiring the addon here is the
+ * actual failure in msgpack-node#60: a non-context-aware NODE_MODULE throws
+ * "Module did not self-register" the second time the .node file is loaded. */
+const { parentPort, workerData } = require('worker_threads');
+const msgpack = require('../../');
+
+/* Bail out rather than hang if the main thread never opens the start gate. */
+const GATE_TIMEOUT_MS = 30000;
+
+/* Build a payload that is unique per worker and per iteration, so a buffer
+ * leaking between threads shows up as wrong data rather than as luck. */
+function payload(id, i) {
+  return {
+    id: id,
+    i: i,
+    tag: 'w' + id + '-' + i,
+    blob: Buffer.from(('x' + id).repeat(1 + (i % 37))),
+    list: [i, i * 2, i * 3, null, true, 'n' + i],
+    nested: { depth: { value: id * 100000 + i } }
+  };
+}
+
+function run(op) {
+  switch (op) {
+    case 'roundtrip': {
+      const value = { a: 1, b: Buffer.from('hi') };
+      const unpacked = msgpack.unpack(msgpack.pack(value));
+      return {
+        a: unpacked.a,
+        bIsBuffer: Buffer.isBuffer(unpacked.b),
+        b: unpacked.b.toString('latin1'),
+        keys: Object.keys(unpacked)
+      };
+    }
+    case 'cycle': {
+      const o = { name: 'loop' };
+      o.self = o;
+      try {
+        msgpack.pack(o);
+        return { threw: false, message: null };
+      } catch (err) {
+        return { threw: true, message: err.message };
+      }
+    }
+    case 'unpack-remaining': {
+      /* A different buffer than the main thread used, leaving a different
+       * number of trailing bytes behind. */
+      const buf = Buffer.concat([msgpack.pack('worker'), Buffer.alloc(7)]);
+      const value = msgpack.unpack(buf);
+      return { value: value, bytesRemaining: msgpack.unpack.bytes_remaining };
+    }
+    case 'concurrent-pack': {
+      /* Many pack/unpack round-trips of distinct payloads, racing the main
+       * thread and the other workers. Exercises the thread_local sbuf pool:
+       * a shared pool would hand the same sbuffer to two threads at once. */
+      const id = workerData.id;
+      const iterations = workerData.iterations;
+      const gate = new Int32Array(workerData.gate);
+
+      /* Announce that the addon is loaded, then block until the main thread
+       * opens the gate. Spawning a Worker takes far longer than this loop
+       * does, so without the barrier the main thread would be done packing
+       * before any worker started and nothing would actually overlap. */
+      parentPort.postMessage({ ready: true });
+      if (Atomics.load(gate, 0) === 0) {
+        Atomics.wait(gate, 0, 0, GATE_TIMEOUT_MS);
+      }
+      if (Atomics.load(gate, 0) === 0) {
+        throw new Error('worker ' + id + ' timed out waiting for the start gate');
+      }
+
+      for (let i = 0; i < iterations; i++) {
+        const value = payload(id, i);
+        const unpacked = msgpack.unpack(msgpack.pack(value));
+        if (unpacked.id !== id || unpacked.i !== i || unpacked.tag !== value.tag ||
+            unpacked.nested.depth.value !== value.nested.depth.value ||
+            unpacked.blob.toString('latin1') !== value.blob.toString('latin1') ||
+            unpacked.list.length !== value.list.length ||
+            unpacked.list[5] !== value.list[5]) {
+          throw new Error('worker ' + id + ' round-trip mismatch at ' + i + ': ' +
+                          JSON.stringify(unpacked));
+        }
+      }
+      return { id: id, iterations: iterations, ok: true };
+    }
+    default:
+      throw new Error('unknown op: ' + op);
+  }
+}
+
+parentPort.postMessage(run(workerData.op));

test/worker.test.js

@@ -0,0 +1,170 @@
+'use strict';
+
+const { describe, it } = require('node:test');
+const assert = require('node:assert/strict');
+const path = require('path');
+const { Worker } = require('worker_threads');
+
+/* Load the addon on the main thread first: the regression only shows up on
+ * the *second* load of msgpackBinding.node, i.e. inside the worker. */
+const msgpack = require('../');
+const binding = require('../build/Release/msgpackBinding');
+
+const WORKER = path.join(__dirname, 'fixtures', 'msgpack-worker.js');
+
+function runWorker(op, extra, onReady) {
+  return new Promise((resolve, reject) => {
+    const workerData = Object.assign({ op: op }, extra);
+    const worker = new Worker(WORKER, { workerData: workerData });
+    let message;
+    let settled = false;
+    worker.on('message', (m) => {
+      /* A {ready:true} note means the worker has loaded the addon and is
+       * parked on the start gate; the real result comes later. */
+      if (m && m.ready === true) {
+        if (onReady) onReady(worker);
+        return;
+      }
+      message = m;
+    });
+    worker.once('error', (err) => {
+      settled = true;
+      reject(err);
+    });
+    worker.once('exit', (code) => {
+      if (settled) return;
+      if (code !== 0) return reject(new Error('worker exited with code ' + code));
+      resolve(message);
+    });
+  });
+}
+
+describe('worker_threads', () => {
+  it('loads the addon inside a worker and round-trips values', async () => {
+    const result = await runWorker('roundtrip');
+    assert.deepEqual(result.keys, ['a', 'b']);
+    assert.equal(result.a, 1);
+    assert.equal(result.bIsBuffer, true, 'bin stays a Buffer in the worker');
+    assert.equal(result.b, 'hi');
+  });
+
+  it('still detects cycles inside a worker', async () => {
+    const result = await runWorker('cycle');
+    assert.equal(result.threw, true);
+    assert.match(result.message, /circular/);
+  });
+
+  it('leaves the main thread working after a worker has loaded the addon', async () => {
+    await runWorker('roundtrip');
+
+    const value = { a: 1, b: Buffer.from('hi'), c: [1, 'two', null] };
+    const unpacked = msgpack.unpack(msgpack.pack(value));
+    assert.equal(unpacked.a, 1);
+    assert.ok(Buffer.isBuffer(unpacked.b));
+    assert.equal(unpacked.b.toString('latin1'), 'hi');
+    assert.deepEqual(unpacked.c, [1, 'two', null]);
+
+    /* Cycle detection uses a process-wide private key; check it survived the
+     * worker's module init. */
+    const cyclic = {};
+    cyclic.self = cyclic;
+    assert.throws(() => msgpack.pack(cyclic), /circular/);
+  });
+
+  it('does not let a worker unpack clobber the main thread bytes_remaining', async () => {
+    const buf = Buffer.concat([msgpack.pack(1), Buffer.alloc(3)]);
+    assert.equal(msgpack.unpack(buf), 1);
+    assert.equal(msgpack.unpack.bytes_remaining, 3);
+
+    const result = await runWorker('unpack-remaining');
+    assert.equal(result.value, 'worker');
+    assert.equal(result.bytesRemaining, 7);
+
+    /* The JS-level snapshot is per-thread by construction; the native counter
+     * behind it is a single global unless it is thread_local. */
+    assert.equal(msgpack.unpack.bytes_remaining, 3);
+    assert.equal(binding.bytesRemaining(), 3);
+  });
+
+  it('packs on several workers concurrently with the main thread', async () => {
+    const WORKERS = 4;
+    const ITERATIONS = 4000;
+
+    /* gate[0] flips to 1 when every worker has loaded the addon, releasing
+     * all of them at once so their packing really does overlap this thread's. */
+    const gate = new Int32Array(new SharedArrayBuffer(4));
+
+    let ready = 0;
+    let releaseAll;
+    const allReady = new Promise((resolve) => { releaseAll = resolve; });
+
+    const running = [];
+    for (let id = 1; id <= WORKERS; id++) {
+      running.push(runWorker('concurrent-pack', { id: id, iterations: ITERATIONS, gate: gate.buffer }, () => {
+        if (++ready === WORKERS) releaseAll();
+      }));
+    }
+
+    /* If a worker dies before signalling ready, surface that error instead of
+     * waiting on a barrier that will never be reached. */
+    await Promise.race([allReady, Promise.all(running)]);
+
+    Atomics.store(gate, 0, 1);
+    Atomics.notify(gate, 0);
+
+    /* Pack on the main thread while the workers are packing. The pack buffer
+     * pool is thread_local, so nobody should ever see another thread's bytes. */
+    for (let i = 0; i < ITERATIONS; i++) {
+      const value = {
+        id: 0,
+        i: i,
+        tag: 'main-' + i,
+        blob: Buffer.from('m'.repeat(1 + (i % 53))),
+        list: [i, 'main' + i, null, false],
+        nested: { depth: { value: -1 - i } }
+      };
+      const unpacked = msgpack.unpack(msgpack.pack(value));
+      assert.equal(unpacked.id, 0);
+      assert.equal(unpacked.i, i);
+      assert.equal(unpacked.tag, 'main-' + i);
+      assert.equal(unpacked.blob.toString('latin1'), value.blob.toString('latin1'));
+      assert.deepEqual(unpacked.list, value.list);
+      assert.equal(unpacked.nested.depth.value, -1 - i);
+    }
+
+    const results = await Promise.all(running);
+    assert.equal(results.length, WORKERS);
+    results.forEach((result, index) => {
+      assert.ok(result, 'worker ' + (index + 1) + ' posted a result');
+      assert.equal(result.ok, true);
+      assert.equal(result.id, index + 1);
+      assert.equal(result.iterations, ITERATIONS);
+    });
+
+    /* The main thread keeps working once the workers are gone: each worker's
+     * pool was freed with its thread, not handed back to this one. */
+    assert.deepEqual(msgpack.unpack(msgpack.pack({ after: [1, 2, 3] })), { after: [1, 2, 3] });
+  });
+
+  it('reuses the pack buffer pool across many sequential packs', () => {
+    /* The first pack mallocs its sbuffer and hands the memory to Node
+     * (NewBuffer); every later one takes a pooled sbuffer and copies out
+     * (CopyBuffer). Both paths have to produce identical bytes. */
+    const value = { a: 1, b: Buffer.from('hi'), c: [1, 'two', null], d: { e: -7 } };
+    const first = msgpack.pack(value);
+    for (let i = 0; i < 2000; i++) {
+      const packed = msgpack.pack(value);
+      assert.ok(packed.equals(first), 'pack #' + i + ' differs from the first pack');
+      const unpacked = msgpack.unpack(packed);
+      assert.equal(unpacked.a, 1);
+      assert.equal(unpacked.b.toString('latin1'), 'hi');
+      assert.deepEqual(unpacked.c, [1, 'two', null]);
+      assert.equal(unpacked.d.e, -7);
+    }
+
+    /* A pooled sbuffer is reused after a much larger pack, so a stale size or
+     * leftover bytes would show up here. */
+    msgpack.pack({ big: Buffer.alloc(200000, 0x41) });
+    assert.ok(msgpack.pack(value).equals(first), 'pack after a large pack differs');
+  });
+});