Pub/Sub across threads, can the library handle it?

[PaulMoore-PaymentsFoundry]

When the server is running with a single instance, everything appears fine.

If we spin up multiple threads using workers, pub/sub fails... presumably because each thread is isolated and cannot see messages in other threads.

With that in mind, what's the most appropriate way to handle this, in terms of scaling? Even if we scale horizontally with a single instance on each machine, neither machine can communicate without some method of passing data between. I'm hoping this can be achieved without a stateful DB or similar.

Any advice would be much appreciated.

[hst-m]

You can send messages between workers https://nodejs.org/api/worker_threads.html#worker_threads_class_messagechannel

You could use websockets to send between machines. Use a websocket client in each server to connect to the other uWebSocket servers https://www.npmjs.com/package/ws This would be good reason to add websocket client to uWebSockets but that is not included yet so you have to use different client library https://github.com/uNetworking/uWebSockets/issues/791

redis pub/sub as mentioned below is probably better option to connect servers, this requires a central server to broadcast messages

[jbenguira]

You can send messages between workers https://nodejs.org/api/worker_threads.html#worker_threads_class_messagechannel

You could use websockets to send between machines. Use a websocket client in each server to connect to the other uWebSocket servers https://www.npmjs.com/package/ws This would be good reason to add websocket client to uWebSockets but that is not included yet so you have to use different client library uNetworking/uWebSockets#791

I'm doing exactly this (workers communication) to publish the messages on all workers. It's working well with a lag below 1ms between workers

[uasan]

Another option, Redis his pub/sub
#259

[PaulMoore-PaymentsFoundry]

You can send messages between workers https://nodejs.org/api/worker_threads.html#worker_threads_class_messagechannel
You could use websockets to send between machines. Use a websocket client in each server to connect to the other uWebSocket servers https://www.npmjs.com/package/ws This would be good reason to add websocket client to uWebSockets but that is not included yet so you have to use different client library uNetworking/uWebSockets#791

I'm doing exactly this (workers communication) to publish the messages on all workers. It's working well with a lag below 1ms between workers

Thanks for your reply.

Do you have an example of this please? Would be much appreciated.

[jbenguira]

Sure, here it is:

first of all I override the publish function of the app to change the behavior, instead of directly publishing, I publish + postMessage to the parent process controlling all workers

//override publish to allow multithread replication of publish()
                var uApp = {
                    publish: function(channel, msg){
                        
                        //publish on the current Thread
                        app.publish(channel, msg);

                        //send a copy to other threads
                        if ( parentPort != null ){
                            var clusteredProcessIdentifier = require('os').hostname() + "_" + require('worker_threads').threadId;
                            var content = msg;
                            var obj = { type: "CG_WS_MSG", channel: channel, message: content, source: clusteredProcessIdentifier };
                            parentPort.postMessage(obj);
                        }
                    }
                }

Then each worker will receive this message and process it like this:

var parentPort = require('worker_threads').parentPort;
parentPort.on('message', (msg) => {

                var clusteredProcessIdentifier = require('os').hostname() + "_" + require('worker_threads').threadId;
                if ( msg.source == clusteredProcessIdentifier ){
                    //same worker, let's discard it!
                    //console.log("Thread replication discarded because it's from the same origin!");
                }
                else if ( msg.type == "CG_WS_MSG" ){
                    app.publish(msg.channel, msg.message);
                    //console.log("msg published in this worker!");
                }
                
            });

[PaulMoore-PaymentsFoundry]

@jbenguira That's a huge help, thank you.

[uasan]

postMessage is good, but as soon as you have more than one server, you will need Redis or its equivalent, and if you are already sending to Redis you will receive messages in all subscribed threads.

[jbenguira]

Best approach is to do both, postMessage for inter worker communication and websockets for inter server communication