diff --git a/src/msgpack.cc b/src/msgpack.cc index f09dbf1..40202c1 100644 --- a/src/msgpack.cc +++ b/src/msgpack.cc @@ -317,8 +317,11 @@ static v8::Local Error(const char* msg) { return Nan::Error(msg); } -/* Persistent identity flag for cycle detection (not enumerable). */ -static Nan::Persistent 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 stack_key; static v8::Local StackKey() { return Nan::New(stack_key); @@ -607,9 +610,25 @@ static v8::Local 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(remaining_bytes_in_buffer)); @@ -761,6 +781,12 @@ NAN_MODULE_INIT(Init) { Nan::GetFunction(Nan::New(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 diff --git a/test/fixtures/msgpack-worker.js b/test/fixtures/msgpack-worker.js new file mode 100644 index 0000000..660465d --- /dev/null +++ b/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)); diff --git a/test/worker.test.js b/test/worker.test.js new file mode 100644 index 0000000..aa60cde --- /dev/null +++ b/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'); + }); +});