About using mg_ws_send on a separate thread

[vb-mich]

Hello, first of all congratulation for the project, as a C lover this is the ONLY web server that really match what I like to code with!

I have a pretty straightforward question: how safe it is to call mg_ws_send from another thread?
At the moment each thread handles one connection hence I'm mutexing the call just within the thread itself.
I know that on the very quick start guide is stated that "all mg * API functions are called from the same thread" but looking inside mg_ws_send I believe there's not much concern as long as the call is atomic in relation of a single connection. Am I wrong?

Cheers
Mich

[cpq]

@vb-mich , there are couple of things happen when you send data over the struct mg_connection *c pointer.

First, the data gets appended to the c->send buffer. Note, it is not gets send, it gets appended. So the first issue is to make sure no other thread deals with the send buffer. And you can't guarantee that, because the IO thread can send some remaining data and modify the c->send whilst your thread adding some stuff to it.

Another issue is that even if you add data, the IO thread does not know that the data was added - it may continue sleeping in mg_poll(). So your data gets sent whenever the main IO thread wakes up from select(), and that depends on the polling interval you've set in mg_mgr_poll() call and on the current network activity. Regardless, you'd like your data be sent immediately - so it is better to wake up the IO thread immediately after you've sent the data.

The proper solution is to use the "pipe" connection - mg_mkpipe() and mg_wakeup(), as described in the docs, see https://mongoose.ws/documentation/#mg_mkpipe . Also see the multithreaded example at https://github.com/cesanta/mongoose/tree/master/examples/multi-threaded .

Now, let's assume you have a bunch of WS connections from multiple clients and you have some thread that wants to push to all of these connections from time to time. Here's how you do it:

static void fn(struct mg_connection *c, int ev, void *ev_data, void *fn_data) {
 ...
}

// Pipe event handler. Called by the IO task
static void pcb(struct mg_connection *c, int ev, void *ev_data, void *fn_data) {
  if (ev == MG_EV_READ) {
    for (struct mg_connection *t = c->mgr->conns; t != NULL; t = t->next) {
       if (t->label[0] == 'W') mg_ws_send(t, ....);   // IMPORTANT: fn() must set the label on WS connections
    }
  }
}

static void my_task(struct mg_connection *c) {
...
   mg_mgr_wakeup(c);  // Something happened, wakeup mgr and let it call pcb handler
...
}

int main(int argc, char *argv[]) {
  ...
  start_thread(my_task, mg_mkpipe(&mgr, pcb, NULL)); // Start another thread and pass a pipe connection as a param
  mg_http_listen(&mgr, "http://localhost:8000", fn, &mgr);  // Start listener which creates WS connections
  for (;;) mg_mgr_poll(&mgr, 1000);
  ...
}

[vb-mich]

Hi @cpq , thank you very much for the detailed explanation, now I understand much better.

I've just tried the multi-threaded example and I've noticed that the pipe handler callback pcb is called as many time as mg_mgr_wakeup. So, let's say I have 10 threads and 10 active websocket connections, each thread works to provide data to 1 websocket connection in a producer-consumer type of exercise. When when a thread finishes then it calls mg_mgr_wakeup(c) and thread-safely put data to send into fn_data. Then pcb is called as many times as mg_mgr_wakeup(c) got called and for each time has to iterate through all the connections to find data to send even if all the data have been sent the first time where flags in t->label are reset, so all the other times pcb is called sounds a bit of a waste of resources. Perhaps I could keep another flag so that pcb returns if there's nothing to do.
Am I getting the usage of pipes wrong?

Thanks,
Mich

[cpq]

Use different pipes with different event handler functions if you want granularity.

[vb-mich]

Use different pipes with different event handler functions if you want granularity.

That was easy, silly me!
One last concern, when the ws connection closes then the thread dies, also the pipe connection should be free-ed as well.
Should the application explicitly tell the manager to close the pipe connection by setting pipe->is_closing = 1 or the library will figure it out?
I see from the code that the closing flag is already widely handled when polling the connection manager, but even if the thread that spawned the pipe connection no longer exists I'm afraid that the connection still exists and there's no reason for the manager to close it on its own.

Thanks,
Mich

[cpq]

Good question.
Yeah, the application should set pipe->is_closing = 1.

mg_mkpipe() creates a connection which is added to the manager and has a socketpair-ed socket connected to it. mg_mgr_wakeup sends a byte to that socket, which wakes up a manager and calls a connection's event handler. It looks like there is a leak - if a connection gets closed, then a socketpair-ed socket does not get closed.

We can make mg_mgr_wakeup() to accept a buffer to send - instead of sending a byte, a user can send some specific data, and pipe's connection handler is going to catch it.

[t123yh]

Hi,

As mg_mkpipe() is not implemented in the MIP stack, and also the FreeRTOS+TCP stack doesn't support communicating with 127.0.0.1, does this mean we have to use LwIP in order to use this feature? If I have to use MIP stack or FreeRTOS+TCP, how is it possible to use mg_ws_send from another thread?

Thank you!

[cpq]

@t123yh are you using an RTOS or writing a bare metal firmware ?

[t123yh]

I'm using FreeRTOS.

[cpq]

mg_mkpipe() was created to provide a cross-platform API for sending data between threads.
You can always use OS-specific API. For example, you can use MIP + FreeRTOS, and use FreeRTOS's queues to exchange data.
For example, at boot, you can create a worker thread and a queue. When request comes, a handler sends request data to the worker using a queue, and then checks for the answer in MG_EV_POLL.