C++線程安全的隊列關機

Question

我使用這個類的生產者 - 消費者安裝在C++：

#pragma once 

#include <queue> 
#include <mutex> 
#include <condition_variable> 
#include <memory> 
#include <atomic> 

template <typename T> class SafeQueue 
{ 
public: 
    SafeQueue() : 
    _shutdown(false) 
    { 

    } 

    void Enqueue(T item) 
    { 
     std::unique_lock<std::mutex> lock(_queue_mutex); 
     bool was_empty = _queue.empty(); 
     _queue.push(std::move(item)); 
     lock.unlock(); 

     if (was_empty) 
      _condition_variable.notify_one(); 
    } 

    bool Dequeue(T& item) 
    { 
     std::unique_lock<std::mutex> lock(_queue_mutex); 

     while (!_shutdown && _queue.empty()) 
      _condition_variable.wait(lock); 

     if(!_shutdown) 
     { 
      item = std::move(_queue.front()); 
      _queue.pop(); 

      return true; 
     } 

     return false; 
    } 

    bool IsEmpty() 
    { 
     std::lock_guard<std::mutex> lock(_queue_mutex); 
     return _queue.empty(); 
    } 

    void Shutdown() 
    { 
     _shutdown = true; 
     _condition_variable.notify_all(); 
    } 

private: 
    std::mutex _queue_mutex; 
    std::condition_variable _condition_variable; 
    std::queue<T> _queue; 
    std::atomic<bool> _shutdown; 
}; 

我用它是這樣的：

class Producer 
{ 
public: 
    Producer() : 
     _running(true), 
     _t(std::bind(&Producer::ProduceThread, this)) 
    { } 

    ~Producer() 
    { 
     _running = false; 
     _incoming_packets.Shutdown(); 
     _t.join(); 
    } 

    SafeQueue<Packet> _incoming_packets; 

private: 
    void ProduceThread() 
    { 
     while(_running) 
     { 
      Packet p = GetNewPacket(); 
      _incoming_packets.Enqueue(p); 
     } 
    } 

    std::atomic<bool> _running; 
    std::thread _t; 
} 

class Consumer 
{ 
    Consumer(Producer* producer) : 
     _producer(producer), 
     _t(std::bind(&Consumer::WorkerThread, this)) 
    { } 

    ~Consumer() 
    { 
     _t.join(); 
    } 

private: 
    void WorkerThread() 
    { 
     Packet p; 

     while(producer->_incoming_packets.Dequeue(p)) 
      ProcessPacket(p); 
    } 

    std::thread _t; 
    Producer* _producer; 
} 

這工作最的時間。但在同時，一旦當我刪除生產者（並引起它的析構函數調用SafeQueue::Shutdown，在_t.join()塊永遠

我的猜測是，這個問題是這裏（SafeQueue::Dequeue）：

while (!_shutdown && _queue.empty()) 
     _condition_variable.wait(lock); 

SafeQueue::Shutdown從線＃1被調用，而線＃2檢查完_Shutdown但在此之前它執行_condition_variable.wait(lock)，因此「缺失」的notify_all()。能這樣呢？

如果這是問題，解決問題的最佳方法是什麼？

Answer 1

由於SafeQueue對象爲生產者所有，因此當生產者完成時，刪除生產者會導致通知的消費者和被刪除的SafeQueue之間的競爭條件被刪除。

我建議共享資​​源既不是由生產者也不是由消費者擁有，而是作爲引用傳遞給每個構造器。

更改生產者和消費者構造函數;

Producer(SafeQueue<Packet> & queue) : 
    _running(false), _incoming_packets(queue) {} 


Consumer(SafeQueue<Packet> & queue) : 
    _running(false), _incoming_packets(queue) {} 

以這種方式使用您的實例;

SafeQueue<Packet> queue; 
Producer producer(queue); 
Consumer consumer(queue); 

...do stuff... 

queue.shutdown(); 

這也解決了您在Consumer類中與Producer類緊密耦合的糟糕設計問題。

另外，在析構函數中殺死和連接線程可能是個壞主意，就像你爲〜Producer做的那樣。最好爲每個線程類添加一個Shutdown（）方法，並明確地調用它們;

producer.shutdown(); 
consumer.shutdown(); 
queue.shutdown(); 

關機順序並不重要，除非你是擔心失去仍在隊列中，當你停止消費未處理的分組。

Answer 2

在您的SafeQueue::Dequeue中，您可能正在使用std::condition_variable錯誤的方式...更改此：

bool Dequeue(T& item) 
{ 
    std::unique_lock<std::mutex> lock(_queue_mutex); 

    while (!_shutdown && _queue.empty()) 
     _condition_variable.wait(lock); 
    if(!_shutdown) 
    { 
     item = std::move(_queue.front()); 
     _queue.pop(); 
     return true; 
    } 
    return false; 
} 

到

bool Dequeue(T& item) 
{ 
    std::unique_lock<std::mutex> lock(_queue_mutex); 
    _condition_variable.wait(lock, []{ return _shutdown || !_queue.empty() }); 
    if(!_shutdown) 
    { 
     item = std::move(_queue.front()); 
     _queue.pop(); 
     return true; 
    } 

    return false; 
} 

其次，Consumer數據成員的初始化順序是不正確的有關於它的構造

class Consumer 
{ 
    Consumer(Producer* producer) : 
     _producer(producer), 
     _t(std::bind(&Consumer::WorkerThread, this)) 
    { } 
    ...... 
    // _t will be constructed first, regardless of your constructor initializer list 
    // Meaning, the thread can even start running using an unintialized _producer 
    std::thread _t;  
    Producer* _producer; 
} 

應該重新排序到：

class Consumer 
{ 
    Consumer(Producer* producer) : 
     _producer(producer), 
     _t(std::bind(&Consumer::WorkerThread, this)) 
    { } 
    ...... 
    Producer* _producer; 
    std::thread _t;  
} 

您的問題的另一部分由CAB's answer