5

我的程序通过使用空闲工作线程将多行文本打印到控制台。然而，问题是工作人员在打印文本之前没有等待先前的工作人员完成，这导致文本被插入到另一个工作线程的文本中，如下图所示：

在此处输入图像描述

我需要通过使用 std::condition_variable 来解决这个问题 - 称为忙等待问题。我尝试根据在此链接上找到的示例在下面的代码中实现 condition_variable ，并且由于我对 C++ 的一般知识有限，以下 stackoverflow 问题对我有所帮助，但还不够。所以最后我只是把所有的东西都评论出来了，我现在很茫然。

// 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;
}
4

2 回答 2

4

这是一个工作示例：

// 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。但至少我们摆脱了那种该死的忙碌等待！

在一个真实的示例中，您希望并行处理数据或执行任务，并且只同步输出，如果您从头开始，这种结构仍然不是解决此问题的正确方法。

我改变了什么以及为什么

我使用原子布尔作为条件，因为它们在多个线程之间共享时具有确定性行为。并非在所有情况下都绝对必要，但仍然是一种良好的做法。

您应该在循环中包含退出条件while(true)（例如，由处理程序设置的标志SIGINT或其他东西），否则您的程序将永远不会退出。这只是一个任务，所以我们跳过了它，但在生产代码中不要忽视这一点非常重要。

也许分配可以用一个条件变量来解决，但我不确定，无论如何最好使用两个，因为每个条件变量的作用更加清晰易读。基本上，我们等待一个任务，然后让入队者等到它完成，然后告诉它它实际上已被处理，我们为下一个任务做好了准备。最初你走在正确的轨道上，但我认为有两个简历更明显出了什么问题。

此外，在使用之前设置条件 vars (ready和processed)也很重要。notify()

我删除locker.unlock()了，因为该案例是不必要的。c++ std 锁是RAII结构，因此当它超出范围时，锁将被解锁，这基本上是下一行。通常最好避免无意义的分支，因为您使程序不必要地有状态。

教学狂言...

既然手头的问题已经得到解决和解决，我认为总体上需要说一些关于作业的事情，我认为这对你的学习可能比解决上述问题更重要。

如果您对作业感到困惑或沮丧，那么很好，您应该是。您很难将方形钉子放入圆孔中是有道理的，我认为这个问题的真正价值在于学会分辨何时使用正确的工具来完成正确的工作，何时不使用.

条件变量是解决繁忙循环问题的正确工具，但是这个赋值（正如@nm 所指出的）是一个简单的竞争条件。也就是说，这只是一个简单的竞态条件，因为它包含一个不必要且执行不力的线程池，使问题变得复杂且难以理解，完全没有目的。就是说，std::async无论如何，在现代 c++ 中，它应该比手动线程池更受欢迎（它更容易正确实现，并且在许多平台上性能更高，并且不需要一堆全局变量和单例以及专门分配的资源）。

如果这是您的老板而不是您的教授的任务，您将提交以下内容：

for(int i = 0; i < 8; ++i)
{
    std::cout << "Text printed by worker " << i << std::endl;
}

这个问题通过一个简单的for循环来解决（最佳）。繁忙的等待/锁定问题是设计糟糕的结果，“正确”的做法是修复设计，而不是包扎它。我什至不认为这个分配是有指导意义的，因为没有可能的方法或理由来并行化输出，所以它最终只会让每个人都感到困惑，包括 SO 社区。线程只是引入了不必要的复杂性而没有改进计算，这似乎是消极的训练。

从赋值的结构中，很难真正判断教授本人是否非常了解线程和条件变量的概念。出于培训目的，必须将作业归结，简化，并且有些琐碎，但这实际上与这里所做的相反，其中一个复杂的问题是由一个简单的问题组成的。

作为一项规则，我从不回答关于 SO 的作业相关问题，因为我认为给出答案会阻碍学习，而开发人员最有价值的技能是学习如何在一个想法出现之前用头撞墙。然而，像这样的人为作业只会带来负面的训练，虽然在学校你必须遵守教授的规则，但重要的是要学会在看到人为的问题时识别它们，解构它们，并得出结论。简单而正确的解决方案。

于 2017-02-02T13:20:06.873 回答
1

我认为这是正常的，因为互斥体在打印之前没有被锁定。对于循环中的每一轮，不能保证 i 将在 i+1 之前打印。

为了获得良好的打印优先级，您应该在函数入队的互斥锁之后显示消息。

于 2017-02-02T11:14:22.613 回答