MPMC lock-free queue

Question

I'm pretty new to the world of lock-free programming and would love some feedback on this code. I have been testing the queue with multiple producers and consumers, and with no unexpected output (yet). Is there anything I could have missed or should improve?

template <typename Type, std::size_t Capacity = 100>
struct concurrent_queue
{
    static_assert(Capacity > 2, "not enough room in queue!");

    bool try_push(Type const & object) const
    {
        auto past_writer = writer.load();
        auto next_writer = increment(past_writer);

        if (next_writer != reader.load())
        {
            if (writer.compare_exchange_weak(past_writer, next_writer))
            {
                array[past_writer] = object;

                return true;
            }
        }

        return false;
    }

    bool try_push(Type & object) const
    {
        auto past_writer = writer.load();
        auto next_writer = increment(past_writer);

        if (next_writer != reader.load())
        {
            if (writer.compare_exchange_weak(past_writer, next_writer))
            {
                array[past_writer] = std::move_if_noexcept(object);

                return true;
            }
        }

        return false;
    }

    bool try_pop(Type & object) const
    {
        auto past_reader = reader.load();
        auto next_reader = increment(past_reader);

        if (past_reader != writer.load())
        {
            if (reader.compare_exchange_weak(past_reader, next_reader))
            {
                object = std::move_if_noexcept(array[past_reader]);

                return true;
            }
        }

        return false;
    }

    bool empty() const
    {
        return reader.load() == writer.load();
    }

private:

    std::size_t increment(std::size_t index) const
    {
        return (index + 1) % Capacity;
    }

    mutable std::atomic_size_t reader { 0 };
    mutable std::atomic_size_t writer { 0 };

    mutable std::array<Type, Capacity> array;
};

Answer 1

Welcome to Code Review. This queue implementation is not truly concurrent-ready.

Arbitrary initial capacity

template <typename Type, std::size_t Capacity = 100>

Why have you decided that the default size should be 100? Different people have different needs; having default values like this isn't a good idea because there is no true advantage to having a default size of 100.

While this is subjective, I suggest you remove that default size and just have users be required to specify the size that they want.

Compare/exchange confusion

There seems to be a misunderstanding in regards to the use of compare/exchange. The compare/exchange operations work as follows:

• Let X be the atomic variable
• Let E be the expected value of X
• Let N be the new value of X

The operations has the following signature:

X.compare_exchange(E, N);

This roughly translates (but is not an implementation of the compare/exchange operation) to:

bool compare_exchange(E, N)
{
    if( X == E )
    {
        X = N;
        return true;
    }
    else
    {
        E = X;
        return false;
    }
}

What this means is that when the comparison fails, E is updated to the current value of X. If the comparison succeeds, X is updated with the value of N. Therefore, you don't need to update X again! That is the whole point of a compare/exchange operation; you're basically saying: "I was the last to modify X."

Issue #1: Unsynchronized access to writer

In your code, you perform a store to the atomic variable with a copy of a value you last obtained from the update performed in the compare/exchange operation, but this is not atomic.

if (next_writer != reader.load())
{
    if (writer.compare_exchange_weak(past_writer, next_writer))
    {
        // thread can be stalled here and try_pop() will access garbage memory
        // since at this point writer != reader
        array[past_writer] = object; 
        return true;
    }
}

Imagine a first call to try_push() in thread A makes it inside the if (next_writer != reader.load()) statement and gets stalled just before executing array[past_writer] = object;. Now another thread B calls try_pop() and successfully gets past if (past_reader != writer.load()); the writer now has a value different than reader because of the compare/exchange operation.

However, the store of an element has not happened yet since thread A is still stalled. What is thread B going to pop? It's going to access whatever garbage memory is at that location.

Issue #2: Malfunctioning empty function

The load from reader and the load from writer followed by a comparison is not thread safe.

bool empty() const
{
    return reader.load() == writer.load();
}

Consider what happens if writer.load() returns value A, but before the expression is evaluated, writer is changed to some other value B. This could cause empty() to say that the queue is empty when it isn't or that it isn't when it actually is.

Conclusion

These are two of many issues that are currently in your code. I've decided to stop here since that's a pretty bad problem. This current implementation if not safe at all. I recommend you do some more reading and learning about thread interaction. Good luck.