c++ - 单个生产者和多个单线程消费者

Question

我的应用程序从网络接收数据包并将它们分派(dispatch)给一个或多个“处理器”。 (每个数据包都属于一个预定义的“流”，可以通过查看数据包数据来识别。)

目前有一个线程可以完成所有工作:

从网络设备获取数据包

识别每个数据包的处理器

将数据包分派(dispatch)到其处理器

以每秒 2000 万个数据包(10Gbps 的 60 字节数据包)的速率接收传入数据。

然而，该解决方案只能跟上非常少量的流和处理器。例如，在 10 个流的情况下，已经有大约 10-20% 的数据包丢失。

由于步骤 (3) 是最昂贵的，我计划将该工作委托(delegate)给工作线程池。

但是，我必须小心，因为处理器本身不是线程安全的。所以只有一个工作线程可以同时将数据包分派(dispatch)到同一个处理器。

这似乎是基于任务的编程的一个很好的用例。但是我无法轻松地将 TBB 文档中解释的设计模式与我的问题相匹配。

所以我的问题是:如何组织我的消费者线程，以便它们将数据包均匀地分配给单线程处理器？

我不期待一个完整的解决方案，但我会对你的建议或随意的想法感到满意:)

Answer 1

我已经完成了一些嵌入式编程，我不得不处理相对较高的吞吐量 - 不像你在这里那么快!希望您使用的硬件比我习惯的要强大得多……有一些简单的策略应该适用于您的情况!
1. 输入/处理队列和相关的内存管理很关键。
如果您的数据速率很高，则传入数据的队列必须非常高效。您应该进行尽可能少的处理，否则可能会丢失设备中的数据。 (我习惯于从某种具有相对较小缓冲区的快速串行设备读取数据，因此对于在不丢失数据的情况下可以保留设备多长时间没有实时限制。这让我养成了习惯处理从设备读取作为一个完全独立的任务，它只处理读取数据而不是别的。)
一系列非常简单的固定大小的预分配缓冲区与它获得的效率差不多:有一个“空闲”缓冲区队列和一个“填充”缓冲区队列。如果您使用无锁链表，维护这些链表会非常快，而且入队/出队操作在许多操作系统中都很常见。
避免使用 malloc 或其他动态分配，因为当它们需要管理自己的“空闲”和“分配”块的数据结构时，它们具有显着的(通常是不可预测的)开销。如果生产者或工作线程大约在同一时间释放或分配内存，它们也可能会执行可能会不可预测地阻塞生产者或工作线程的锁。相反，尝试找到较低级别的例程，用于分配和释放操作系统为队列提供的整个页面(unixy 平台上的 mmap，VirtualAllocEx)。这些通常需要做的工作要少得多，因为它们使用 MMU 功能来映射 RAM 的物理页面，并且在内存中没有复杂的数据结构需要维护，每次调用都有更可靠的运行时，并且速度可能很快如果它的数量不足，足以扩展您的免费列表。
在生产者中，不要担心小于整个块的单位。从队列中取出一个空闲块，将一个装满数据的块打包，加入到待处理的队列中。如果您必须确保在固定时间段内处理每个数据包，或者您需要处理“突发”数据速率，那么仍然尝试从您的输入设备读取一个完整的缓冲区，但要么将块的大小减小到是一个“合理”的时间量，或者使用超时并将部分填充的块排入队列以进行处理，并用某种空包“填充”剩余部分。我发现这样做通常比包含大量代码来处理部分填充的缓冲区要快。
如果可以，请非常仔细地设置生产者线程的处理器关联性和线程优先级。理想情况下，您希望生产者线程的优先级高于任何消费者线程，并绑定(bind)到特定的内核。没有什么可以防止传入数据在缓冲区空间不足的情况下被读取。
2.加工
你说过有:

几个流

几个“处理器”，不是线程安全的

这里有用的是在数据包上并行运行处理器，但是从您的问题中不清楚这在多大程度上是可能的。
处理器跨流线程安全吗？ (只要它们在两个不同的流上运行，我们就可以在两个不同的线程中运行处理器吗？)
处理器在同一流中的不同处理器之间是线程安全的吗？ (我们可以在单独的线程中在同一个流上运行多个处理器吗？)
处理器是否需要按特定顺序运行？
在不知道这一点的情况下，仍然有一些通用的东西是有用的建议。
有第二个线程处理从生产者读取完整缓冲区并将它们分派(dispatch)到适当的处理器(在其他线程中)，然后将完整缓冲区放回“空”队列进行处理。虽然您会损失一些直线效率(一个线程进行读取和调度会比两个线程稍微“快”一些)，但如果有暂时锁定，这种方式至少不会阻止从输入设备读取。
创建或查找允许您将作业分配给线程池的库，特别是如果与可以并行运行的线程数相比，您有许多处理器。实现某种作业队列也相对简单，允许作业之间存在一些简单的关系(例如“此作业要求作业 X 和 Y 已先完成”，“此作业不能与使用的任何其他作业并行运行相同的处理器”)。即使是作业管理器只在第一个可用线程上运行第一个可运行作业的简单策略也可能非常有效。
尽量避免复制。如果处理器可以“就地”处理数据包而无需从缓冲区复制它们，那么您就节省了大量无意义的周期。即使您必须进行复制，让多个线程从“只读”共享缓冲区复制数据也比让单个线程将消息复制和分派(dispatch)消息到多个线程要好。
如果检查是否应该为给定的数据包运行处理器的速度非常快，那么您最好有多个作业，每个作业都检查它是否应该进行一些处理。与其让单个线程确定哪个处理器应该在哪个数据包上运行，不如让多个线程更快，每个处理器或一组处理器一个线程，检查每个数据包一次是否应该运行它的处理器。这归结为这样一种想法，即在多个线程中多次对只读资源进行简单检查可能比在线程之间进行同步花费的时间更少。
如果您可以并行运行处理器(如果它们正在处理来自不同流的数据)，那么通过数据传递以获取流列表，然后为每个流启 Action 业是一个好主意。您还可以收集属于每个流的数据包列表，但同样，这是在作业检查每个数据包的速度与在单线程中收集该列表并将每个数据包传递给所需的时间之间的权衡他们各自的工作。
希望其中一些策略对您的情况有用!让我们知道它是如何工作的......这是你必须处理的大量数据，最好知道在比我习惯的更快的数据速率上什么是有效的和无效的!祝你好运!