Thread safe container for queue

I have an application where I deal with sockets. In this application, I REALLY NEED to run send() to the same socket, from several different threads. To avoid errors or messing with the sent data, I want to create a kind of queue. For this queue, I need a container that will store the data.

I thought of using something like std::vector, where a thread will check at all times if there is any data in it, and if there is, it will send one item at a time, and remove each item after sending. The items would be added to this std::vector when it was necessary to send something to the socket, through the push_back function.

The problem is that I don't know how to do this safely, avoiding errors of any kind, as the std::vector would MOST LIKELY be accessed at the same time by different threads, whether to add, read or remove items.

If there's a better way to do this that isn't like what I'm talking about or doesn't use std::vector, I'm open to suggestions

Solution

First of all, building a queue is one of the rare times that you're probably justified using a container other than vector. std::deque is probably the most obvious choice, since it actually is a queue (with a couple extra things you don't care about for now).

Since you want to access it (and in particular, write to it) from multiple threads, you need do protect access to it. Multiple threads writing concurrently will give undefined behavior.

Anthony Williams has published the source for a pretty solidly written concurrent queue.

There are a few changes I prefer. First of all, I think the empty member function he provides is basically useless--almost any use of it is subject to race conditions (i.e., by the time you try to decide something based on what it's returned, that may have changed).

He also provides two pop functions:

try_pop: pops something immediately, or fails
wait_and_pop waits (indefinitely) for something to be present, then pops it

Personally, I prefer a pop that allows you to specify a timeout, and always returns a Boolean to indicate success/failure. If you specify a timeout of 0, it's equivalent to his try_pop. If you specify a long timeout, you can get pretty much the behavior of his wait_and_pop. But, importantly, you can also specify times in between those two extremes, which is often quite useful. I usually end up using a timeout somewhere in the range of 100 ms to 1 second or so.

For example, you'll probably have a thread that pops an item from the queue, and sends it out the socket. But you also want to be able to exit cleanly from that thread, so you want it to check its stop token once in a while, and exit clean if the stop token is set. So you end up with a loop something like this:

while (!stop_token) {
    if (the_queue.pop(item, timeout)) {
        socket.send(item);
    }
}

A timeout around the 0.1 to 1.0 second range balances prompt exit when asked to do so with minimal CPU usage to poll the stop token when it would otherwise be sitting idle.

Of course, that's not the only way to do things, but I find that it works well in practice.