How to setup bidirectional communication with worker thread?

[sebhub]

I would like to use a separate thread which sends and receives data via a websocket. Apparently mg_mkpipe() is not the right thing to use: #1511.

What works is creating a worker thread and a socketpair(PF_UNIX, SOCK_STREAM, ...) in the MG_EV_WS_OPEN. What is missing is an API to create a connection from a user-specified socket (or file descriptor in general). I added something like this locally:

struct mg_connection *mg_mkfdconn(struct mg_mgr *mgr, int fd,
                                  mg_event_handler_t fn, void *fn_data) {
  struct mg_connection *c = mg_alloc_conn(mgr);
  if (c != NULL) {
    MG_DEBUG(("mkfdconn %d", fd));
    c->fd = S2PTR(fd);
    c->pfn = pf1;
    c->pfn_data = S2PTR(fd);
    c->fn = fn;
    c->fn_data = fn_data;
    mg_call(c, MG_EV_OPEN, NULL);
    LIST_ADD_HEAD(struct mg_connection, &mgr->conns, c);
  }
  return c;
}

In MG_EV_WS_OPEN I simply assign the new connection to the websocket connection: c->fn_data = mg_mkfdconn(...).

In the MG_EV_CLOSE I simply call mg_close_conn(c->fn_data). Due to SOCK_STREAM the worker thread notices that the manager thread closed the other side of the socket pair and can terminate itself.

[apuder]

Thanks for your code sniplet. Do you have a complete example somewhere? I would definitely encourage to add that as an example to Mongoose. The current web socket server example is not very helpful for how web sockets are usually used. Thx!

[cpq]

@sebhub can you elaborate on a use case when such thing is useful, please? Or even as @apuder suggested, share an example? Adding such function won't be an issue, provided it serves a relatively general use case.

[sebhub]

The use case is a JavaScript client which uses a WebSocket for communication with a server thread. The server thread may block on resources so the work cannot be carried out in the Mongoose thread. In MG_EV_WS_OPEN I create a socket pair:

  int sd[ 2 ];
  int rv = socketpair( PF_UNIX, SOCK_STREAM, 0, sd );
  if ( rv != 0 ) {
    mg_close_conn( c );
    return;
  }

  struct mg_connection *fdc =
    mg_mkfdconn( c->mgr, sd[ 0 ], callback, c );
  if ( fdc == NULL ) {
    (void) close( sd[ 0 ] );
    (void) close( sd[ 1 ] );
    mg_close_conn( c );
    return;
  }

  memcpy( c->label, LABEL, sizeof( LABEL ) );
  setfdc( c, fdc );

I had to use this hack to store connection-specific data:

static void setfdc( struct mg_connection *c, struct mg_connection *fdc )
{
  *(struct mg_connection **) &c->label[ FDC_OFFSET ] = fdc;
}

static struct mg_connection *getfdc( const struct mg_connection *c )
{
  return *(struct mg_connection **) &c->label[ FDC_OFFSET ];
}

In MG_EV_WS_MSG I simply forward the data to the server thread:

  struct mg_connection *fdc = getfdc( c );
  mg_send( fdc, wm->data.ptr, wm->data.len );

The server thread sends the data via normal system calls which will wake up the Mongoose thread from the select() invoking the callback:

static void callback(
  struct mg_connection *c,
  int ev,
  void *ev_data,
  void *fn_data
)
{
  struct mg_connection *ws = fn_data;

  if ( ev == MG_EV_READ ) {
    mg_ws_send( ws, (char *) c->recv.buf, c->recv.len, WEBSOCKET_OP_BINARY );
  } else if ( ev == MG_EV_CLOSE ) {
    /*
     * Sometimes the socket of the pair is closed before the websocket
     * connection close.  Clearing the label prevents a double close.
     */
    memset( ws->label, 0, sizeof( ws->label ) );
  }
}

This solution works quite well, however, is not really efficient since the data is copied multiple times. To avoid this we would need an alternative communication path which also wakes up the Mongoose thread from the select(). This would be easy with kqueue().

[cpq]

@sebhub Thank you.

A somewhat similar mechanism was employed in a recent versions of 7.x see https://github.com/cesanta/mongoose/blob/7.0/examples/multi-threaded/main.c . There, there was no special API, just communication over a socket pair. There was no mgr wakeup, but polling instead. The communication was bidirectional though, however over the UDP pair to preserve message boundaries.
Then, that was changed in 7.4 to the existing API.

The existing API does practically the very same thing as your code does: creates a socket pair, wraps one socket into the mg_connection, and saves its peer in the c->label. Then, a wrapped connection is given to a thread. Which, potentially, is a problem, since a wrapped connection may be closed by mongoose thread and another thread may end up with an invalidated pointer.

We can change the API and leave a wrapped connection in the mgr, and give back a raw socket peer to another thread. Then another thread can use raw socket API to read/write, with no fear of invalid pointers or other race conditions. Also, that would be bidirectional.
Take a look at the mt branch: https://github.com/cesanta/mongoose/tree/mt . 35be36a implemented that change, and usage example is at https://github.com/cesanta/mongoose/blob/mt/examples/multi-threaded/main.c Note: socket pair is UDP - that is easier to preserve message boundaries.

Would that work for your case?

Side note: your WS connection may terminate. An associated fdc must be signalled (ws pointer cleared), otherwise ws in callback() would point to garbage.

[sebhub]

Using mg_mkpipe() doesn't work in my use case: #1511. I also cannot use UDP since I need a reliable in-order byte stream.

Using a socket pair from socketpair() works quite well since one file descriptor is used by the Mongoose thread and the other is used by the server thread. They can close their own file descriptor individually. The threads don't share data.

The socket pair connection is closed by the socket callback in MG_EV_CLOSE (see above) or by the manager callback in MG_EV_CLOSE:

bool close_callback( struct mg_connection *c )
{
  if ( c->is_websocket && strcmp( c->label, LABEL ) == 0 ) {
    struct mg_connection *fdc = getfdc( c );
    mg_close_conn( fdc );
    return true;
  }

  return false;
}

[cpq]

@sebhub I think you did not get the point, and did not follow the change. The mt branch has a refactored mg_mkpipe

[sebhub]

I don't like the iteration over all connections in the callback.

Using a local variable to pass the file descriptor is broken and may lead undefined behaviour. Independent of this, when the worker thread uses the file descriptor it may be already closed or even worse reused for another file object.

[cpq]

You're correct about the local variable - amended PR, passing socket by value.
W.r.t. descriptor - changed mkpipe() and example so that the worker thread owns the socket, and thus must close it.
Also amended the example to avoid iteration, but interconnect parent and pipe connections via c->fn_data pointer. Alternatively, that could be done via the c->label.

Please take a look.

[sebhub]

The example looks now correct, however, it is still an unidirectional example. The worker thread is not able to receive data. Also the communication is done via UDP. I still think a more generic API which creates a connection for an arbitrary file descriptor is more useful.

Using c->fn_data in this example is not really helpful since it will break if HTTPS is used. I think this was the reason why I changed my code to use c->label.

[cpq]

We can't use a generic file descriptor, and rely on the sockepair syscall because it wont't work everywhere.
The API should be functional on all platforms that support BSD sockets, not only on unix-like OSs.
For example, on windows there is no socketpair. Also, you can't select() on non-socket fd on windows.

What's the issue with an UDP socketpair? On a loopback connection, an in-order data IO is guaranteed.
Also, UDP provides a nice property : if you send a data structure, you'll receive it un-fragmented.
So far I don't see any disadvantages in a UDP socketpair.
There might be one though: if you send data larger than a kernel socket buffer size, it might be dropped, I am not sure what send() call would return in that case.

FYI - branch amended, mg_socketpair()creates a TCP pair now. mk_mkpipe() returns a blocking TCP socket.

[sebhub]

If you return a file descriptor in mk_mkpipe(), then you already expose file descriptors to the API (through the return value). I don't know why using a file descriptor as a parameter in the proposed mg_mkfdconn() is an issue. You could change the parameter to void *fd to be in line with struct mg_connection.

[cpq]

@sebhub that question was already answered. We can't know what fd is passed to us. Is it a socket? Is it non-blocking? Can it be select-ed? Will it work on all systems with socket API?
In the suggested API, we know what we return - it is a blocking TCP socket, connected to a normal non-blocking Mongoose connection.

[sebhub]

Exactly, Mongoose doesn't know what kind of file descriptor is used to create the connection structure. That is the key point of this API. It offers more flexibility. It is up to the user to provide a file descriptor which works with the Mongoose event loop.

[cpq]

Added mg_wrapfd(), which gets used by mg_mkpipe().

[sebhub]

Thanks, it works with the mg_wrapfd(). I just had to call mg_iobuf_free( &c->recv ); in my callback. This was previously done by pf1().