Thread-safe message queue

Question

In my learning course, I've implemented a message queue to which data gets pushed by some thread and later gets processed by some other thread. My implementation isn't that efficient as it involves creating minimum three copies of the same data which is not acceptable. So is there any way to avoid these unnecessary copies? This is my sample working code:

#include <iostream>
#include <string>
#include <list>
#include <thread>
#include <mutex>
#include <condition_variable>

struct Data {
    std::string topic {};
    std::string msg {};

    Data(const std::string& topic, const std::string& msg) {
        this->topic = topic;
        this->msg = msg;
    }
};

std::mutex net_mutex {};
std::mutex pro_mutex {};

std::condition_variable net_cond_var {};
std::condition_variable pro_cond_var {};

std::list<Data> net_list;
std::list<Data> pro_list;

void pro_thread() {
    while (true) {
        std::unique_lock<std::mutex> ul(pro_mutex);

        pro_cond_var.wait(ul, [] () { return not pro_list.empty(); });
        Data data = pro_list.front(); // third copy
        pro_list.pop_front();

        ul.unlock();

        // do processing
    }
}

void relay_thread() {
    while (true) {
        // relays received network data to different processing threads based upon topic

        std::unique_lock<std::mutex> ul(net_mutex);

        net_cond_var.wait(ul, [] () { return not net_list.empty(); });
        Data data = net_list.front(); // second copy 
        net_list.pop_front();

        ul.unlock();

        if (data.topic == "A") { // push data into pro_list queue
            pro_mutex.lock();

            pro_list.emplace_back(data);
            pro_cond_var.notify_one();

            pro_mutex.unlock();
        }
    }
}

void net_thread() {
    while (true) {
        // receives data from socket and pushes into net_list queue

        Data data("A", "Hello, world!");
        net_mutex.lock();

        net_list.emplace_back(data); // first copy
        net_cond_var.notify_one();

        net_mutex.unlock();
    }
}

int main() {
    std::thread net(net_thread);
    std::thread relay(relay_thread);
    std::thread pro(pro_thread);

    net.join();
    relay.join();
    pro.join();
}

Answer 1

Create a class that implements the thread-safe queue

Your program has two queues, and already you see a lot of code duplication. It would be great to create a class that implements everything necessary for a thread-safe queue. For example:

template <typename T>
class ThreadSafeQueue {
    std::mutex mutex;
    std::condition_variable cond_var;
    std::queue<T> queue;

public:
    void push(T&& item) {
        {
            std::lock_guard lock(mutex);
            queue.push(item);
        }

        cond_var.notify_one();
    }

    T& front() {
        std::unique_lock lock(mutex);
        cond_var.wait(lock, [&]{ return !queue.empty(); });
        return queue.front();
    }

    void pop() {
        std::lock_guard lock(mutex);
        queue.pop(item);
    }
};

This way, the rest of the code now simplifies a lot:

ThreadSafeQueue<Data> net_queue;
ThreadSafeQueue<Data> pro_queue;
...
void relay_thread() {
    while (true) {
        Data &data = net_queue.front(); // no copy, just a reference

        if (data.topic == "A") {
            pro_queue.push(std::move(data)); // move the data to the other queue
        }

        net_queue.pop();
    }
}

There are a few other improvements I've put into the above code, which I'll discuss below.

Use std::queue for queues

The STL provides a type for queues: std::queue. Prefer to use that one over a "raw" std::list or other container. In particular, it enforces the properties of a queue: things can only be pushed in one end and popped from the other end.

Don't pop() a queue item until after you used it

In order to avoid copies, don't pop items from the queue until you've fully used them. front() returns a reference, so you can access the item that way. The only drawback of this approach is that this only allows one consumer thread per queue.

Alternatively, you could std::move() the item out of the queue, like so:

Data data = std::move(pro_list.front()); // move constructor used if available

Or if you use a std::list to store the queue items, you could use splice() to move a list entry to a temporary list; this will even work with types that can neither be copied nor moved. For example:

std::list<Data> temp_list;

{
    std::unique_lock lock(pro_mutex);
    pro_cond_var.wait(...);
    temp_list.splice(temp_list.begin(), pro_list, pro_list.begin());
    // no need to pop()
}

Data &data = temp_list.front();
// do processing

Call notify_one() without the mutex locked

It is generally more efficient to call notify_one() without the mutex being locked, otherwise the notified thread could be woken and immediately attempt to lock the still locked mutex as well.

Ensure the threads can terminate gracefully

Your main() function calls join() on all the threads it started, which is good, but unfortunately this means it will wait forever, since the threads themselves never return. In a real application you want to be able to terminate those threads gracefully. This requires waking up the threads that are blocked on cond_var.wait(), and having some way to signal that they should exit their loop, either using a separate flag, or by pushing a special item to the queue that signals that they should exit their loop.

Avoid manual locking of mutexes

I see you call net_mutex.lock() and net_mutex.unlock() in your code, but it's safer to use std::lock_guard(). You can limit the scope of the lock by using braces, like so:

Data data("A", "Hello, world!");

{
    std::lock_guard lock(net_mutex);
    net_list.emplace_back(data);
}

net_cond_var.notify_one();

When to use emplace_back()

In see this in your code:

Data data("A", "Hello, world!");
...
net_list.emplace_back(data); // first copy

Indeed, emplace_back() makes a copy here. There's nothing special emplace_back() can do here, you can call push_back() instead and it would be just as efficient. If the Data type has a move constructor, then push_back(std::move(data)) is also just as efficient as emplace_back(std::move(data)). Where emplace_back() shines is when you use it to construct the item directly in place, like so:

net_list.emplace_back("A", "Hello, world!");

This is the most efficient way to add an item to the list, as not having to move at all is faster than moving (and of course much faster than copying).

Implicitly generated move constructors

From the comments:

I've a doubt since struct Data doesn't implement neither move constructor nor move assignment operator. Does std::move make any sense?

Yes, the compiler can generate implicit constructors, including implicit move constructors. Whether this is done depends on whether you did not declare or delete any copy or move constructors/assignment operators yourself, and whether all the member variables have move constructors (obviously). Your struct Data satisfies all the requirements.