Remove empty()

A big issue with thread-safe data structures is that they are going to be accessed by multiple threads. You also cannot assume in which order and what times threads will actually run. And while an operation might look atomic, the problem is that when you combine multiple atomic operations, the result is no longer atomic.

While empty() itself atomically checks whether the queue is empty, the answer it returns might no longer be true when the caller checks it. That matters even in a single-consumer queue, where there is only one thread calling empty(). Consider your main(): first you are starting 8 threads, and then you are checking if the queue is empty. You are assuming that the first time you call q.empty(), some of the producer threads have already got a time slice and pushed an item. But that's not necessarily the case; the operating system is free to delay starting those producer threads until after your main thread calls q.empty().

Note that even adding a while (q.empty()) {} right after starting the producer threads is not enough to be safe: your consumer thread might outpace the 8 producer threads at any time.

You should remove the member function empty(), because you just cannot trust the return value. You have to find some other way to signal that all your producers are done producing.

Consider making pop() return a std::optional<T>

Your push() moves a value into a new mpsc_queue_node. Since you are already paying that overhead, you could have pop() move the value from the node into the return value. And to handle the case of an empty queue, wrap this in a std::optional. That way, your while-loop in main() could be rewritten to:

while (auto value = q.pop()) {
    std::cout << "Popping " << *value << '\n';
}

It can look like:

std::optional<T> pop() {
    auto head_ptr = m_head.load(std::memory_order_acquire);
    if (head_ptr == nullptr) {
        return {}; // empty optional
    }
    …
    std::optional<T> value = std::move(head_ptr->value);
    delete head_ptr;
    return value;
}

First impressions - it looks generally good, and compiles cleanly with a comprehensive set of warnings - nice.

Minor - given that we declare copy construction and assignment, the move equivalents are automatically deleted, so no need to declare those.

push() should probably take its argument by value, so we can copy lvalues into the queue without having to manually invoke T's copy constructor. That would be in line with user expectations.

This test is unnecessary:

        mpsc_queue_node<T>* node =
            new mpsc_queue_node<T>{std::move(value), nullptr};
        if (!node) {
            throw std::bad_alloc{};
        }

The new operator cannot return a null pointer, and will throw std::bad_alloc itself.

I'm concerned about the special case in push() to an empty queue. If two threads arrive there, I don't see how we get the "loser" to re-try.

To show this, let's make each thread push to many queues, then check whether they all have the expected content:

#include <cstdlib>
#include <print>
int main()
{
    std::vector<mpsc_queue<int>> queues(10000);

    auto push_fn = [&](int i) {
        for (auto& q: queues) {
            q.push(i);
        }
    };

    std::vector<std::thread> threads(8);

    int i = 0;
    for (auto& t : threads) {
        t = std::thread(push_fn, i++);
    }
    for (auto& t : threads) {
        t.join();
    }
    unsigned failures = 0;
    for (auto& q: queues) {
        unsigned mask = 0;
        while (!q.empty()) {
            mask |= ( 1u << q.front()->value);
            q.pop();
        }
        if (mask != 0xff) {
            std::println(std::cerr, "{:08b}", mask);
            ++failures;
        }
    }

    if (failures) {
            std::println(std::cerr, "Got {} failure(s) from {} tests.",
                         failures, queues.size());
        return EXIT_FAILURE;
    }
}

Sample output:

10111111
10111111
10111111
10111111
01111111
11101111
11101111
11011111
01111111
11011111
Got 10 failure(s) from 10000 tests.

There's also something wrong when we push from a single thread:

#include <algorithm>
#include <array>
#include <cassert>
#include <cstdlib>
#include <thread>

int main()
{
    std::array<char, 1000000> a = {};
    mpsc_queue<char*> q;

    auto push_fn = [&]() {
        for (auto& c: a) {
            q.push(&c);
        }
    };

    auto t = std::thread(push_fn);

    for (auto i = 0u;  i < a.size();  ++i) {
        while (q.empty())
            ;
        *q.front()->value = 1;
        q.pop();
    }

    t.join();
    assert(q.empty());
    assert(std::ranges::all_of(a, std::identity{}));
}

Sometimes this segfaults from the while loop in push() - evidently m_tail becomes null while it's executing.

Some other times, the pop() loop doesn't complete - I haven't determined whether not all values are successfully pushed or whether they are not all retrieved (or both).

In the test program, we should show the need to continue fetching from the queue after joining all the threads (if we don't do that, we destroy the queue with elements still in it).

    for (auto& t : threads) {
        t.join();
    }
    while (!q.empty()) {
        std::cout << "Popping " << q.front()->value << '\n';
        q.pop();
    }

Also in the test program, there's a possibility of the threads becoming serialised due to hidden locks on std::cout (but I do get interrupted output lines between the string and the integer and the newline, so there's hope that's not interfering too much).

Missing functionality:

• We could do with a function for the consumer to wait for an item to become available. Polling front() is inefficient.

• It would be handy to have an emplace() function, for types that can't be moved.