c++ - C++中的多线程队列销毁

Question

所以我有一个共享的并发队列。它似乎工作得很好，除了破坏。

队列的实现方式是它包含一个条件变量和互斥量对。启动了几个等待此条件变量的工作线程。当有新对象可供处理时，它们将被插入队列并通知条件变量。

问题是当主线程退出时，销毁队列，条件变量被销毁，但是由于条件变量正在使用中，所以失败了。这会抛出一个异常，一切都会爆炸。

我想向 worker 发出信号，唤醒他们并让他们退出，并等待他们完成，然后在主线程上继续。我的问题是在这些线程完成时解决问题——我是否需要额外的同步原语？

无论如何，这里是队列的代码:

// Based on code from http://www.justsoftwaresolutions.co.uk/threading/implementing-a-thread-safe-queue-using-condition-variables.html
// Original version by Anthony Williams
// Modifications by Michael Anderson

#include "boost/thread.hpp"
#include <deque>

template<typename Data>
class concurrent_queue
{
private:
    std::deque<Data> the_queue;
    mutable boost::mutex the_mutex;
    boost::condition_variable the_condition_variable;
    bool is_canceled;

public:
    concurrent_queue() : the_queue(), the_mutex(), the_condition_variable(), is_canceled(false) {}
    struct Canceled{};
    void push(Data const& data)
    {
        boost::mutex::scoped_lock lock(the_mutex);
        if (is_canceled) throw Canceled();
        the_queue.push_back(data);
        lock.unlock();
        the_condition_variable.notify_one();
    }

    bool empty() const
    {
        boost::mutex::scoped_lock lock(the_mutex);
        if (is_canceled) throw Canceled();
        return the_queue.empty();
    }

    bool try_pop(Data& popped_value)
    {
        boost::mutex::scoped_lock lock(the_mutex);
        if (is_canceled) throw Canceled();
        if(the_queue.empty())
        {
            return false;
        }

        popped_value=the_queue.front();
        the_queue.pop_front();
        return true;
    }

    void wait_and_pop(Data& popped_value)
    {
        boost::mutex::scoped_lock lock(the_mutex);

        while(the_queue.empty() && !is_canceled)
        {
            the_condition_variable.wait(lock);
        }
        if (is_canceled) throw Canceled();

        popped_value=the_queue.front();
        the_queue.pop_front();
    }

    std::deque<Data> wait_and_take_all()
    {
        boost::mutex::scoped_lock lock(the_mutex);

        while(the_queue.empty() && !is_canceled)
        {
            the_condition_variable.wait(lock);
        }
        if (is_canceled) throw Canceled();

        std::deque<Data> retval;
        std::swap(retval, the_queue);
        return retval;
    }

    void cancel()
    {
       boost::mutex::scoped_lock lock(the_mutex);
       if (is_canceled) throw Canceled();
       is_canceled = true;
       lock.unlock();
       the_condition_variable.notify_all();
    }

};

Answer 1

您可以在每个线程上调用 join() 以等待它完成执行。像这样:

void DoWork() {};

int main()
{
    boost::thread t(&DoWork);
    // signal for the thread to exit
    t.join();    // wait until it actually does exit

    // destroy the queue
}

或者您可以为多线程使用 boost::thread_group。

int main()
{
    boost::thread_group tg;

    for(int i = 0 ; i < 10 ; ++i)
        tg.create_thread(&DoWork);

    // signal to stop work

    tg.join_all();

    // destroy the queue
}