multithreading - std::atomic を使用したリーダー/ライターの実装 (mutex フリー)

Question

以下は、ミューテックスと条件変数の代わりに std::atomics とビジー待機を使用して、リーダーとライターの間で同期する複数のリーダー/ライター共有データの試みです。そこにあるアサーションがなぜヒットしているのか、私は困惑しています。ロジックのどこかにバグがあることは確かですが、どこにあるのかはわかりません。

実装の背後にある考え方は、ライターが書き込みを完了するまで読み取りスレッドが回転しているということです。読み取り関数に入ると m_numReaders カウントが増加し、ライターを待機しているときは m_numWaiting カウントが増加します。

m_numWaiting が常に m_numReaders の後にインクリメントされ、m_numReaders の前にデクリメントされる場合、m_numWaiting は常に m_numReaders より小さいか等しい必要があるという考え方です。

m_numWaiting が m_numReaders よりも大きい場合 (または表示されていない場合) はありません。リーダーは常に最初にリーダー カウンターをインクリメントし、待機カウンターをインクリメントする場合があり、待機カウンターは常に最初にデクリメントされるためです。

それでも、アサートがヒットしているため、これが起こっているようです。もしあなたがそれを見たら、誰かが論理エラーを指摘できますか?

ありがとう！

#include <iostream>
#include <thread> 
#include <assert.h>

template<typename T>
class ReadWrite
{

public:

    ReadWrite() : m_numReaders(0), m_numWaiting(0), m_writing(false)
    {
        m_writeFlag.clear();
    }

    template<typename functor>
    void read(functor& readFunc)
    {
        m_numReaders++;
        std::atomic<bool>waiting(false);
        while (m_writing)
        {
            if(!waiting)
            {
                m_numWaiting++; // m_numWaiting should always be increased after m_numReaders
                waiting = true;
            }
        }

        assert(m_numWaiting <= m_numReaders);

        readFunc(&m_data);

        assert(m_numWaiting <= m_numReaders); // <-- These asserts get hit ?

        if(waiting)
        {
            m_numWaiting--; // m_numWaiting should always be decreased before m_numReaders
        }

        m_numReaders--;

        assert(m_numWaiting <= m_numReaders); // <-- These asserts get hit ?
    }

    //
    // Only a single writer can operate on this at any given time.
    //
    template<typename functor>
    void write(functor& writeFunc)
    {
        while (m_writeFlag.test_and_set());

        // Ensure no readers present
        while (m_numReaders);

        // At this point m_numReaders may have been increased !
        m_writing = true;

        // If a reader entered before the writing flag was set, wait for
        // it to finish
        while (m_numReaders > m_numWaiting);

        writeFunc(&m_data);

        m_writeFlag.clear();
        m_writing = false;
    }
private:
    T m_data;
    std::atomic<int64_t> m_numReaders;
    std::atomic<int64_t> m_numWaiting;
    std::atomic<bool> m_writing;
    std::atomic_flag m_writeFlag;
};

int main(int argc, const char * argv[])
{
    const size_t numReaders = 2;
    const size_t numWriters = 1;
    const size_t numReadWrites = 10000000;

    std::thread readThreads[numReaders];
    std::thread writeThreads[numWriters];

    ReadWrite<int> dummyData;

    auto writeFunc = [&](int* pData)    { return; }; // intentionally empty
    auto readFunc = [&](int* pData)     { return; }; // intentionally empty

    auto readThreadProc = [&]()
    {
        size_t numReads = numReadWrites;
        while (numReads--)
        {
            dummyData.read(readFunc);
        }
    };

    auto writeThreadProc = [&]()
    {
        size_t numWrites = numReadWrites;
        while (numWrites--)
        {
            dummyData.write(writeFunc);
        }
    };

    for (std::thread& thread : writeThreads)    { thread = std::thread(writeThreadProc);}
    for (std::thread& thread : readThreads)     { thread = std::thread(readThreadProc);}
    for (std::thread& thread : writeThreads)    { thread.join();}
    for (std::thread& thread : readThreads)     { thread.join();}
}