我已经做了一些嵌入式编程,我不得不处理相对较高的吞吐量-没有这里的速度快!希望你使用的硬件比我习惯的更强大。。。有一些简单的策略应该适用于您的情况!

1.输入/处理队列和相关内存管理至关重要。

如果您具有高数据速率,则传入数据的队列必须非常高效。您应该尽可能少地进行处理,否则可能会丢失设备中的数据。(我习惯于用相对较小的缓冲区从某种快速串行设备读取数据,因此,在设备不被读取而不丢失数据的时间上存在实时限制。这让我习惯于将从设备读取数据作为一项完全独立的任务来处理,只处理读取数据而不处理其他任务。)

非常简单的一系列固定大小的预分配缓冲区的效率与它所能达到的一样高:有一个“空闲”缓冲区队列,和一个“已填充”缓冲区的队列。如果使用无锁链接列表,维护这些列表可能非常快,在许多操作系统中,入队/出队操作非常常见。

避免使用malloc或其他动态分配,因为当他们需要管理自己的“空闲”和“已分配”块的数据结构时,它们会产生大量(通常是不可预测的)开销。如果它们同时释放或分配内存,它们也可能执行锁,从而不可预测地阻塞生产者线程或工作线程。相反,尝试找到较低级别的例程来分配和释放操作系统为队列提供的整个页面(unixy平台上的mmap,VirtualAllocEx)。由于它们使用MMU功能来映射RAM的物理页面,而且内存中没有复杂的数据结构需要维护,每次调用都有更可靠的运行时间,因此通常需要做的工作要少得多 能够 如果空闲列表不足,请快速扩展。

在生产商中,不要担心单位小于整块。从队列中取出一个空闲块,打包一个数据块,将其添加到要处理的队列中。如果您必须确保每个数据包在固定的时间段内被处理,或者您需要处理“突发”数据速率,那么仍然可以尝试从输入设备读取完整的缓冲区,但要么将数据块的大小减小到“合理”的时间量,要么使用超时并将部分填充的数据块排队处理,然后用某种空数据包“填充”剩余部分。我发现这样做通常比包含大量代码来处理部分填充的缓冲区更快。

如果可以,请非常仔细地设置生产者线程的处理器相关性和线程优先级。理想情况下,您希望生产者线程具有比任何消费者线程更高的优先级,并与特定的核心绑定。没有什么可以阻止传入数据在缓冲区空间不足时被读取。

2.加工

你说过有:

1. 多个流
2. 几个非线程安全的“处理器”

这里需要做的是并行运行数据包上的处理器,但这在多大程度上是可能的,这从你的问题中还不清楚。

处理器跨流线程安全吗?(我们是否可以在两个不同的线程中运行一个处理器,只要它们在两种不同的流上运行?)

同一个流中的不同处理器之间的处理器线程安全吗?(我们可以在不同的线程中在同一个流上运行多个处理器吗?)

处理器是否需要按特定顺序运行?

在不知道这一点的情况下,仍然有一些通用的东西是有用的建议。

有一个第二个线程,它负责从生产者读取完整的缓冲区,并将它们分派给适当的处理器(在其他线程中),然后将完整的缓冲放回“空”队列中进行处理。虽然你失去了一些直线效率(一个线程执行读取和调度会比两个线程稍微快一点),但如果存在瞬时锁定,至少这种方式不会阻止从输入设备读取数据。

使用创建或查找库允许您将作业分配给线程池,特别是如果与可以并行运行的线程数相比,您拥有多个处理器。还可以相对简单地实现某种允许作业之间建立简单关系的作业队列(例如,“此作业要求先完成作业X和Y”、“此作业不能与使用相同处理器的任何其他作业并行运行”)。即使是作业管理器只在第一个可用线程上运行第一个可运行作业的简单策略也是非常有效的。

尽量避免复制。如果处理器可以“就地”处理数据包,而无需从缓冲区复制数据包,那么就节省了大量无意义的周期。即使您确实需要复制,让多个线程从“只读”共享缓冲区复制数据也比让单个线程复制消息并将消息分派给多个线程要好。

如果检查一个处理器是否应该为一个给定的数据包运行速度非常快,那么最好有几个作业,每个作业检查它是否应该执行一些处理。与其让一个线程决定哪些处理器应该在哪些数据包上运行,不如让多个线程(每个处理器或一组处理器一个)检查每个数据包一次是否应该运行它的处理器。这归结为这样一个想法,即在多个线程中多次对只读资源进行简单检查可能比在线程之间进行同步所花费的时间要少。

如果您可以并行运行处理器,如果它们正在处理来自不同流的数据,那么对数据进行传递以获得流列表,然后为每个流启动作业是一个好主意。您还可以收集属于每个流的数据包列表,但同样,这是一个权衡,即作业检查每个数据包的速度与在单个线程中收集该列表并将每个数据包传递给各自的作业所需的时间。

希望这些策略中的一些对您的情况有用!让我们知道它是如何运作的。。。这是一个你必须处理的大量数据的地狱,知道什么是有效的,什么是无效的,比我习惯的更快的数据速率会更好!祝你好运

这是我对可能的解决方案的想法。

假设我们有n个处理器。让我们介绍n个互斥体,每个处理器一个。我们还要介绍一个数据包队列。所有传入的数据包都放入此队列。

工作线程的操作方式如下:

1. 从传入数据包队列中获取数据包。
2. 确定必要的处理器。
3. 尝试获取相应的互斥锁。如果锁获取成功,则处理数据包。否则,请重新排队并转到1。
4. 处理完成后,转至步骤1。

可能的缺点:

1. 数据包被重新排队,这意味着它们可能会被延迟/处理,这可能会破坏您的交易(不确定)。
2. 队列上的争用可能很高。您可能希望研究如何使用无锁队列。
3. 队列显然会消耗额外的内存,我不知道您是否有多余的内存。

编辑:更多关于内存消耗的思考-当然,可以对队列可以消耗的内存量设置上限-那么,问题是当内存用完时该怎么办。我想说,最好的做法是开始丢弃数据包(我的印象是,在你的情况下,丢弃一些数据包并不是什么大事),直到队列耗尽一点。

与此相关的是——我认为这个用例的好队列实现应该不惜一切代价避免动态内存分配——预先分配内存,并确保关键代码路径上没有分配。

为什么不能使用多个队列,每个处理器一个? 这些队列可以是无锁的(没有互斥锁)。

1. 从网络设备获取数据包
2. 识别每个数据包的处理器(PID)
3. 将数据包推送到队列
4. 工作人员:处理队列中的数据包[k]

对于类似的问题,我使用了无锁环形缓冲区的轮询,自动覆盖最旧的数据包。