multithreading - 线程构建 block concurrent_bounded_queue — 我如何“关闭”它？

Question

我正在使用 concurrent_bounded_queue Intel TBB 4.1 Update 3 用于生产者和消费者线程之间的通信:

concurrent_queue

concurrent_bounded_queue

队列类有一个名为 abort 的方法。抛出 tbb::user_abort所有阻塞 pop的线程和 push队列实例。两个线程之间的通信可能如下所示:

ConsThread | ProdThread
-----------+-------------
q.pop      |  get new data
(wait)     |  q.push
process    |  get new data
q.pop      |  no more data!
(wait)     |  q.abort
quit       |  quit

不幸的是，即使在这个简单的示例中，我也无法使用它来可靠地关闭队列，因为如果某些消费者之前没有完成处理 pop调用 abort 之前的 ped 数据，他们将完成迭代并返回到 pop 上的阻塞。 :

ConsThread | ProdThread
-----------+-------------
q.pop      |  get new data
(wait)     |  q.push
process    |  get new data
process    |  no more data!
process    |  q.abort
process    |  quit
process    |
q.pop      |
(wait)     |
(wait)     |
(wait)     |
(so lonely)|

现在我正在使用一个中度恶心的 hack，它产生另一个非分离线程(它加入消费者池线程)并等待它完成，同时发送更多 abort s 不时为迟到者:

bool areConsumerThreadsJoinedThankYou = false;
std::thread joiner(Joiner(consumerPool, &areConsumerThreadsJoinedThankYou));

while (!areConsumerThreadsJoinedThankYou) {
    rawQueue.abort();
    MAGIC_MSLEEP(100);
}

class Joiner的实现差不多

void Joiner::operator()(void)
{
    for (auto it = this->m_threadPool.begin();
         it < this->m_threadPool.end();
         it++)
        (*it)->join();
    this->m_done = true;
    *(this->m_flag) = true;
}

这当然是非常丑陋的。有没有更根本的解决方案？

Answer 1

创建一个指定的“EndOfData”项。如果您知道您有 K 个消费者，则让生产者在完成推送数据项后推送 K“EndOfData”项。让每个消费者在弹出“EndOfData”项目后退出。

如果事先不知道 K，让生产者推送一个“EndOfData”项。然后让每个弹出“EndOfData”项目的消费者在离开之前推送另一个“EndOfData”项目。在所有消费者完成后，将剩余一个“EndOfData”项，该项将在队列被销毁时被销毁。