1

我正在尝试通过 packaged_task 创建生产者-消费者模式代码如下： test_thread9_producer1并将test_thread9_producer2任务推送到队列中并test_thread9_consumer1从队列中检索任务以执行

但是，在运行时test_thread9，它会正确执行任务但以调试错误结束：已调用 abort。我不确定为什么？任何人都可以帮助我更多地了解packaged_task吗？

第二个问题：消费者正在循环运行，当两个生产者完成将所有任务推入队列并 完成执行队列中的所有任务时while(1)，我想不出优雅的方式让test_thread9_consumer1退出。test_thread9_consumer1谁能给我一些建议？

void test_thread9()
{
    std::thread t1(test_thread9_producer1);
    std::thread t2(test_thread9_producer2);
    std::thread t3(test_thread9_consumer1);

    t1.join();
    t2.join();
    t3.join();
} 

std::deque<std::packaged_task<int()>>task_q;
std::mutex lock9;

int factial_calc2(int in)
{
    int ret = 1;
    for (int i = in; i > 1; i--)
    {
        ret = ret*i;
    }
    std::lock_guard<std::mutex> locker(lock9);
    std::cout << "input is " << in << "result is " << ret << std::endl;
    return ret;
}

void test_thread9_producer1()
{
    for (int i = 0; i < 10; i = i + 2)
    {
        std::lock_guard<std::mutex> locker(lock9);
        std::packaged_task<int()> t1(std::bind(factial_calc2, i));
        task_q.push_back(std::move(t1));
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

void test_thread9_producer2()
{
    for (int i = 1; i < 10; i = i + 2)
    {
        std::lock_guard<std::mutex> locker(lock9);
        std::packaged_task<int()> t1(std::bind(factial_calc2, i));
        task_q.push_back(std::move(t1));
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}


void test_thread9_consumer1()
{
    std::packaged_task<int()>t;
    while (1)
    {
        {
            std::lock_guard<std::mutex> locker(lock9);
            if (!task_q.empty())
            {
                t = std::move(task_q.front());
                task_q.pop_front();
            }
        }
        t();
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}
4

2 回答 2

0

为什么会崩溃？

如果你的消费者线程发现一个空队列，它仍然会尝试执行打包的任务，尽管它没有被移动。这是 UB，因此是运行时错误！

您可以通过检查 packaged_task 是否有效来改进这一点： 

while (1)
{
    std::packaged_task<int()>t;  // to make sure that valid() checks this iteration
    {
       ...
    }
    if (t.valid())
        t();  // execute only if it's a valid task
    ...
}

如何避免无限循环？

您必须以某种方式跟踪正在运行的内容。一种简单的技术是使用atomic变量来管理共享状态信息（可以在没有锁定的情况下同时访问）。

例如，您可以计算成品生产者的数量

std::atomic<int>finished{0};  // count the producers that are finished
...

void test_thread9_producerN() { cout <<"开始生产者"<

然后，您可以调整您的消费者以考虑以下信息： 

void test_thread9_consumer1()
{
    bool nothing_to_do{false}; 
    while (!nothing_to_do && finished<2)
    {
    ...   
        nothing_to_do=task_q.empty();  // in the lock protected section 
        if (!nothing_to_do)     
    ...
    }
}

在线演示

于 2016-07-17T18:07:55.380 回答
0 
void test_thread9_consumer1()
{
    std::packaged_task<int()>t;
    while (1)
    {
        {
            std::lock_guard<std::mutex> locker(lock9);
            if (!task_q.empty())
            {
                t = std::move(task_q.front());
                task_q.pop_front();
            }
        }
        t();
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

让我们修剪一下这段代码：

    std::packaged_task<int()>t;
    while (1)
    {
            if (!task_q.empty())
                t = std::move(task_q.front());
        t();
    }

现在我们可以清楚地看到错误：无论我们是否获取它，您都尝试调用 t()。

void test_thread9_consumer1()
{
    std::packaged_task<int()>t;
    while (1)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));

        {
            std::lock_guard<std::mutex> locker(lock9);
            if (task_q.empty())
                continue;
            t = std::move(task_q.front());
            task_q.pop_front();
        }

        t();
    }
}

您可能希望通过查看“条件变量”来继续探索线程

对于问题的第二部分，您可以考虑将主线程更改为join所有提供程序，然后设置一个全局标志，指示工作已完成。

std::atomic g_produce;

void test_thread9()
{
    std::thread t1(test_thread9_producer1);
    std::thread t2(test_thread9_producer2);
    g_producing = true;

    std::thread t3(test_thread9_consumer1);

    t1.join();
    t2.join();

    g_producing = false;

    t3.join();
} 

std::deque<std::packaged_task<int()>>task_q;
std::mutex lock9;

int factial_calc2(int in)
{
    int ret = 1;
    for (int i = in; i > 1; i--)
    {
        ret = ret*i;
    }
    std::lock_guard<std::mutex> locker(lock9);
    std::cout << "input is " << in << "result is " << ret << std::endl;
    return ret;
}

void test_thread9_producer1()
{
    for (int i = 0; i < 10; i = i + 2)
    {
        std::lock_guard<std::mutex> locker(lock9);
        std::packaged_task<int()> t1(std::bind(factial_calc2, i));
        task_q.push_back(std::move(t1));
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

void test_thread9_producer2()
{
    for (int i = 1; i < 10; i = i + 2)
    {
        std::lock_guard<std::mutex> locker(lock9);
        std::packaged_task<int()> t1(std::bind(factial_calc2, i));
        task_q.push_back(std::move(t1));
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

void test_thread9_consumer1()
{
    std::packaged_task<int()>t;
    for (;;)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        {
            std::lock_guard<std::mutex> locker(lock9);
            if (task_q.empty()) {
                if (!g_producing)
                    break;
                continue;
            }
            t = std::move(task_q.front());
            task_q.pop_front();
        }
        t();
    }
}
于 2016-07-17T18:55:07.817 回答