【问题标题】：How to implement lock-free counter with std::atomic?如何用 std::atomic 实现无锁计数器？
【发布时间】：2015-06-13 11:38:52
【问题描述】：

在我的程序中，多个线程（检查器）请求网页，如果这些页面包含一些数据，则另一个线程（消费者）处理数据。我只需要预定义的消费者数量就可以开始处理（不是全部）。我尝试使用 std::atomic 计数器和 fetch_add 来限制工作消费者的数量。但是，尽管计数器保持在界限内，但消费者获得相同的计数器值，并且实际处理消费者的数量超过了限制。行为取决于处理持续时间。简化代码包含 sleep_for 而不是获取页面和处理页面函数。

#include <iostream>
#include <thread>
#include <atomic>
#include <chrono>

class cConsumer
{
public:

    cConsumer::cConsumer(
        const size_t aNumber,
        std::atomic<bool> &aFire,
        std::atomic<size_t> &aCounter) :
        mNumber(aNumber),
        mFire(aFire),
        mCounter(aCounter){}

    void cConsumer::operator ()()
    {
        while (true)
        {
            while (!mFire.load()) std::this_thread::sleep_for(mMillisecond);

            size_t vCounter = mCounter.fetch_add(1);
            if (vCounter < 5)
            {
                std::cout << "      FIRE! consumer " << mNumber << ", counter " << vCounter << "\n";
                std::this_thread::sleep_for(mWorkDuration);
            }
            if (vCounter == 5)
            {
                mFire.store(false);
                mCounter.store(0);
            }
        }
    }

private:

    static const std::chrono::milliseconds 
        mMillisecond,
        mWorkDuration;

    const size_t mNumber;

    std::atomic<bool> &mFire;
    std::atomic<size_t> &mCounter;
};

const std::chrono::milliseconds 
    cConsumer::mMillisecond(1),
    cConsumer::mWorkDuration(1300);

class cChecker
{
public:

    cChecker(
        const size_t aNumber,
        std::atomic<bool> &aFire) :
        mNumber(aNumber),
        mFire(aFire),
        mStep(1){ }

    void cChecker::operator ()()
    {
        while (true)
        {
            while (mFire.load()) std::this_thread::sleep_for(mMillisecond);

            std::cout << "checker " << mNumber << " step " << mStep << "\n";
            std::this_thread::sleep_for(mCheckDuration);
            if (mStep % 20 == 1) mFire.store(true);         
            mStep++;
        }
    }

private:

    static const std::chrono::milliseconds 
        mMillisecond,
        mCheckDuration;

    const size_t mNumber;

    size_t mStep;

    std::atomic<bool> &mFire;
};

const std::chrono::milliseconds 
    cChecker::mMillisecond(1),
    cChecker::mCheckDuration(500);

void main()
{
    std::atomic<bool> vFire(false);
    std::atomic<size_t> vCounter(0);

    std::thread vConsumerThreads[16];

    for (size_t i = 0; i < 16; i++)
    {
        std::thread vConsumerThread((cConsumer(i, vFire, vCounter)));
        vConsumerThreads[i] = std::move(vConsumerThread);       
    }

    std::chrono::milliseconds vNextCheckerDelay(239);

    std::thread vCheckerThreads[3];

    for (size_t i = 0; i < 3; i++)
    {
        std::thread vCheckerThread((cChecker(i, vFire)));
        vCheckerThreads[i] = std::move(vCheckerThread);
        std::this_thread::sleep_for(vNextCheckerDelay);
    }

    for (size_t i = 0; i < 16; i++) vConsumerThreads[i].join();

    for (size_t i = 0; i < 3; i++) vCheckerThreads[i].join();
}

输出示例（部分）

...
checker 1 step 19
checker 0 step 20
checker 2 step 19
checker 1 step 20
checker 0 step 21
checker 2 step 20
checker 1 step 21
      FIRE! consumer 10, counter 0
      FIRE! consumer 13, counter 4
      FIRE! consumer 6, counter 1
      FIRE! consumer 0, counter 2
      FIRE! consumer 2, counter 3
checker 0 step 22
checker 2 step 21
      FIRE! consumer 5, counter 3
      FIRE! consumer 7, counter 4
      FIRE! consumer 4, counter 1
      FIRE! consumer 15, counter 2
      FIRE! consumer 8, counter 0
checker 1 step 22
      FIRE! consumer 9, counter 0
      FIRE! consumer 11, counter 1
      FIRE! consumer 3, counter 2
      FIRE! consumer 14, counter 3
      FIRE! consumer 1, counter 4
checker 0 step 23
checker 2 step 22
checker 1 step 23
checker 2 step 23
checker 0 step 24
checker 1 step 24

我找到了一种可行但并不优雅的解决方案：等待所有消费者尝试工作并了解火已熄灭。

#include <iostream>
#include <thread>
#include <atomic>
#include <chrono>

class cConsumer
{
public:

    cConsumer::cConsumer(
        const size_t aNumber,
        const size_t aConsumerCount,
        std::atomic<bool> &aFire,
        std::atomic<size_t> &aCounter) :
        mNumber(aNumber),
        mConsumerCount(aConsumerCount),
        mFire(aFire),
        mCounter(aCounter){}

    void cConsumer::operator ()()
    {
        while (true)
        {
            while (!mFire.load()) std::this_thread::sleep_for(mMillisecond);

            const size_t vCounter = mCounter.fetch_add(1);

            if (vCounter < 5)
            {
                std::cout << "      FIRE! consumer " << mNumber << ", counter " << vCounter << "\n";
                std::this_thread::sleep_for(mWorkDuration); //stub for process function
            }

            if (vCounter >= 5)
            {
                std::this_thread::sleep_for(mWorkDuration); //wait for other threads to increase counter
                std::this_thread::sleep_for(mWorkDuration); //double wait for long processing
                mFire.store(false);
            }

            if (vCounter == mConsumerCount)
            {               
                mCounter.store(0);
            }
        }
    }

private:

    static const std::chrono::milliseconds 
        mMillisecond,
        mWorkDuration;

    const size_t 
        mNumber,
        mConsumerCount;

    std::atomic<bool> &mFire;
    std::atomic<size_t> &mCounter;
};

const std::chrono::milliseconds 
    cConsumer::mMillisecond(1),
    cConsumer::mWorkDuration(1300);

class cChecker
{
public:

    cChecker(
        const size_t aNumber,
        std::atomic<bool> &aFire) :
        mNumber(aNumber),
        mFire(aFire),
        mStep(1){ }

    void cChecker::operator ()()
    {
        while (true)
        {
            while (mFire.load()) std::this_thread::sleep_for(mMillisecond);

            std::cout << "checker " << mNumber << " step " << mStep << "\n";
            std::this_thread::sleep_for(mCheckDuration);
            if (mStep % 20 == 1) mFire.store(true);         
            mStep++;
        }
    }

private:

    static const std::chrono::milliseconds 
        mMillisecond,
        mCheckDuration;

    const size_t mNumber;

    size_t mStep;

    std::atomic<bool> &mFire;
};

const std::chrono::milliseconds 
    cChecker::mMillisecond(1),
    cChecker::mCheckDuration(500);

void main()
{
    std::atomic<bool> vFire(false);
    std::atomic<size_t> vCouter(0);

    std::thread vConsumerThreads[16];

    for (size_t i = 0; i < 16; i++)
    {
        vConsumerThreads[i] = std::move(std::thread(cConsumer(i, 16, vFire, vCouter)));
    }

    std::chrono::milliseconds vNextCheckerDelay(239);

    std::thread vCheckerThreads[3];

    for (size_t i = 0; i < 3; i++)
    {
        vCheckerThreads[i] = std::move(std::thread(cChecker(i, vFire)));
        std::this_thread::sleep_for(vNextCheckerDelay);
    }

    for (size_t i = 0; i < 16; i++) vConsumerThreads[i].join();

    for (size_t i = 0; i < 3; i++) vCheckerThreads[i].join();

我认为存在更好的解决方案。

【问题讨论】：

输出中有问题的值在哪里？
我需要“开火！”只有 5 次，但发生了 15 次。
每个checker都有私有的step计数器，当发现是20*n+1时，消费者被重新武装（mFire.store(true)被执行）。这就是为什么你有 15 个“火！”事件 - 每个检查器 5 个（0,1 和 2），达到 step 21。
为什么你只期望 5 个？将 mCouter 设置为 5 的消费者会将其重置为 0，这也会导致其他人触发。

标签： c++ multithreading

【解决方案1】：

这里发生了什么？

运气好的话，一旦你着火了，可能会有超过 5 个工人通过这条线：

    while(!mFire.load()) std::this_thread::sleep_for(mMillisecond);

假设有 10 个工作人员处于唤醒状态，并且该计数器为 0。然后每 10 个工作人员将执行此操作：

    size_t vCounter = mCouter.fetch_add(1);

现在 10 个工人中的每一个都有一个介于 1 和 11 之间的不同计数器。第 5 个将执行 if 子句：

        if(vCounter < 5)

任何具有更高计数器的线程都将继续。其中第 6 个线程，将重置火并重置计数器：

        if(vCounter == 5)
        {
            mFire.store(false);
            mCouter.store(0);
            cout << "RESET!!!!!! by consume "<<mNumber << endl; // useful to understand
        }

然后所有这些中间线程将继续循环等待下一次火灾。

但是现在坏事可能会发生，因为你还有一些工人还在工作，而你有一堆跳棋等待再次放火：

while(mFire.load()) std::this_thread::sleep_for(mMillisecond);
...   // now that fire is reset, they will go on

有些可以达到以下行：

        if(mStep % 20 == 1) {
            mFire.store(true); 
            cout << "SET FIRE" << endl;   // to make the problem visual
        }

由于原子计数器为 0，您将立即拥有 5 个新工作人员，除了仍在运行的工作人员之外，他们还将开始一项新工作。

你能做些什么呢？

我不完全清楚你打算做什么：

你想让 5 名工人为每场新火激活吗？在这种情况下，就像你做的那样没关系。然后，工人总数可能超过 5 人。

您想在任何时候让最多 5 个工作人员处于活动状态吗？在这种情况下，您永远不应该像以前那样将工作人员的数量重置为 0，但您应该为所有增加它的线程递减计数器。因此 conter 将包含当前在火处理部分的线程数：

while(true)
{
    while(!mFire.load()) std::this_thread::sleep_for(mMillisecond);

    size_t vCounter = mCouter.fetch_add(1);   // FIRE PROCESSING: INCREMENT COUNTER
    if(vCounter < 5)
    {
        std::cout << "      FIRE! consumer " << mNumber << ", counter " << vCounter << "\n";
        std::this_thread::sleep_for(mWorkDuration);
        std::cout << "         finished consumer "<< mNumber<<endl;
    }
    if(vCounter == 5)
    {
        mFire.store(false);
        //mCouter.store(0);
        cout << "RESET!!!!!! by consumer "<<mNumber << endl; 
    }
    mCouter.fetch_sub(1);                    // END OF PROCESSING: DECREMENT COUNTER

【讨论】：

真正的火灾仅由跳棋产生，并且在时间上以相对较长的间隔（超过 30 秒长）分开。我的错误解决方案会产生额外的虚假火灾。我希望每场真正的独立火灾都有 5 名工人。

【解决方案2】：

可能的解决方案是为消费者完成标志使用辅助数组。当消费者完成处理时，它会将 true 存储到其完成的数组单元中。一个额外的控制线程扫描 done 数组以查找所有为 true 的单元格并重置程序状态。

#include <iostream>
#include <thread>
#include <atomic>
#include <chrono>

class cConsumer
{
public:

    cConsumer::cConsumer(
        const size_t aNumber,
        const size_t aFiresLimit,
        std::atomic<bool> &aFire,
        std::atomic<bool> &aDone,
        std::atomic<size_t> &aCounter) :
        mNumber(aNumber),
        mFiresLimit(aFiresLimit),
        mFire(aFire),
        mDone(aDone),
        mCounter(aCounter){}

    void cConsumer::operator ()()
    {
        while (true)
        {
            while (!mFire.load()) std::this_thread::sleep_for(mMillisecond);

            const size_t vCounter = mCounter.fetch_add(1);

            if (vCounter < mFiresLimit)
            {
                std::cout << "      FIRE! consumer " << mNumber << ", counter " << vCounter << "\n";
                std::this_thread::sleep_for(mWorkDuration); // instead real processing
            }   

            mDone.store(true);

            while (mDone.load()) std::this_thread::sleep_for(mMillisecond);
        }
    }

private:

    static const std::chrono::milliseconds 
        mMillisecond,
        mWorkDuration;

    const size_t 
        mNumber,
        mFiresLimit;

    std::atomic<bool> 
        &mFire,
        &mDone;

    std::atomic<size_t> &mCounter;
};

const std::chrono::milliseconds 
    cConsumer::mMillisecond(1),
    cConsumer::mWorkDuration(1300);

class cChecker
{
public:

    cChecker(
        const size_t aNumber,
        std::atomic<bool> &aFire) :
        mNumber(aNumber),
        mFire(aFire),
        mStep(1){ }

    void cChecker::operator ()()
    {
        while (true)
        {
            while (mFire.load()) std::this_thread::sleep_for(mMillisecond);

            std::cout << "checker " << mNumber << " step " << mStep << "\n";
            std::this_thread::sleep_for(mCheckDuration);
            if (mStep % 20 == 1) // dummy condition instead real checker function
            {
                mFire.store(true);
            }
            mStep++;
        }
    }

private:

    static const std::chrono::milliseconds 
        mMillisecond,
        mCheckDuration;

    const size_t mNumber;

    size_t mStep;

    std::atomic<bool> &mFire;
};

const std::chrono::milliseconds 
    cChecker::mMillisecond(1),
    cChecker::mCheckDuration(500);

class cController
{
public:

    cController(
        const size_t aConsumerCount,
        std::atomic<bool> &aFire,
        std::atomic<bool> * const aConsumersDone,
        std::atomic<size_t> &aCounter) :
        mConsumerCount(aConsumerCount),
        mFire(aFire),
        mConsumersDone(aConsumersDone),
        mCounter(aCounter){}

    void cController::operator ()()
    {
        while (true)
        {       
            while(!mFire.load()) std::this_thread::sleep_for(mMillisecond);

            bool vAllConsumersDone = false;

            while (!vAllConsumersDone)
            {
                size_t i = 0;
                while ((i < mConsumerCount) && (mConsumersDone[i].load())) i++;
                vAllConsumersDone = (i == mConsumerCount);
                std::this_thread::sleep_for(mMillisecond);
            }

            mFire.store(false);
            for (size_t i = 0; i < mConsumerCount; i++) mConsumersDone[i].store(false);
            mCounter.store(0);
        }
    }

private:

    const size_t mConsumerCount;

    static const std::chrono::milliseconds mMillisecond;

    std::atomic<bool> 
        &mFire,
        * const mConsumersDone;

    std::atomic<size_t> &mCounter;
};

const std::chrono::milliseconds cController::mMillisecond(1);

void main()
{
    static const size_t 
        vCheckerCount = 3,
        vConsumersCount = 16,
        vFiresLimit = 5;

    std::atomic<bool> vFire(false);

    std::atomic<bool> vConsumersDone[vConsumersCount];
    for (size_t i = 0; i < vConsumersCount; i++) vConsumersDone[i].store(false);

    std::atomic<size_t> vCounter(0);    

    std::thread vControllerThread(cController(vConsumersCount, vFire, vConsumersDone, vCounter));

    std::thread vConsumerThreads[vConsumersCount];

    for (size_t i = 0; i < vConsumersCount; i++)
    {
        vConsumerThreads[i] = std::move(std::thread(cConsumer(i, vFiresLimit, vFire, vConsumersDone[i], vCounter)));
    }

    std::chrono::milliseconds vNextCheckerDelay(239);

    std::thread vCheckerThreads[vCheckerCount];

    for (size_t i = 0; i < vCheckerCount; i++)
    {
        vCheckerThreads[i] = std::move(std::thread(cChecker(i, vFire)));
        std::this_thread::sleep_for(vNextCheckerDelay);
    }

    for (size_t i = 0; i < vConsumersCount; i++) vConsumerThreads[i].join();

    for (size_t i = 0; i < vCheckerCount; i++) vCheckerThreads[i].join();

    vControllerThread.join();
}

输出（部分）示例：

...
checker 2 step 19
checker 1 step 19
checker 0 step 19
checker 2 step 20
checker 0 step 20
checker 1 step 20
checker 2 step 21
checker 0 step 21
checker 1 step 21
      FIRE! consumer 11, counter 0
      FIRE! consumer 3, counter 2
      FIRE! consumer 4, counter 3
      FIRE! consumer 10, counter 4
      FIRE! consumer 14, counter 1
checker 0 step 22
checker 2 step 22
checker 1 step 22
checker 2 step 23
checker 0 step 23
checker 1 step 23
checker 2 step 24
checker 0 step 24

【讨论】：