如何用 std::atomic 实现无锁计数器？
Posted
技术标签:
【中文标题】如何用 std::atomic 实现无锁计数器？【英文标题】：How to implement lock-free counter with 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，这也会导致其他人触发。
【参考方案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
【讨论】：
以上是关于如何用 std::atomic 实现无锁计数器？的主要内容，如果未能解决你的问题，请参考以下文章
C++并发编程----无锁实现线程安全队列（《C++ Concurrency in Action》 读书笔记）
C++并发编程----无锁实现线程安全队列（《C++ Concurrency in Action》 读书笔记）
ARM 上的 std::atomic<bool> 无锁不一致（树莓派 3）