Subscription and Publish in a multi-threaded ws server...

[vidcentum]

Hi,

usockets, uWebSockets are very helpful and easy to use libraries. Thank you.

I am trying to get handle on using uWebSockets in a multi-thread server environment
Ref: examples/EchoServerThreaded.cpp

Please see the sample code below.

The server runs and accepts ws connections from clients. I connected 4 browser clients (for testing) successfully.

1. I sent an "alert" message that results in "broadcast" of the message. I observed the broadcast message was not received by the clients.

2. I modified the server to run a single thread and repeated the above test. Now the publish and subscription works well.

Is there any way to publish to all clients in a multi-threaded server? Could you please advice?

int main() {
    /* ws->getUserData returns one of these */
    struct PerSocketData {    };
    std::vector<std::thread *> threads(std::thread::hardware_concurrency());
    std::transform(threads.begin(), threads.end(), threads.begin(), [](std::thread *t) {  
      return new std::thread([]() {
          /* Very simple WebSocket broadcasting echo server */
        uWS::App().ws<PerSocketData>("/*", {      
            .open = [](auto *ws, auto *req) {
                /* Let's make every connection subscribe to the "broadcast" topic */
                ws->subscribe("broadcast");
            },
            .message = [](auto *ws, std::string_view message, uWS::OpCode opCode) {
                /* Exit gracefully if we get a closedown message (ASAN debug) */
                if (message == "closedown") {
                  /* Bye bye */
                  us_listen_socket_close(0, listen_socket);
                  ws->close();
                }
                if (message == "alert") {
                  /* Simply broadcast alert message we get */
                  ws->publish("broadcast", message, opCode);
                }
                else {
                  ws->send(message, opCode);
                }
            },
        }).listen(9001, [](auto *token) {
            listen_socket = token;
        }).run();
      });
    });
    std::for_each(threads.begin(), threads.end(), [](std::thread *t) {
        t->join();
    });
}

[ghost]

This should be added to the user manual under Scaling up,

Every thread has its own Loop, and uWS::Loop::get() returns the Loop for current thread.

Loop::defer(function) is the only thread-safe function here; you use it to defer the calling of a function to the thread that runs the Loop. Use this to signal every Loop that it should publish some message.

[vidcentum]

Thank you, Alex Hultman

[vidcentum]

Thank you, Alex Hultman. Could you please provide an example to use Loop::defer(function) in this context in the FAQ / user manual of Scaling up?

[vidcentum]

Hi,

I have created a sample program that may be added in the FAQ / Scaling up manual.

I am stuck at how to trigger Loop::defer() that call publish(). Could you please advice?

// Refer to https://github.com/uNetworking/uWebSockets/issues/888

struct worker_t
{
  void work();
  struct uWS::Loop *loop_;
  struct us_listen_socket_t *listen_socket_;
  std::thread *thread;
};

// To hold a few worker threads (std::thread::hardware_concurrency)
std::vector<worker_t> workers;

// Thread function
void worker_t::work()
{
    // save uWS::Loop in the worker_t object.
    loop_ = uWS::Loop::get();
    
    /* Very simple WebSocket broadcasting server */
    uWS::App().ws<PerSocketData>("/*", {
        .open = [](auto *ws, auto *req) {
            /* Let's make every connection subscribe to the "broadcast" topic */
            ws->subscribe("broadcast");
        },
        // Raise SIGUSR1 interrupt. The interrupt handler will trigger "broadcast" message.
        .message = [](auto *ws, std::string_view message, uWS::OpCode opCode) {
            if (message == "alert") {
              /* Simply broadcast alert message we get */
              std::raise(SIGUSR1);
            }
        }
        // Other handlers go here.
    }).listen(9001, [this](auto *token) {
        listen_socket_ = token;
        if (listen_socket_) {
          std::cout << "Thread " << std::this_thread::get_id() << " listening on port " << 9001 << std::endl;
        }
        else{
          std::cout << "Thread " << std::this_thread::get_id() << " failed to listen on port " << 9001 << std::endl;
        }
    }).run();
}

// Signal handler to broadcast message (to subscribers)
void onSignal (int signal)
{
    // SIGUSR1 is recived. 
    std::for_each(workers.begin(), workers.end(), [](worker_t &w) {
        w.loop_->defer([w]() {
            // Trigger each loop to send a broadcast message to the subscribers.
            // TBD
        });
    });
}

//
int main() {
  workers.resize(std::thread::hardware_concurrency());
  // Install signal handler.
  std::signal(SIGUSR1, onSignal);

  // Create multi-threaded server.
  std::transform(workers.begin(), workers.end(), workers.begin(), [](worker_t &w) {
    w.thread = new std::thread([&w]() {
      w.work();
    });
    return w;
  });
  
  std::for_each(workers.begin(), workers.end(), [](worker_t &w) {
      w.thread->join();
      delete w.thread;
  });
  
  std::cout << "Main exiting" << std::endl;  
  return 0;
}

[ghost]

You don't need to involve signals. I meant "signal" in an abstract sense.

Loop::defer is really all you need

[vidcentum]

Thank you, Alex Hultman. Yes, I tried without SIGUSR1 'signals'. I am able to call it Loop::defer from .message handler.

I am not able to follow how to publish from / within the lambda of w.loop_->defer({ }); I got the mechanism of Loop::defer . It adds the callback (the lambda) and wakesup the loop. So, the lambda shall have the socket context and call publish on it. I got stuck there.

Sorry to bother you - I will prepare the FAQ / Scaling up manual to contribute to the project.

// .message handler 
std::for_each(workers.begin(), workers.end(), [](worker_t &w) {
        w.loop_->defer([w]() {
            // Trigger each loop to send a broadcast message to the subscribers.
            // TBD
        });
    });

[ghost]

Yes it becomes a bit tricky since you need to capture the uWS::App inside a lambda that constructs the app itself.

You need to capture the address of the stack allocated app (or any such solution):

uWS::App app = uWS:App().ws(
...
[&app]() {
app.publish(...
...
)...

Or just put the app in global namespace with thread_local scope

[vidcentum]

Thank you, Alex Hultman. I am able to get uWS::App to the Loop::defer lambda function.
Now I am able to test the broadcast messages from a multi-threaded server.

Thank you for your guidance. I will prepare a note on this for FAQ / Scaling up and send it to you.

    app_ = new uWS::App();  // app_ is saved to worker_t. 
    /* Very simple WebSocket broadcasting echo server */
    app_ = new uWS::App();
    app_->ws<PerSocketData>("/*", { ... }).listen(...)

Best Regards
Maruthi

[vidcentum]

Thank you, Alex Hultman. Closing the issue.

Best Regards
Maruthi

[vidcentum]

Scaling up

One event-loop per thread, isolated and without shared data. That's the design here. Just like Node.js, but instead of per-process, it's per thread (well, obviously you can do it per-process also).

If you want to, you can simply take the previous example, put it inside of a few std::thread and listen to separate ports, or share the same port (works on Linux). More features like these will probably come, such as master/slave set-ups but it really isn't that hard to understand the concept - keep things isolated and spawn multiple instances of whatever code you have.

Recent Node.js versions may scale using multiple threads, via the new Worker threads support. Scaling using that feature is identical to scaling using multiple threads in C++.

Sample Code

This sample code demonstrates uWS::Loop per thread and signaling each thread - for example broadcasting a message to all the clients serving in all the threads. It demonstrates using thread safe uWS::Loop::defer (function).

/* 
   This example demonstrates basic idea of handling multi-threaded uWebSockets sever.
   There may be better ways to capture and use uWS::Loop and uWS::App objects. 
*/

#include <thread>
#include <vector>
#include <memory>
#include "uWebSockets/App.h"

/* ws->getUserData returns one of these */
struct PerSocketData {
  std::string_view user_secure_token;
};

/* uWebSocket worker runs in a separate thread */
struct worker_t
{
  void work();
  
  /* uWebSocket worker listens on separate port, or share the same port (works on Linux). */
  struct us_listen_socket_t *listen_socket_;

  /* Every thread has its own Loop, and uWS::Loop::get() returns the Loop for current thread.*/
  struct uWS::Loop *loop_;

  /* Need to capture the uWS::App object (instance). */
  std::shared_ptr<uWS::App> app_;

  /* Thread object for uWebSocket worker */
  std::shared_ptr<std::thread> thread_;
};

/* uWebSocket workers. */
std::vector<worker_t> workers;

/* uWebSocket worker thread function. */
void worker_t::work()
{
  /* Every thread has its own Loop, and uWS::Loop::get() returns the Loop for current thread.*/ 
  loop_ = uWS::Loop::get();

  /* uWS::App object / instance is used in uWS::Loop::defer(lambda_function) */
  app_ =  std::make_shared<uWS::App>();
  
  /* Very simple WebSocket broadcasting echo server */
  app_->ws<PerSocketData>("/*", {
    /* Settings */
    .compression = uWS::SHARED_COMPRESSOR,
    .maxPayloadLength = 16 * 1024 * 1024,
    .idleTimeout = 0,
    .maxBackpressure = 1 * 1024 * 1204,
    /* Handlers */
    // Refer to  6735bad  commit. Don't pass req in open handler
    // https://github.com/uNetworking/uWebSockets/commit/625efbc499460a21d27c5811ce7c948a69a04eea
    .open = [](auto *ws) {
      /* Let's make every connection subscribe to the "broadcast" topic.
         Since our example server runs multiple worker threads, the client connection
         may be served by any of the worker thread. 
         However, this example demonstrates publishing message(s) to all subscribers 
         served by all workers (threads).
      */
      ws->subscribe("broadcast");
    },
    .message = [this](auto *ws, std::string_view message, uWS::OpCode opCode) {
      /* Exit gracefully if we get a closedown message (ASAN debug) */
      if (message == "closedown") {
        /* Bye bye */
        us_listen_socket_close(0, listen_socket_);
        ws->close();
      }
      /* When an "alert" message is recieved on any worker thread, it simply invokes
         uWS::Loop::defer(funcion) of each thread. 
         Loop::defer(function) is the only thread-safe function here; you use it to defer 
         the calling of a function to the thread that runs the Loop. Use this to signal 
         every Loop that it should publish some message.
      */
      if (message == "alert") {
        /* Simply broadcast alert message we get */
        std::for_each(::workers.begin(), ::workers.end(), [message, opCode](worker_t &w) {
          /* uWs::Loop of each worker. */
          w.loop_->defer([w, message, opCode]() {
              /* uWs::App of each worker. uWs::App object has knowledge of all socket contexts. */
              w.app_->publish("broadcast", message, opCode);
          });
        });
      }
      /* Simple echo the message to the client. */
      else {
        ws->send(message, opCode);
      }
    },
    .drain = [](auto *ws) {
      /* Check getBufferedAmount here */
    },
    .ping = [](auto */*ws*/, std::string_view) {
          /* Not implemented yet */
    },
    .pong = [](auto */*ws*/, std::string_view) {
          /* Not implemented yet */
    },
    .close = [](auto *ws, int code, std::string_view message) {
      /* We automatically unsubscribe from any topic here */
    }
  }).listen(9001, [this](auto *token) {
    listen_socket_ = token;
    if (listen_socket_) {
      std::cout << "Thread " << std::this_thread::get_id() << " listening on port " << 9001 << std::endl;
    }
    else{
      std::cout << "Thread " << std::this_thread::get_id() << " failed to listen on port " << 9001 << std::endl;
    }
  }).run();

  std::cout << "Thread " << std::this_thread::get_id() << " exiting" << std::endl;
}

/* Main */
int main() {
  
  workers.resize(std::thread::hardware_concurrency());
  
  std::transform(workers.begin(), workers.end(), workers.begin(), [](worker_t &w) {
    w.thread_ = std::make_shared<std::thread>([&w]() {
      /* create uWebSocket worker and capture uWS::Loop, uWS::App objects. */
      w.work();
    });
    return w;
  });
  
  std::for_each(workers.begin(), workers.end(), [](worker_t &w) {
      w.thread_->join();
  });
  
  return 0;
}

[mfallavol]

make.log

I attached the entire make log but I think this is the pertinent entry:

examples/uWS_mt.cpp: In member function ‘void worker_t::work()’:
examples/uWS_mt.cpp:105:4: error: no matching function for call to ‘uWS::TemplatedApp::ws(const char [3], )’
}).listen(9001, [this](auto token) {
^
In file included from examples/uWS_mt.cpp:9:
src/App.h:152:20: note: candidate: ‘template uWS::TemplatedApp&& uWS::TemplatedApp::ws(std::__cxx11::string, uWS::TemplatedApp::WebSocketBehavior&&) [with UserData = UserData; bool SSL = false]’
TemplatedApp &&ws(std::string pattern, WebSocketBehavior &&behavior) {
^~
src/App.h:152:20: note: template argument deduction/substitution failed:
examples/uWS_mt.cpp:105:4: note: cannot convert ‘{SHARED_COMPRESSOR, ((16 * 1024) * 1024), 0, ((1 * 1024) * 1204), worker_t::work()::<lambda(auto:11)>{}, worker_t::work()::<lambda(auto:12*, std::string_view, uWS::OpCode)>{((worker_t*)this)}, worker_t::work()::<lambda(auto:13*)>{}, worker_t::work()::<lambda(auto:14*)>{}, worker_t::work()::<lambda(auto:15*)>{}, worker_t::work()::<lambda(auto:16*, int, std::string_view)>{}}’ (type ‘’) to type ‘uWS::TemplatedApp::WebSocketBehavior&&’
}).listen(9001, [this](auto *token) {
^

[mfallavol]

uWS_mt.txt
I just uploaded my source file just in case something got mis-copied.

[vidcentum]

@mfallavol The example code compiled on Linux Ubuntu 20.04 with GCC-9 -std=c++17 without any errors.
Library Code: uWebSockets v18.20.0
Host: Ubuntu 20.04
Compiler: GCC-9 / Options -std=c++17

Note: I did not compile using C++20.

Also, please ensure to supply required flags for uSockets (https://github.com/uNetworking/uSockets).

[vidcentum]

@mfallavol The example code compiled on Linux Ubuntu 20.04 with GCC-9 -std=c++17 without any errors.

Library Code: uWebSockets v18.22.0 and the latest of uSockets.
Host: Ubuntu 20.04
Compiler: GCC-9 / Options -std=c++17
Also, please ensure to supply required flags for uSockets (https://github.com/uNetworking/uSockets).

Note: I did not compile using C++20.

[mfallavol]

Thanks. I was using Ubuntu 18.04 and gcc-8. I upgraded to gcc-9 and it compiled although with a lot of warnings within uWebsockets.