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

Question

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

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

问题是，当主线程退出并销毁队列时，条件变量也会被销毁，但这会失败，因为条件变量正在使用中。这会抛出一个异常，一切都会糟糕地崩溃。

我想向工作线程发出信号，唤醒它们并让它们退出，等待它们完成，然后继续主线程。我的问题是，当这些线程完成时，我需要额外的同步原语吗？

不管怎样，下面是队列的代码：

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

Answer 2

您有两个选择。当队列超出作用域时，当其他线程引用它时，它实际上不会被销毁(例如，使用shared_ptr，将它传递给其他线程；在main()的末尾调用cancel()；一旦其他线程抛出cancel并可能退出，队列就会被销毁)。

或者，如果您希望确保它在main()的末尾被销毁，那么您需要等待其他线程。如果你可以在析构函数之外处理等待，你可以按照JaredC的建议去做。要在析构函数中做到这一点，似乎不存储所有线程，而只是保留一个计数和另一个同步原语。无论采用哪种方式，您都需要队列保持某种状态，以便等待所有线程完成。

在我看来，第一种解决方案(使用shared_ptr)更简洁。