异步网络竞赛

Question

我正在为一个网络应用程序编写一个客户端，我想将接收和处理不同线程的消息分开。

这是我目前的解决办法：

Mutex mutex;
Queue queue;

recv()
{
    while(true)
    {
        messages = receive_some_messages();

        mutex.lock();
        queue.add(messages);
        mutex.unlock();

        process.notify();
    }
}

proc()
{
    while(true)
    {
        block_until_notify();
        Queue to_process;

        mutex.lock();
        to_process.add( queue.take_all() );
        mutex.unlock();

        foreach(message in to_process)
        {
            process_message(message);
        }

    }
}

然而，这有一个竞赛条件：

1. recv接收大量消息并将它们放入队列。
2. 通知程序。
3. proc从队列中获取所有消息并开始工作。
4. recv接收更多的消息并将它们放入队列。
5. recv会通知proc，但是由于proc仍然在工作，所以什么也不做。
6. proc完成它的迭代。
7. proc块-队列中仍有未处理的消息

我可以想出几种方法来解决这个问题，但是没有一个是有利的。

解决方案1

我可以调整同步，以便在处理过程中保持互斥锁：

proc()
{
    while(true)
    {
        block_until_notify();
        Queue to_process;

        mutex.lock();
        to_process.add( queue.take_all() );

        foreach(message in to_process)
        {
            process_message(message);
        }
        mutex.unlock(); 
    }
}

但这意味着线程只运行: recv或proc都是活动的，但两者都不是。

解决方案2

我可以移除那块并通知。

recv()
{
    while(true)
    {
        messages = receive_messages();

        mutex.lock();
        queue.add(messages);
        mutex.unlock();
    }
}

proc()
{
    while(true)
    {
        Queue to_process;

        mutex.lock();
        to_process.add( queue.take_all() );
        mutex.unlock();

        foreach(message in to_process)
        {
            process_message(message);
        }

    }
}

但这意味着proc将在繁忙的等待循环中运行，只有在recv向队列中添加消息时才会阻塞。

问题是

我想要一个解决方案，在没有繁忙等待的情况下，proc和recv不单独运行。有人知道我能做什么吗？

Answer 1

我认为，如果使用者在队列耗尽后检查空队列，您就可以过日子了。

proc()
{
    while(true)
    {
        Queue to_process;

        mutex.lock();

        if (queue.empty()) {
           mutex.unlock();
           block_until_notify();
           mutex.lock();
        }

        to_process.add( queue.take_all() );

        mutex.unlock();

        foreach(message in to_process)
        {
            process_message(message);
        }

    }
}

我相信这能解决你提到的种族问题。

Answer 2

您的block_until_notify()函数可能是一个条件变量。这样做的方法是更改该函数，以便锁定互斥锁，检查队列是否为空，然后等待通知。如果队列不是空的，则继续处理它。在处理队列之后，返回到block_until_notify函数并重复该进程，再次检查阻塞前队列是否为空。

如果您没有使用某种条件变量，那么我建议使用信号量。在Windows上，每当消息被添加到队列中时，您都会调用ReleaseSemaphore。接收方将调用信号量句柄上的WaitForSingleObject。这将在循环中完成，即使队列为空，循环也会继续重复，直到等待阻塞为止。