CODEWITHSUNDEEP
c++multithreadingsocketsboost
rahul
rahul

Reputation: 2279

boost ASIO and message passing between thread

I am working on designing a websocket server which receives a message and saves it to an embedded database. For reading the messages I am using boost asio. To save the messages to the embedded database I see a few options in front of me:

  1. Save the messages synchronously as soon as I receive them over the same thread.
  2. Save the messages asynchronously on a separate thread.

I am pretty sure the second answer is what I want. However, I am not sure how to pass messages from the socket thread to the IO thread. I see the following options:

  1. Use one io service per thread and use the post function to communicate between threads. Here I have to worry about lock contention. Should I?
  2. Use Linux domain sockets to pass messages between threads. No lock contention as far as I understand. Here I can probably use BOOST_ASIO_DISABLE_THREADS macro to get some performance boost.

Also, I believe it would help to have multiple IO threads which would receive messages in a round robin fashion to save to the embedded database.

Which architecture would be the most performant? Are there any other alternatives from the ones I mentioned?

A few things to note:

Upvotes: 6

Views: 3951

Answers (5)

rahul
rahul

Reputation: 2279

After benchmarking/profiling I found the facebook folly implementation of MPMC Queue to be the fastest by at least a 50% margin. If I use the non-blocking write method, then the socket thread has almost no overhead and the IO threads remain busy. The number of system calls are also much less than other queue implementations.

The SPSC queue with cond variable in boost is slower. I am not sure why that is. It might have something to do with the adaptive spin that folly queue uses.

Also, message passing (UDP domain sockets in this case) turned out to be orders of magnitude slower especially for larger messages. This might have something to do with copying of data twice.

Upvotes: 1

call me Steve
call me Steve

Reputation: 1727

Assuming that the constraint of using cxx11 is not too hard in your situtation, I would try to use the std::async to make an asynchronous call to the embedded DB.

Upvotes: 0

rhashimoto
rhashimoto

Reputation: 15841

I don't think you want to use a unix socket, which is always going to require a system call and pass data through the kernel. That is generally more suitable as an inter-process mechanism than an inter-thread mechanism.

Unless your database API requires that all calls be made from the same thread (which I doubt) you don't have to use a separate boost::asio::io_service for it. I would instead create an io_service::strand on your existing io_service instance and use the strand::dispatch() member function (instead of io_service::post()) for any blocking database tasks. Using a strand in this manner guarantees that at most one thread may be blocked accessing the database, leaving all the other threads in your io_service instance available to service non-database tasks.

Why might this be better than using a separate io_service instance? One advantage is that having a single instance with one set of threads is slightly simpler to code and maintain. Another minor advantage is that using strand::dispatch() will execute in the current thread if it can (i.e. if no task is already running in the strand), which may avoid a context switch.

For the ultimate optimization I would agree that using a specialized queue whose enqueue operation cannot make a system call could be fastest. But given that you have network i/o by producers and disk i/o by consumers, I don't see how the implementation of the queue is going to be your bottleneck.

Upvotes: 2

Dave M.
Dave M.

Reputation: 1537

void Consumer( lockfree::queue<uint64_t> &message_queue ) {
    // Connect to database...
    while (!Finished) {
        message_queue.consume_all( add_to_database ); // add_to_database is a Functor that takes a message
        cond_var.wait_for( ... ); // Use a timed wait to avoid missing a signal.  It's OK to consume_all() even if there's nothing in the queue.
    }
}

void Producer( lockfree::queue<uint64_t> &message_queue ) {
    while (!Finished) {
        uint64_t m = receive_from_network( );
        message_queue.push( m );
        cond_var.notify_all( );
    }
}

Upvotes: 0

Dave M.
Dave M.

Reputation: 1537

You probably only need one io_service -- you can create additional threads which will process events occurring within the io_service by providing boost::asio::io_service::run as the thread function. This should scale well for receiving 8-byte messages from clients over the network socket.

For storing the messages in the database, it depends on the database & interface. If it's multi-threaded, then you might as well just send each message to the DB from the thread that received it. Otherwise, I'd probably set up a boost::lockfree::queue where a single reader thread pulls items off and sends them to the database, and the io_service threads append new messages to the queue when they arrive.

Is that the most efficient approach? I dunno. It's definitely simple, and gives you a baseline that you can profile if it's not fast enough for your situation. But I would recommend against designing something more complicated at first: you don't know whether you'll need it at all, and unless you know a lot about your system, it's practically impossible to say whether a complicated approach would perform any better than the simple one.

Upvotes: 0

Related Questions