多線程C++消息傳遞

Question

我負責修改同步C程序，以便它可以並行運行。目標是讓它儘可能便攜，因爲它是許多人使用的開源程序。正因爲如此，我認爲最好將程序包裝在C++層中，以便我可以利用便攜式boost庫。我已經這樣做了，一切似乎都按預期工作。

我遇到的問題是決定什麼是在線程之間傳遞消息的最佳方法。幸運的是，該計劃的架構是多生產者和單一消費者的架構。更好的是，消息的順序並不重要。我已經讀過單生產者/單消費者（SPSC）隊列將受益於這種架構。那些有多線程編程經驗的人有什麼建議？我對這個東西很陌生。另外任何使用boost來實現SPSC的代碼示例都將不勝感激。

Answer 1

以下是我用於合作性多任務/多線程庫（MACE）http://bytemaster.github.com/mace/的技術。除了隊列爲空時外，它具有無鎖的好處。

struct task { 
    boost::function<void()> func; 
    task* next; 
}; 


boost::mutex      task_ready_mutex; 
boost::condition_variable  task_ready; 
boost::atomic<task*>    task_in_queue; 

// this can be called from any thread 
void thread::post_task(task* t) { 
    // atomically post the task to the queue. 
    task* stale_head = task_in_queue.load(boost::memory_order_relaxed); 
    do { t->next = stale_head; 
    } while(!task_in_queue.compare_exchange_weak(stale_head, t, boost::memory_order_release)); 

    // Because only one thread can post the 'first task', only that thread will attempt 
    // to aquire the lock and therefore there should be no contention on this lock except 
    // when *this thread is about to block on a wait condition. 
    if(!stale_head) { 
     boost::unique_lock<boost::mutex> lock(task_ready_mutex); 
     task_ready.notify_one(); 
    } 
} 

// this is the consumer thread. 
void process_tasks() { 
    while(!done) { 
    // this will atomically pop everything that has been posted so far. 
    pending = task_in_queue.exchange(0,boost::memory_order_consume); 
    // pending is a linked list in 'reverse post order', so process them 
    // from tail to head if you want to maintain order. 

    if(!pending) { // lock scope 
     boost::unique_lock<boost::mutex> lock(task_ready_mutex);     
     // check one last time while holding the lock before blocking. 
     if(!task_in_queue) task_ready.wait(lock); 
    } 
} 

Answer 2

如果只有一個消費者但是有多個生產者，那麼我會使用一個數組或者一些具有O（1）訪問時間的數組結構的數據結構，其中每個數組槽都代表一個生產者消費者隊列。單生產者消費者隊列的巨大優勢在於，您可以在沒有任何顯式同步機制的情況下使其無鎖，從而使其成爲多線程環境中非常快速的數據結構。請參閱my answer here瞭解單生產者消費者隊列的基本實現。

Answer 3

在網上有很多生產者 - 消費者隊列的例子，對多個生產者/消費者是安全的。 @bytemaster發佈了一個在每個消息中使用鏈接來消除隊列類本身的存儲 - 這是一個很好的方法，我自己在嵌入式作業中使用它。

在隊列類必須提供存儲的地方，我通常會使用大小爲N的「池隊列」，並在啓動時加載N *個消息類實例。需要通信的線程必須從池中彈出*消息，將其加載並將其排隊。當最終「用完」時，*消息被推回到池中。這限制了消息的數量，因此所有隊列只需要長度爲N - 不需要調整大小，不需要新建（），不需要刪除（），便於泄漏檢測。