【发布时间】：2015-03-10 10:44:48
【问题描述】：
我有以下代码，它从一开始就启动多个线程（一个线程池）（startWorkers()）。随后，在某个时候，我有一个装满myWorkObject 实例的容器，我想同时使用多个工作线程来处理它。 myWorkObject 在内存使用方面与另一个完全隔离。现在让我们假设 myWorkObject 有一个方法 doWorkIntenseStuffHere() 需要一些 cpu 时间来计算。
在对以下代码进行基准测试时，我注意到此代码不能很好地随线程数扩展，并且初始化/同步工作线程的开销超过了多线程的好处，除非有 3-4 个线程处于活动状态。我已经研究了这个问题并阅读了关于错误共享问题的信息，并且我认为我的代码遇到了这个问题。但是，我想调试/分析我的代码以查看是否存在某种饥饿/错误共享。我怎样才能做到这一点？请随时批评我的代码，因为我仍在学习很多关于内存/cpu 和多线程的知识。
#include <boost/thread.hpp>
class MultiThreadedFitnessProcessingStrategy
{
public:
MultiThreadedFitnessProcessingStrategy(unsigned int numWorkerThreads):
_startBarrier(numWorkerThreads + 1),
_endBarrier(numWorkerThreads + 1),
_started(false),
_shutdown(false),
_numWorkerThreads(numWorkerThreads)
{
assert(_numWorkerThreads > 0);
}
virtual ~MultiThreadedFitnessProcessingStrategy()
{
stopWorkers();
}
void startWorkers()
{
_shutdown = false;
_started = true;
for(unsigned int i = 0; i < _numWorkerThreads;i++)
{
boost::thread* workerThread = new boost::thread(
boost::bind(&MultiThreadedFitnessProcessingStrategy::workerTask, this,i)
);
_threadQueue.push_back(new std::queue<myWorkObject::ptr>());
_workerThreads.push_back(workerThread);
}
}
void stopWorkers()
{
_startBarrier.wait();
_shutdown = true;
_endBarrier.wait();
for(unsigned int i = 0; i < _numWorkerThreads;i++)
{
_workerThreads[i]->join();
}
}
void workerTask(unsigned int id)
{
//Wait until all worker threads have started.
while(true)
{
//Wait for any input to become available.
_startBarrier.wait();
bool queueEmpty = false;
std::queue<SomeClass::ptr >* myThreadq(_threadQueue[id]);
while(!queueEmpty)
{
SomeClass::ptr myWorkObject;
//Make sure queue is not empty,
//Caution: this is necessary if start barrier was triggered without queue input (e.g., shutdown) , which can happen.
//Do not try to be smart and refactor this without knowing what you are doing!
queueEmpty = myThreadq->empty();
if(!queueEmpty)
{
chromosome = myThreadq->front();
assert(myWorkObject);
myThreadq->pop();
}
if(myWorkObject)
{
myWorkObject->doWorkIntenseStuffHere();
}
}
//Wait until all worker threads have synchronized.
_endBarrier.wait();
if(_shutdown)
{
return;
}
}
}
void doWork(const myWorkObject::chromosome_container &refcontainer)
{
if(!_started)
{
startWorkers();
}
unsigned int j = 0;
for(myWorkObject::chromosome_container::const_iterator it = refcontainer.begin();
it != refcontainer.end();++it)
{
if(!(*it)->hasFitness())
{
assert(*it);
_threadQueue[j%_numWorkerThreads]->push(*it);
j++;
}
}
//Start Signal!
_startBarrier.wait();
//Wait for workers to be complete
_endBarrier.wait();
}
unsigned int getNumWorkerThreads() const
{
return _numWorkerThreads;
}
bool isStarted() const
{
return _started;
}
private:
boost::barrier _startBarrier;
boost::barrier _endBarrier;
bool _started;
bool _shutdown;
unsigned int _numWorkerThreads;
std::vector<boost::thread*> _workerThreads;
std::vector< std::queue<myWorkObject::ptr >* > _threadQueue;
};
【问题讨论】：
-
取决于你有多少可用的 CPU/内核......然后它们是真正的 CPU 还是在幕后做一些超现实的东西或黑魔法......不应该有比 CPU 更多的线程...
-
这很清楚。我有足够的可用硬件内核（intel core i7），我应该会看到运行 4 到 7 个并发线程的一些好处。
-
这里可能有多个问题，例如仅举几例：任务只执行很少的计算；同步问题（阻塞与计算）；并发数据访问问题。要获得更多信息，您必须使用性能分析工具，例如 gperftool、perf、oprofile 或可能是英特尔的 VTune（价格昂贵，但有试用版）。
-
如果您使用的是 Windows 平台，那么this MSDN article 显示了一种计算虚假共享的方法。
标签： c++ multithreading boost-thread cpu-cache false-sharing