多生產者單消費者延遲任務執行

Question

我想模擬一個系統，其中有多個線程正在生成數據，並且有一個消耗數據的線程。訣竅是我不希望專用線程使用數據，因爲所有線程都在一個池中。相反，我希望其中一個生產者在工作時清空隊列，並在另一個生產者已經清除隊列時收益。

其基本思想是有一個工作隊列和一個處理周圍的鎖。每個生產者將其有效負載推入隊列，然後嘗試來輸入鎖。該嘗試是非阻塞的，並返回true（獲取鎖）或false（該鎖由其他人持有）。

如果獲得了鎖定，那麼該線程將處理隊列中的所有數據，直到其爲空（包括處理期間由其他生​​產者引入的任何新的有效載荷）。一旦所有工作都被處理完畢，線程就會釋放鎖定並退出。

以下是該算法的C++代碼：

void Process(ITask *task) { 
    // queue is a thread safe implementation of a regular queue 
    queue.push(task); 

    // crit_sec is some handle to a critical section like object 
    // try_scoped_lock uses RAII to attempt to acquire the lock in the constructor 
    //     if the lock was acquired, it will release the lock in the 
    //     destructor 
    try_scoped_lock lock(crit_sec); 

    // See if this thread won the lottery. Prize is doing all of the dishes 
    if (!lock.Acquired()) 
     return; 

    // This thread got the lock, so it needs to do the work 
    ITask *currTask; 
    while (queue.try_pop(currTask)) { 
      ... execute task ... 
    } 
} 

一般這種代碼工作正常，我從來沒有親眼目睹我對下面描述的行爲，但執行讓我感到不安。理由是在線程退出while循環和釋放臨界區之間引入競爭條件。

整個算法依賴於這樣的假設：如果正在保持鎖，則線程正在爲隊列服務。

我基本上是在尋找啓蒙的2個問題：

1. 我是正確，有作爲描述的競態條件（獎金爲其他種族）
2. 是否有實現這種機制的標準模式是是高性能的，不會引入競爭條件？

Answer 1

是的，有一個競爭條件。

線程A添加一個任務，獲取lock，處理自身，然後請求queue中的任務。它被拒絕。

此時的線程B將任務添加到queue。然後它嘗試獲取鎖定，並失敗，因爲線程A具有鎖定。線程B退出。

線程A然後退出，queue非空，沒有人處理它上面的任務。

這將很難找到，因爲該窗口相對較窄。爲了使它更容易找到，在while循環引入「睡10秒」後。在調用代碼中，插入一個任務，等待5秒鐘，然後插入第二個任務。再過10秒鐘後，檢查兩個插入任務是否完成，並且queue上仍有任務需要處理。

解決此問題的一種方法是將try_pop更改爲try_pop_or_unlock，並將您的lock傳遞給它。 try_pop_or_unlock然後自動檢查空的queue，如果是的話解鎖lock並返回false。

另一種方法是改善線程池。爲它添加一個基於「消耗」任務啓動器的計數信號量。

semaphore_bool bTaskActive; 
counting_semaphore counter; 

when (counter || !bTaskActive) 
    if (bTaskActive) 
    return 
    bTaskActive = true 
    --counter 
    launch_task(process_one_off_queue, when_done([&]{ bTaskActive=false)); 

當計數信號是積極的，還是購買成品戳消耗的任務，它會啓動消費的任務，如果沒有消費的任務積極。

但這只是我的頭頂。