How to Implement a postMessage & onMessage Solution Based on objectbuffer Instead of IPC?

[aweiu]

Why?

In my application, the original solution was postMessage+ on('message'). Unfortunately, they are too slow for my application! High frequency + large objects are not suitable for IPC. Hence the idea of driving messages based on the objectbuffer.

The first problem to solve: How to notify other threads immediately when the sharedObject changes?

1. Based on Atomics.waitAsync + Atomics.notify?

// main.js
const { Worker } = require("worker_threads");
const {
  createObjectBuffer,
  getUnderlyingArrayBuffer,
} = require("@bnaya/objectbuffer");

const sharedMessage = createObjectBuffer(
  8 ** 5,
  {}, // some messageEntity you want
  {},
  "shared"
);
const sharedMessageArrayBuffer = getUnderlyingArrayBuffer(sharedMessage);

const sharedNotificationBuffer = new SharedArrayBuffer(4);
const sharedNotification = new Int32Array(sharedNotificationBuffer); // just for listen change

const worker = new Worker("./worker.js", {
  workerData: { sharedMessageArrayBuffer, sharedNotification },
});

async function listenSharedObjectChange() {
  while (true) {
    await Atomics.waitAsync(sharedNotification, 0, 0).value;
    // sharedMessage changed
  }
}

listenSharedObjectChange();

// worker.js
const { workerData } = require("worker_threads");
const { loadObjectBuffer } = require("@bnaya/objectbuffer");

const { sharedMessageArrayBuffer, sharedNotification } = workerData;
const sharedMessage = loadObjectBuffer(sharedMessageArrayBuffer);

setInterval(() => {
  // modify the sharedMessage
  sharedMessage.xxx = xxx;
  // notify the sharedMessage changed
  Atomics.notify(sharedNotification, 0);
}, 1000);

The test shows that when the worker executes Atomics.notify, the main thread can indeed end the waiting for Atomics.waitAsync. But once the Worker thread has high-frequency shared data changes, the main thread will miss the notification. For example:

// worker.js
sharedMessage.xxx = xxx;
Atomics.notify(sharedNotification, 0);
sharedMessage.xxx = yyy;
Atomics.notify(sharedNotification, 0);

For the continuous two shared data changes above, the main thread can only receive the first notification. The reason is that the second notify was issued before the main thread entered the next waitAsync loop. So if it is just to get notification of changes faster, is while(true) + waitAsync not as good as directly setInterval 0 + read sharedMessage? Like this:

// main.js
setInterval(() => {
  // try to read updated data
  sharedMessage.xxx;
}, 0);

The above is to discuss the listening scheme alone. Of course setInterval 0 will also miss changes. Is there a better solution? Looking forward to everyone's discussion!

The next problem to solve is how not to miss data changes?

I think it's time for us to use a message queue, just like IPC's postMessage. Here is an example:

// main.js
const { Worker } = require("worker_threads");
const {
  createObjectBuffer,
  getUnderlyingArrayBuffer,
} = require("@bnaya/objectbuffer");

const sharedMessage = createObjectBuffer(
  8 ** 5,
  { m: [] }, // the m is used for message queue
  {},
  "shared"
);

setInterval(() => {
  // cycling through unprocessed messages in the order they were produced
  while (sharedMessage.m.length > 0) {
    const newMsg = sharedMessage.m.shift();
    console.log("m", newMsg.msg);
  }
}, 0);

// worker.js
// always use push for new message
setInterval(() => {
  sharedMessage.m.push({ msg: "msg1" });
  sharedMessage.m.push({ msg: "msg2" });
}, 0);

I wonder this message queue-based solution should not have thread safety issues, because: when producing, always push, when consuming, always shift. In the above example, the expected output of the main thread should always be: msg1, msg2, msg1, msg2... But actually, it's wrong to read it, the actual result is as shown:

The cause of this problem may still be due to simultaneous reading and writing of shared data. I also noticed an issue the author previously mentioned, I wonder if it's because of it: #60

It seems that the method of push + shift is also not feasible now. I am preparing to try Circular_buffer. By the way, I would like to ask: assuming that push + shift is feasible, this kind of operation that frequently causes the length of sharedArray to change, isn't it less efficient than Circular_buffer on performance?

Looking forward to the author's reply and everyone's discussion and suggestions! @Bnaya

[Bnaya]

Hey @aweiu!
Thank you for taking the time to make this research and for your ideas
objectbuffer does not provide thread-safety and doesn't do any locking
Not that I don't want that, but a very difficult guarantee to achieve, and my time and knowledge are limited.

[Bnaya]

I will add some more thoughts:
If you run your code inside 2 worker threads i believe you could use Atomics.wait and not Atomics.waitAsync
Maybe you will get the behaviour you like

The limitations you are having with Atomics are maybe worth to discuss with nodejs guys

[aweiu]

If I use while (true)+Atomics.wait, my main thread cannot do anything else..
And I am currently trying the solution for Circular_buffer, fortunately there are not many problems left.

[aweiu]

@Bnaya Hi!

const sharedMessage = createObjectBuffer(
  8 ** 5,
  { m: new Array(10).fill({ bigObj: xxx }) },
  {},
  "shared"
);

// multiple reassignments
for (let i = 0; i < 10000; i++) {
  sharedMessage.m[1] = newObj1
  sharedMessage.m[2] = newObj2
  ...
}

May I ask, for the reassignment of array elements, does objectbuffer automatically release memory? For example, the bigObj and newObj in the above code.

At present, such array reassignment operations in my program seem to be unable to automatically release memory, no matter how large I set the size, it always runs for a while and then OutOfMemoryError.

[aweiu]

I had a thought, before the array object is overwritten, it might be persisted and stored elsewhere, which could be why the memory is not being released. Is there any way to effectively transform the objectbuffer object into a normal object? This way after storing the normal object, it probably should not affect memory release, but there might be performance issues again...🤔

[Bnaya]

@aweiu
you need to use processQueuedReclaims and also maybe queueReclaim if you don't want to trust finalizationregistry
You can see on these tests how do use them:
https://github.com/Bnaya/objectbuffer/blob/develop/src/apiTests/queueReclaim.test.ts
https://github.com/Bnaya/objectbuffer/blob/develop/src/externalPerspectiveTests/memoryGCTest.test.ts#L4