实现一个 condition_variable 来解决多线程忙等待
Implementing a condition_variable to solve a multithreaded busy-wait
我的程序通过使用空闲工作线程将多行文本打印到控制台。然而，问题在于 worker 没有等待之前的 worker 完成打印文本，这导致文本被插入到另一个 worker 线程的文本中，如下图所示：
我需要通过使用 std::condition_variable 解决这个问题 - 称为忙等待问题。我尝试在下面的代码中实现 condition_variable，基于 the example found at this link, and the following Whosebug 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;
}
我认为这是正常的，因为互斥体在打印前没有被锁定。
对于循环中的每一轮，不能保证 i 会在 i+1 之前打印。
为了获得良好的打印优先级，您应该在函数 enqueue 的互斥锁之后显示消息。
这是一个工作示例：
// 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
为什么不在生产代码中这样做
因为赋值是按顺序打印字符串，这段代码实际上不能真正并行化，所以我们设计了一种方法让它完全按顺序使用std::condition_variable 所需的 Golden hammer。但至少我们摆脱了该死的忙碌等待！
在一个真实的例子中，你想要并行处理数据或执行任务，并同步只是输出，但这种结构仍然不是正确的方法如果你是从头开始做的话。
我改变了什么以及为什么
我对条件使用了原子布尔值，因为它们在多个线程之间共享时具有确定性行为。并非在所有情况下都绝对必要，但一个好的做法 none 越少。
你 应该 在 while(true) 循环中包含一个退出条件（例如，由 SIGINT 处理程序或其他东西设置的标志）或你的程序将 永远不会 退出。这只是一项任务，所以我们跳过了它，但在生产代码中不要忽视这一点非常重要。
也许赋值可以用一个条件变量来解决，但我不确定，无论如何最好使用两个，因为 每个条件变量更清晰易读确实。基本上，我们等待一个任务，然后让入队者等到它完成，然后告诉它它实际上已经被处理了，我们已经为下一个任务做好了准备。您最初的方向非常正确，但我认为有了两个简历，问题就更明显了。
此外，使用 notify().
before 设置条件变量（ready 和 processed）很重要
我删除了locker.unlock()，因为这个案例是不必要的。 c++ std 锁是 RAII 结构，所以当它超出范围时锁将被解锁，这基本上是下一行。通常最好避免无意义的分支，因为您会使程序不必要地有状态。
教学咆哮...
既然手头的问题已经得到解决，我认为关于作业的一般情况需要说明一些事情，我认为这些事情对你的学习可能比解决问题更重要问题如前所述。
如果您对作业感到困惑或沮丧，那很好，您应该。很难将方形钉子放入圆孔中，这是有道理的，我认为这个问题的真正价值在于学会分辨何时使用正确的工具来完成正确的工作，何时不使用.
条件变量是解决busy-loop问题的正确工具，但是这个赋值（正如@n.m指出的那样）是一个简单的竞争条件。也就是说，这只是一个简单的竞争条件，因为它包含一个不必要且实施不力的 thread-pool，使问题变得复杂并且毫无意义地难以理解。也就是说，无论如何，std::async 应该比现代 c++ 中的 hand-rolled 线程池更受欢迎（它更容易正确实现，并且在许多平台上性能更高，并且不需要一堆全局变量和单例和独占分配的资源）。
如果这是你老板而不是你教授的作业，这就是你要上交的内容：
for(int i = 0; i < 8; ++i)
{
std::cout << "Text printed by worker " << i << std::endl;
}
这个问题通过一个简单的 for 循环（最佳）解决了。繁忙的 wait/locking 问题是糟糕设计的结果，"right" 要做的是修复设计，而不是包扎它。我什至不认为分配有指导意义，因为没有可能的方法或理由来并行化输出，所以它最终只会让每个人感到困惑，包括 SO 社区。线程只是引入了不必要的复杂性而没有改进计算，这似乎是消极的训练。
从作业的结构上很难判断教授本人是否很好地理解了线程和条件变量的概念。出于训练目的，作业必须 boiled-down、简化并略微简化，但这实际上与此处所做的相反，其中一个复杂的问题是由一个简单的问题组成的。
通常我从不回答关于 SO 的家庭作业相关问题，因为我认为放弃答案会阻碍学习，而开发人员最有价值的技能是学习如何用头撞墙，直到一个想法突然出现。然而，像这样的人为作业只会带来负面训练，虽然在学校你必须遵守教授的规则，但重要的是要学会在你看到人为问题时识别它们，解构它们，并得出结论简单而正确的解决方案。
我的程序通过使用空闲工作线程将多行文本打印到控制台。然而，问题在于 worker 没有等待之前的 worker 完成打印文本，这导致文本被插入到另一个 worker 线程的文本中，如下图所示：
我需要通过使用 std::condition_variable 解决这个问题 - 称为忙等待问题。我尝试在下面的代码中实现 condition_variable，基于 the example found at this link, and the following Whosebug 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;
}
我认为这是正常的，因为互斥体在打印前没有被锁定。 对于循环中的每一轮，不能保证 i 会在 i+1 之前打印。
为了获得良好的打印优先级，您应该在函数 enqueue 的互斥锁之后显示消息。
这是一个工作示例：
// 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
为什么不在生产代码中这样做
因为赋值是按顺序打印字符串，这段代码实际上不能真正并行化，所以我们设计了一种方法让它完全按顺序使用std::condition_variable 所需的 Golden hammer。但至少我们摆脱了该死的忙碌等待！
在一个真实的例子中，你想要并行处理数据或执行任务，并同步只是输出，但这种结构仍然不是正确的方法如果你是从头开始做的话。
我改变了什么以及为什么
我对条件使用了原子布尔值，因为它们在多个线程之间共享时具有确定性行为。并非在所有情况下都绝对必要，但一个好的做法 none 越少。
你 应该 在 while(true) 循环中包含一个退出条件（例如，由 SIGINT 处理程序或其他东西设置的标志）或你的程序将 永远不会 退出。这只是一项任务，所以我们跳过了它，但在生产代码中不要忽视这一点非常重要。
也许赋值可以用一个条件变量来解决，但我不确定，无论如何最好使用两个，因为 每个条件变量更清晰易读确实。基本上，我们等待一个任务，然后让入队者等到它完成，然后告诉它它实际上已经被处理了，我们已经为下一个任务做好了准备。您最初的方向非常正确，但我认为有了两个简历，问题就更明显了。
此外，使用 notify().
ready 和 processed）很重要
我删除了locker.unlock()，因为这个案例是不必要的。 c++ std 锁是 RAII 结构，所以当它超出范围时锁将被解锁，这基本上是下一行。通常最好避免无意义的分支，因为您会使程序不必要地有状态。
教学咆哮...
既然手头的问题已经得到解决，我认为关于作业的一般情况需要说明一些事情，我认为这些事情对你的学习可能比解决问题更重要问题如前所述。
如果您对作业感到困惑或沮丧，那很好，您应该。很难将方形钉子放入圆孔中，这是有道理的，我认为这个问题的真正价值在于学会分辨何时使用正确的工具来完成正确的工作，何时不使用.
条件变量是解决busy-loop问题的正确工具，但是这个赋值（正如@n.m指出的那样）是一个简单的竞争条件。也就是说，这只是一个简单的竞争条件，因为它包含一个不必要且实施不力的 thread-pool，使问题变得复杂并且毫无意义地难以理解。也就是说，无论如何，std::async 应该比现代 c++ 中的 hand-rolled 线程池更受欢迎（它更容易正确实现，并且在许多平台上性能更高，并且不需要一堆全局变量和单例和独占分配的资源）。
如果这是你老板而不是你教授的作业，这就是你要上交的内容：
for(int i = 0; i < 8; ++i)
{
std::cout << "Text printed by worker " << i << std::endl;
}
这个问题通过一个简单的 for 循环（最佳）解决了。繁忙的 wait/locking 问题是糟糕设计的结果，"right" 要做的是修复设计，而不是包扎它。我什至不认为分配有指导意义，因为没有可能的方法或理由来并行化输出，所以它最终只会让每个人感到困惑，包括 SO 社区。线程只是引入了不必要的复杂性而没有改进计算，这似乎是消极的训练。
从作业的结构上很难判断教授本人是否很好地理解了线程和条件变量的概念。出于训练目的，作业必须 boiled-down、简化并略微简化，但这实际上与此处所做的相反，其中一个复杂的问题是由一个简单的问题组成的。
通常我从不回答关于 SO 的家庭作业相关问题，因为我认为放弃答案会阻碍学习，而开发人员最有价值的技能是学习如何用头撞墙，直到一个想法突然出现。然而，像这样的人为作业只会带来负面训练，虽然在学校你必须遵守教授的规则，但重要的是要学会在你看到人为问题时识别它们，解构它们，并得出结论简单而正确的解决方案。