Naive lock free work stealing queue

Question

Recently, I read a article about work stealing queue Job System 2.0: Lock-Free Work Stealing – Part 3: Going lock-free, and this is my c++11 naive implementation based on my understand of c++11 memory order model. This works on x86 , as x86 has a strong memory model, is there any issue running on other architecture (weak memory order)?

#include <atomic>

class Job;

struct WorkStealingQueue
{
    // only called by owner work thread
    void Push(Job* job) noexcept
    {
        // m_bottom -> stealing thread read, owner thread read, write.
        auto bottom = m_bottom.load(std::memory_order_relaxed);

        m_jobs[bottom & MASK] = job;

        // need  let stealing thread see the new job.
        m_bottom.fetch_add(1, std::memory_order_release);
    }

    // only called by owner worker thread 
    Job* Pop(void) noexcept
    {
        auto bottom = m_bottom.load(std::memory_order_relaxed);

        auto top = m_top.load(std::memory_order_acquire);

        if (bottom > top)
        {
            auto job = m_jobs[(bottom - 1) & MASK];

            if(top < (bottom - 1)) 
            {
                m_bottom.fetch_sub(1, std::memory_order_release);
                return job;   
            }
            if (m_top.compare_exchange_weak(
                top, top + 1,
                std::memory_order_release,
                std::memory_order_relaxed))
            {
                return job;
            } 
        }

        return nullptr;
    }

    // called by stealing thread, not owner thread
    Job* Steal(void) noexcept
    {
        auto top = m_top.load(std::memory_order_acquire);

        // Release-Acquire ordering, so the stealing thread see new job
        auto bottom = m_bottom.load(std::memory_order_acquire);

        if (bottom > top)
        {
            auto job = m_jobs[top & MASK];
            // check if other stealing thread stealing this work 
            // or owner thread pop this job.
            // no data should do sync, so use relaxed oreder
            if (m_top.compare_exchange_weak(
                top, top + 1,
                std::memory_order_release,
                std::memory_order_relaxed))
            {
                return job;
            }
        }

        return nullptr;
    }

private:

    static constexpr auto MAX_COUNT = 4096u;
    static constexpr auto MASK = MAX_COUNT - 1u;
    static_assert((MAX_COUNT & MASK) == 0, "the max number of job must be power of two.");

    Job* m_jobs[MAX_COUNT];
    std::atomic<unsigned int> m_bottom{ 0 }, m_top{ 0 };
};

Answer 1

Pop() has problem

Consider what happens if your queue is in the following state with two jobs:

m_top    = 0
m_bottom = 2

The owner thread calls pop(), and loads in to local variables:

top    = 0
bottom = 2

Now, before the owner thread does anything else, two stealer threads call steal(), leaving the state of the queue like this:

m_top    = 2
m_bottom = 2

When the owner thread continues inside of pop(), it will go into this case and return a job that no longer exists:

    if(top < (bottom - 1)) 
    {
        m_bottom.fetch_sub(1, std::memory_order_release);
        return job;   
    }

Your code deviated from the code you linked. In the linked code, the first thing that pop() did was to decrement m_bottom so that the stealer threads could not steal the bottom job.

Answer 2

is there any issue running on other architecture (weak memory order)?

Well, std::atomic isn't guaranteed by the standard to be lock free. It totally depends on the target CPU's architecture, and capabilities.

If there are single operations supported by the CPU to access and change specific values, they will be lock free. Otherwise any operations on std::atomic may need to use an implementation that involve a synchronization mechanism (semaphore or mutex).

For reference you can read

• std::atomic_is_lock_free, ATOMIC_xxx_LOCK_FREE how to determine that at compile time

and

• Are atomic variables lock-free?

Answer 3

Rewrite Pop() followed by reference article.

#include <atomic>

class Job;

struct WorkStealingQueue
{
    // only called by owner work thread
    void Push(Job* job) noexcept
    {
        // m_bottom -> stealing thread read, owner thread read, write.
        auto bottom = m_bottom.load(std::memory_order_relaxed);

        m_jobs[bottom & MASK] = job;

        // need  let stealing thread see the new job.
        m_bottom.fetch_add(1, std::memory_order_release);
    }

    // only called by owner worker thread 
    Job* Pop(void) noexcept
    {
        auto bottom = m_bottom.fetch_sub(std::memory_order_relaxed);

        auto top = m_top.load(std::memory_order_acquire);

        if (bottom >= top)
        {
            auto job = m_jobs[bottom  & MASK];

            if(top != bottom) 
            {
                return job;
            }
            auto top1 = top + 1;
            if (!m_top.compare_exchange_weak(
                top, top1,
                std::memory_order_release,
                std::memory_order_relaxed))
            {
                job = nullptr;
            } 
            m_bottom.store(top1);
            return job;
        }
        m_bottom.store(top, std::memory_order_release);
        return nullptr;
    }

    // called by stealing thread, not owner thread
    Job* Steal(void) noexcept
    {
        auto top = m_top.load(std::memory_order_acquire);

        // Release-Acquire ordering, so the stealing thread see new job
        auto bottom = m_bottom.load(std::memory_order_acquire);

        if (bottom > top)
        {
            auto job = m_jobs[top & MASK];
            // check if other stealing thread stealing this work 
            // or owner thread pop this job.
            // no data should do sync, so use relaxed oreder
            if (m_top.compare_exchange_weak(
                top, top + 1,
                std::memory_order_release,
                std::memory_order_relaxed))
            {
                return job;
            }
        }

        return nullptr;
    }

private:

    static constexpr auto MAX_COUNT = 4096u;
    static constexpr auto MASK = MAX_COUNT - 1u;
    static_assert((MAX_COUNT & MASK) == 0, "the max number of job must be power of two.");

    Job* m_jobs[MAX_COUNT];
    std::atomic<unsigned int> m_bottom{ 0 }, m_top{ 0 };
};