2
\$\begingroup\$

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;
};
\$\endgroup\$
  • \$\begingroup\$ Is there anything I could have missed Well yes probably. The real thing you need to watch with lock free programming is throughput. Just because it is lock free does not mean it is faster. Have you written a locking version to test throughput characteristics against? \$\endgroup\$ – Martin York Oct 12 '15 at 1:58
3
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ you beat me to the explanation about CAS \$\endgroup\$ – Koby Becker Oct 12 '15 at 1:08
  • \$\begingroup\$ Thank you you for your answer, I figured it out after seeing I was missing a few elements. Removed the second store and everything is accounted for, see my latest edit. \$\endgroup\$ – Grandstack Oct 12 '15 at 1:15
  • \$\begingroup\$ @Grandstack The issue I describe is still present. I've added a comment to indicate where a thread can stall. \$\endgroup\$ – cr_oag Oct 12 '15 at 1:22
  • \$\begingroup\$ Thank you for your patience, I see the problems now. Will post again when I sort this out. \$\endgroup\$ – Grandstack Oct 12 '15 at 1:53

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.