高吞吐量非阻塞服务器设计：忙等待的替代方案

Question

我一直在构建一个用于多媒体消息的高吞吐服务器应用程序，实现语言为C++。每个服务器可以单独使用，也可以将多个服务器组合在一起创建基于DHT的覆盖网络；这些服务器就像Skype中的超级对等点。

目前正在进行中。目前，该服务器可以处理大约每秒200,000条消息（256字节消息），并且在我的机器上具有大约256 MB/s的最大吞吐量（Intel i3 Mobile 2 GHz，Fedora Core 18（64位），4 GB RAM），用于长度为4096字节的消息。服务器有两个线程，一个线程处理所有IO（基于epoll，边缘触发），另一个线程处理传入的消息。还有另一个线程用于覆盖管理，但在当前讨论中不重要。

讨论中的两个线程使用两个循环缓冲区共享数据。第1个线程使用一个循环缓冲区将新鲜的消息排队给第2个线程，而第2个线程通过另一个循环缓冲区返回已处理的消息。服务器完全无锁。我没有使用任何同步原语，甚至没有使用原子操作。

循环缓冲区永远不会溢出，因为消息是预分配的（在启动时预先分配）。实际上，为了减少内存碎片和增加缓存效率，所有重要/经常使用的数据结构都是预分配的，因此我们知道服务器启动时我们将创建的最大消息数量，因此我们可以预先计算缓冲区的最大大小，然后相应地初始化循环缓冲区。

现在我的问题是：线程#1一次将序列化的消息（实际上是指向消息对象的指针）排队，而线程#2按块（32/64/128个块）从队列中取出消息，并通过第二个循环缓冲区返回已处理的消息。如果没有新消息，线程#2会保持繁忙等待，从而始终保持一个CPU核心繁忙。如何进一步改进设计？繁忙等待策略的替代方案是什么？我想以优雅和高效的方式完成这项工作。我考虑过使用信号量，但我担心这不是最好的解决方案，因为每次在线程#1中排队消息时我必须调用“sem_post”，这可能会显着降低吞吐量，第二个线程必须调用“sem_post”相同次数以防止信号量溢出。此外，我担心信号量实现可能在内部使用互斥。

第二个好选择可能是使用信号，如果我能发现一种算法，只有在第二个线程“清空队列并正在调用sigwait”或“已经在sigwait上等待”的情况下才会引发信号，简而言之，必须最少次数引发信号，尽管如果比所需次数多引发几次信号也不会有害。是的，我确实使用了Google搜索，但我在互联网上找到的解决方案都不令人满意。以下是一些考虑因素：

A.服务器在进行系统调用时必须尽量浪费最少的CPU周期，并且必须最少使用系统调用次数。

B. 必须有非常低的开销，并且算法必须高效。

C. 无需任何锁定。

我希望所有选项都被考虑在内。

这是我分享关于我的服务器信息的网站链接，以更好地了解功能和用途：www.wanhive.com

- user2856296

---

"A"（和“C”）听起来像“XY要求”。为什么您关心在系统调用中浪费了多少个周期？或者您愿意为避免一个系统调用周期支付多少用户周期？ - CB Bailey

使用选择器怎么样？ - Michael D

目前我正在探索所有可能性。我们必须避免在线程#1的整个IO周期中出现任何可能成为瓶颈的调用。在线程#2上进行忙等待在多核架构上可能不会对服务器吞吐量产生任何明显的负面影响，但我们浪费了CPU周期而没有做任何有用的工作。 - user2856296

顺便提一下，你可以看一下LMAX Disruptor架构。它与你正在做的非常相似。 - Evgeny Lazin

Answer 1

如果您需要尽快唤醒第二个线程，那么繁忙等待是一个不错的选择。实际上，这是通知另一个处理器所做更改的最快方法。您需要在两端生成内存栅栏(write fence和read fence)。但是，只有在两个线程都在专用处理器上执行时才成立。在这种情况下，无需上下文切换，只有缓存一致性流量。

有一些改进可以进行。

1. 如果第二个线程总体上是CPU绑定的并且正在进行繁忙等待，则调度程序可能会对其进行惩罚(至少在Windows和Linux上是如此)。操作系统调度程序动态地调整线程优先级以提高整体系统性能。它降低消耗大量CPU时间的CPU绑定线程的优先级，以防止线程饥饿。您需要手动增加第二个线程的优先级来预防这种情况。
2. 如果您有多核或多处理器计算机，则处理器数量不足，应用程序无法利用硬件并发性。您可以通过使用多个处理器线程(线程#2)来缓解这种情况。

处理步骤的并行化。 有两种选项。

1. 您的消息是完全有序的，并且需要按照到达的顺序进行处理。
2. 消息可以被重新排序。可以以任何顺序进行处理。

在第一种情况下，您需要N个循环缓冲区、N个处理线程和N个输出缓冲区以及一个消费者线程。线程#1以轮询方式向这些循环缓冲区中排队的消息。

// Thread #1 pseudocode
auto message = recv()
auto buffer_index = atomic_increment(&message_counter);
buffer_index = buffer_index % N;  // N is the number of threads
// buffers is an array of cyclic buffers - Buffer* buffers[N];
Buffer* current_buffer = buffers[buffer_index];
current_buffer->euqueue(message);

每个线程从其中一个缓冲区中消耗消息，并将结果排队到其专用输出缓冲区。

// Thread #i pseudocode
auto message = my_buffer->dequeue();
auto result = process(message);
my_output_buffer->enqueue(result);

现在你需要按照到达顺序处理所有这些消息。你可以通过专用的消费者线程来完成，通过循环方式从输出循环缓冲区中出队已处理的消息。

// Consumer thread pseudocode
// out_message_counter is equal to message_counter at start
auto out_buffer_index = atomic_increment(&out_message_counter);
out_buffer_index = out_buffer_index % N;
// out_buffers is array of output buffers that is used by processing
// threads
auto out_buffer = out_buffers[out_buffer_index];
auto result = out_buffer->dequeue();
send(result);  // or whatever you need to do with result

在第二种情况下，当您不需要保留消息顺序时 - 您不需要使用消费者线程和输出循环缓冲区。您只需在处理线程中执行所需的操作即可。

在第一种情况下，N必须等于num CPU's - 3（"- 3"是一个I/O线程+一个消费者线程+一个DHT线程），而在第二种情况下，N必须等于num CPU's - 2（"- 2"是一个I/O线程+一个DHT线程）。这是因为如果处理器超额订阅，则繁忙等待无法有效。

- Evgeny Lazin

---

你的回答加强了我的设计决策。通过采用忙等待策略，我减少了代码的复杂性。但是，如果应用程序在后台运行于桌面机上，100% 的 CPU 使用率可能会被用户视为一种烦恼。 - user2856296

你需要看一下LMAX disruptor。它们有可调的等待策略。它们可以使用阻塞等待和忙碌等待两种方式。 - Evgeny Lazin

如果延迟很关键，那么100％的CPU使用率是正常的事情。如果您只需要吞吐量-您可以使用某种阻塞等待，可能与批处理相结合。 - Evgeny Lazin

Answer 2

听起来你想协调一个生产者和一个消费者，他们通过一些共享状态相连。至少在Java中，为了避免忙等待，可以使用wait和notify。使用这种方法，如果线程#2发现队列为空，它可以通过调用wait进入阻塞状态，避免旋转CPU。一旦线程#1将一些东西放入队列中，它就可以进行通知。在C++中进行此类机制的快速搜索得到以下结果：wait and notify in C/C++ shared memory

- Dhruv

---

虽然我已经为你的答案点赞，但它在所提供的上下文中并没有提供令人满意的解决方案。请仔细阅读问题中的上下文。我们如何优化在不同线程中使用“wait”和“notify”调用的方法？ - user2856296

谢谢。与其在第1个线程将东西放入队列时立即通知，不如在队列半满或类似情况下通知，这样当第2个线程醒来时就有更多的工作可用。这可能会减少通知广播。这完全取决于生产者和消费者的相对速度。我甚至不确定您是否需要担心这种低级优化。只有在完成应用程序编写后才能发现这是否会成为大问题。过早地进行优化是万恶之源，记住了吗？ - Dhruv

Answer 3

当队列为空时，您可以让线程#2休眠X毫秒。

X可以通过所需队列的长度+一些保护带来确定。

顺便说一句，在用户模式（ring3）中，您不能使用MONITOR/MWAIT指令，这对您的问题非常理想。

编辑

您绝对应该尝试TBB's RWlock（有免费版本）。听起来就是您要找的。

编辑2

另一个选择是使用条件变量。它们涉及互斥锁和条件。基本上，您等待条件变为“真”。低级pthread内容可以在此处找到。

- egur

---

不需要使用读写锁，因为实现对于单个生产者和单个消费者是完全线程安全的。这是一个精心设计的循环缓冲区的固有属性。在我看来，休眠X毫秒听起来像是反模式，缺乏优雅性。 - user2856296

请参见编辑2 - 条件变量。 - egur