Node.js worker threads shared object/store

Question

So, I was reading some stuff regarding Node.js and I was amazed when I came across Worker Threads.

Having threads in my opinion is a great plus especially if you combine it with shared memory access. As you might think already -> SharedArrayBuffer...

Yeap that's what I thought. So The first thing that came into my mind was to give it a little test and try to implement a simple store (a simple object for now) that would be shared among the threads.

The question is, (unless I'm missing something here) how can you make an object accessible from n threads with the use of SharedArrayBuffer?

I know that for a simple Uint32Array is doable, but regarding the object what can be done?

At first I thought to convert it to a Uint32Array as you may see bellow, but even looking at the damn source code makes me wanna cry...

const {
    Worker,
    isMainThread,
    workerData
} = require('worker_threads');

const store = {
    ks109: {
        title: 'some title 1',
        description: 'some desciption 1',
        keywords: ['one', 'two']
    },
    ks110: {
        title: 'some title 2',
        description: 'some desciption 2',
        keywords: ['three', 'four']
    },
    ks111: {
        title: 'some title 3',
        description: 'some desciption 3',
        keywords: ['five', 'six']
    }
}

const shareObject = (obj) => {

    let OString = JSON.stringify(obj);
    let SABuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * OString.length);
    let sArray = new Int32Array(SABuffer);

    for (let i = 0; i < OString.length; i++) {
        sArray[i] = OString.charCodeAt(i);
    }

    return sArray;

}

if (isMainThread) {

    const sharedStore = shareObject(store);

    console.log('[Main][Data Before]:', sharedStore.slice(-10));

    let w = new Worker(__filename, {
        workerData: sharedStore
    });

    w.on('message', (msg) => {
        console.log(`[Main][Thread message]: ${msg}`);
    });
    w.on('error', (err) => {
        console.error(`[Main][Thread error] ${err.message}`);
    });
    w.on('exit', (code) => {
        if (code !== 0) {
            console.error(`[Main][Thread unexpected exit]: ${code}`);
        }
    });

    setInterval(() => {
        // At some point you ll see a value change,
        // it's 'six' becoming 'sax' (big time!) 
        console.log('[Main][Data Check]:', sharedStore.slice(-10));
    }, 1000);

} else {

    let str = String.fromCharCode.apply(this, workerData);
    let obj = JSON.parse(str);

    obj.ks111.keywords[1] = 'sax'; // big time!

    let OString = JSON.stringify(obj);

    for (let i = 0; i < OString.length; i++) {
        workerData[i] = OString.charCodeAt(i);
    }

}

In conclusion, shared object among threads in Node.js 10.5.0, is it possible? How?

Answer 1

ECMA Script contains no shared objects but it has SharedArrayBuffer. And you can implement such behavior on your own writing data directly in buffer using DataView and wrapper:

// Shared value
class SharedPoint {
  constructor(array) {
    this.dataview = new DataView(array);
  }

  set x(value) {
    this.dataview.setUint8(0, value);
  }

  set y(value) {
    this.dataview.setUint8(1, value);
  }

  get x() {
    return this.dataview.getUint8(0);
  }

  get y() {
    return this.dataview.getUint8(1);
  }
}

// Usage

const buffer = new SharedArrayBuffer(2);

// Create two instances of shared point.
const point0 = new SharedPoint(buffer);
const point1 = new SharedPoint(buffer);

// Get initial values for point #1
console.log('x', point1.x); // 0
console.log('y', point1.y); // 0

// Update point #0
point0.x = 64;
point0.y = 32;

// Get changes in point #1
console.log('x', point1.x); // 64
console.log('y', point1.y); // 32

You are able to create class which can manipulate strings or C-like structures. While SharedArrayBuffer is transferable object it can be shared between worker and main process.

⚠️ Note Due to Spectre attack SharedArrayBuffer was disabled by all major browsers and reenabled. Though the API is mature, its' support could be lower than one might expect. Check browsers support at can i use.

Answer 2

No native solution, but you can create a serializer/deserializer that's backed by SharedArrayBuffer.

You can check this repo - a library I haven't yet published - it takes a schema (object structure) and (de)serializes it to (from) ArrayBuffer or SharedArrayBuffer. This can be used to serialize on main thread and deserialize on worker thread by passing buffers. It supports fixed arrays of objects and nested objects.

Answer 3

The code can be written like below. It can be seen that in worker the object was changed after parentPort.postMessage(sArray). This shows that threads use shared memory with SharedArrayBuffer

const {
    Worker,
    isMainThread,
    workerData,
    parentPort
} = require('worker_threads');

const store = {
    ks109: {
        title: 'some title 1',
        description: 'some desciption 1',
        keywords: ['one', 'two']
    },
    ks110: {
        title: 'some title 2',
        description: 'some desciption 2',
        keywords: ['three', 'four']
    },
    ks111: {
        title: 'some title 3',
        description: 'some desciption 3',
        keywords: ['five', 'six']
    }
}

if (isMainThread) {
    let w = new Worker(__filename, {
        workerData: store
    });

    w.on('message', (data) => {
        console.log("Received message from worker");
        const strArr = []
        for(let i = 0; i < data.byteLength; i++){
            strArr.push(String.fromCharCode(data.getUint8(i)));
        }
        console.log(JSON.parse(strArr.join("")))
    });
    w.on('error', (err) => {
        console.error(`[Main][Thread error] ${err.message}`);
    });
    w.on('exit', (code) => {
        if (code !== 0) {
            console.error(`[Main][Thread unexpected exit]: ${code}`);
        }
    });

} else {
    let OString = JSON.stringify(workerData);
    let SABuffer = new SharedArrayBuffer(OString.length);
    let sArray = new DataView(SABuffer);
    for (let i = 0; i < OString.length; i++) {
        sArray.setUint8(i,OString.charCodeAt(i))
    }
    parentPort.postMessage(sArray);
    let index1 = OString.indexOf("ks111");
    const key1SubString = OString.substring(index1);
    let index2 = key1SubString.indexOf("keywords");
    const key2SubString = key1SubString.substring(index2);
    let index3 = key2SubString.indexOf("six");
    const newIndex = index1+index2+index3+1;
    sArray.setUint8(newIndex,'a'.charCodeAt());
}