我的程序通过使用空闲的工作线程将多行文本打印到控制台。然而，问题是工作线程在打印文本之前没有等待前一个工作线程完成，这导致文本被插入到另一个工作线程的文本中，如下图所示:
我需要通过使用 std::condition_variable 来解决这个问题——称为忙等待问题。我已经尝试在下面的代码中实现条件变量，基于 the example found at this link , 和 the following stackoverflow question对我有帮助，但还不够，因为我对 C++ 的一般知识有限。所以最后我只是把所有的东西都注释掉了，我现在不知所措。
// threadpool.cpp
// Compile with:
// g++ -std=c++11 -pthread threadpool.cpp -o threadpool
#include <thread>
#include <mutex>
#include <iostream>
#include <vector>
#include <deque>
class ThreadPool; // forward declare
//std::condition_variable cv;
//bool ready = false;
//bool processed = false;
class Worker {
public:
Worker(ThreadPool &s) : pool(s) { }
void operator()();
private:
ThreadPool &pool;
};
class ThreadPool {
public:
ThreadPool(size_t threads);
template<class F> void enqueue(F f);
~ThreadPool();
private:
friend class Worker;
std::vector<std::thread> workers;
std::deque<std::function<void()>> tasks;
std::mutex queue_mutex;
bool stop;
};
void Worker::operator()()
{
std::function<void()> task;
while (true)
{
std::unique_lock<std::mutex> locker(pool.queue_mutex);
//cv.wait(locker, [] {return ready; });
if (pool.stop) return;
if (!pool.tasks.empty())
{
task = pool.tasks.front();
pool.tasks.pop_front();
locker.unlock();
//cv.notify_one();
//processed = true;
task();
}
else {
locker.unlock();
//cv.notify_one();
}
}
}
ThreadPool::ThreadPool(size_t threads) : stop(false)
{
for (size_t i = 0; i < threads; ++i)
workers.push_back(std::thread(Worker(*this)));
}
ThreadPool::~ThreadPool()
{
stop = true; // stop all threads
for (auto &thread : workers)
thread.join();
}
template<class F>
void ThreadPool::enqueue(F f)
{
std::unique_lock<std::mutex> lock(queue_mutex);
//cv.wait(lock, [] { return processed; });
tasks.push_back(std::function<void()>(f));
//ready = true;
}
int main()
{
ThreadPool pool(4);
for (int i = 0; i < 8; ++i) pool.enqueue([i]() { std::cout << "Text printed by worker " << i << std::endl; });
std::cin.ignore();
return 0;
}
最佳答案
这是一个工作示例:
// threadpool.cpp
// Compile with:
// g++ -std=c++11 -pthread threadpool.cpp -o threadpool
#include <thread>
#include <mutex>
#include <iostream>
#include <vector>
#include <deque>
#include <atomic>
class ThreadPool;
// forward declare
std::condition_variable ready_cv;
std::condition_variable processed_cv;
std::atomic<bool> ready(false);
std::atomic<bool> processed(false);
class Worker {
public:
Worker(ThreadPool &s) : pool(s) { }
void operator()();
private:
ThreadPool &pool;
};
class ThreadPool {
public:
ThreadPool(size_t threads);
template<class F> void enqueue(F f);
~ThreadPool();
private:
friend class Worker;
std::vector<std::thread> workers;
std::deque<std::function<void()>> tasks;
std::mutex queue_mutex;
bool stop;
};
void Worker::operator()()
{
std::function<void()> task;
// in real life you need a variable here like while(!quitProgram) or your
// program will never return. Similarly, in real life always use `wait_for`
// instead of `wait` so that periodically you check to see if you should
// exit the program
while (true)
{
std::unique_lock<std::mutex> locker(pool.queue_mutex);
ready_cv.wait(locker, [] {return ready.load(); });
if (pool.stop) return;
if (!pool.tasks.empty())
{
task = pool.tasks.front();
pool.tasks.pop_front();
locker.unlock();
task();
processed = true;
processed_cv.notify_one();
}
}
}
ThreadPool::ThreadPool(size_t threads) : stop(false)
{
for (size_t i = 0; i < threads; ++i)
workers.push_back(std::thread(Worker(*this)));
}
ThreadPool::~ThreadPool()
{
stop = true; // stop all threads
for (auto &thread : workers)
thread.join();
}
template<class F>
void ThreadPool::enqueue(F f)
{
std::unique_lock<std::mutex> lock(queue_mutex);
tasks.push_back(std::function<void()>(f));
processed = false;
ready = true;
ready_cv.notify_one();
processed_cv.wait(lock, [] { return processed.load(); });
}
int main()
{
ThreadPool pool(4);
for (int i = 0; i < 8; ++i) pool.enqueue([i]() { std::cout << "Text printed by worker " << i << std::endl; });
std::cin.ignore();
return 0;
}
输出:
Text printed by worker 0
Text printed by worker 1
Text printed by worker 2
Text printed by worker 3
Text printed by worker 4
Text printed by worker 5
Text printed by worker 6
Text printed by worker 7
为什么不在生产代码中这样做
由于分配是按顺序打印字符串，这段代码实际上不能真正并行化，因此我们设计了一种方法，使用所需的 Golden hammer 使其完全按顺序工作。的
std::condition_variable .但至少我们摆脱了那该死的忙碌等待!在一个真实的例子中，您希望并行处理数据或执行任务，并且只同步输出，如果您从头开始，这种结构仍然不是正确的方法。
我改变了什么以及为什么
我使用原子 bool 值作为条件，因为它们在多个线程之间共享时具有确定性的行为。并非在所有情况下都绝对必要，但仍然是一种良好的做法。
您应该在
while(true) 中包含退出条件循环(例如由 SIGINT 处理程序或其他东西设置的标志)，否则您的程序将永远不会退出。这只是一个赋值，所以我们跳过了它，但是在生产代码中不要忽视这一点非常重要。也许赋值可以用一个条件变量解决，但我不确定，无论如何最好使用两个，因为每个变量的作用更清晰易读。基本上，我们等待一个任务，然后要求入队等待它完成，然后告诉它它实际上已经处理，我们准备好下一个。一开始你走在正确的轨道上，但我认为有两个简历更明显哪里出了问题。
此外，在使用
ready 之前设置条件变量( processed 和 notify() )很重要。 .我删除了
locker.unlock()因为这种情况是不必要的。 c++ std 锁是 RAII结构，所以当它超出范围时，锁将被解锁，这基本上是下一行。通常最好避免无意义的分支，因为您使程序不必要地有状态。教学吐槽...
既然手头的问题已经解决并解决了，我认为有一些事情需要说一下关于作业的一般情况，我认为这对您的学习可能比解决所述问题更重要。
如果您对作业感到困惑或沮丧，那很好，您应该如此。您很难将方钉放入圆孔中是有道理的，我认为这个问题的真正值(value)在于学会分辨何时使用正确的工具来完成正确的工作，何时不使用.
条件变量是解决繁忙循环问题的正确工具，但是这种分配(如@n.m. 所指出的)是一个简单的竞争条件。也就是说，这只是一个简单的竞争条件，因为它包含了一个不必要且实现不佳的线程池，使问题变得复杂且毫无意义地难以理解。也就是说，
std::async无论如何，在现代 C++ 中应该优先于手动滚动线程池(它更容易正确实现，并且在许多平台上性能更高，并且不需要一堆全局变量和单例以及专门分配的资源)。如果这是你的老板而不是你的教授的作业，这就是你要上交的:
for(int i = 0; i < 8; ++i)
{
std::cout << "Text printed by worker " << i << std::endl;
}
这个问题通过一个简单的
for 解决(最佳)环形。繁忙的等待/锁定问题是设计糟糕的结果，“正确”的做法是修复设计，而不是包扎它。我什至不认为这个作业有指导意义，因为没有可能的方法或理由来并行化输出，所以它最终会让每个人都感到困惑，包括 SO 社区。线程只会引入不必要的复杂性而没有改进计算，这似乎是负面训练。实际上很难从作业的结构中判断教授本人是否非常了解线程和条件变量的概念。出于培训目的，必须将作业归结、简化和稍微简化，但这实际上与此处所做的相反，在那里，复杂的问题由一个简单的问题构成。
通常，我从不回答有关 SO 的作业相关问题，因为我认为给出答案会阻碍学习，而且开发人员最宝贵的技能是学习如何用头撞墙，直到出现一个想法。然而，像这样的人为任务只会带来负面的训练，虽然在学校你必须遵守教授的规则，但重要的是学会在你看到人为的问题时识别它们，解构它们，然后来到这里。简单而正确的解决方案。
关于c++ - 实现一个条件变量来解决多线程忙等待，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/42000062/