multithreading - 如何处理消息的并行处理？

Question

我正在为我的应用程序重新设计消息传递系统以使用英特尔线程构建 block ，并且在尝试在两种可能的方法之间做出决定时感到困惑。

基本上，我有一系列消息对象，并且对于每种消息类型，都有一系列处理程序。对于每个消息对象，我应用为该消息对象类型注册的每个处理程序。

顺序版本将是这样的(伪代码):

for each message in message_sequence                     <- SEQUENTIAL
    for each handler in (handler_table for message.type)
        apply handler to message                         <- SEQUENTIAL

第一种方法我正在考虑依次(按顺序)处理消息对象并同时应用处理程序。

优点:

消息的可预测顺序(即，我们保证 FIFO 处理顺序)

(可能)处理每条消息的延迟更低

缺点:

比单一消息类型的处理程序更多的可用处理资源(不好的并行化)

处理器缓存使用不当，因为需要为每个处理程序复制消息对象才能使用

小型处理程序的大量开销

这种方法的伪代码如下:

for each message in message_sequence                              <- SEQUENTIAL
    parallel_for each handler in (handler_table for message.type)
        apply handler to message                                  <- PARALLEL

第二种方法是并行处理消息，并按顺序将处理程序应用于每条消息。

优点:

更好地使用处理器缓存(将消息对象保持在所有将要使用它的处理程序的本地)

小型处理程序不会产生太多开销(只要还有其他处理程序也可以运行)

预期的消息多于处理程序的数量，因此并行的可能性更大

缺点:

不可预测的顺序 - 如果消息 A 在消息 B 之前发送，则它们可能同时被处理，或者 B 可能在 A 的所有处理程序完成之前完成处理(顺序是不确定的)

伪代码如下:

parallel_for each message in message_sequence                     <- PARALLEL
    for each handler in (handler_table for message.type)
        apply handler to message                                  <- SEQUENTIAL

第二种方法比第一种方法有更多的优点，但非确定性排序是一个很大的缺点。

你会选择哪种方法，为什么？是否还有其他方法我应该考虑(除了明显的第三种方法:并行消息和并行处理程序，据我所知，这两种方法都有缺点并且没有真正的补救因素)？

谢谢!

编辑:

我想我会默认使用#2，但允许在每条消息上附加一个“对话标签”。任何具有相同标签的消息都会根据其对话顺序进行排序和处理。处理程序会在消息旁边传递对话标签，因此如果需要，他们可以继续对话。像这样的东西:

Conversation c = new_conversation()
send_message(a, c)
...
send_message(b, c)
...
send_message(x)

handler foo (msg, conv)
    send_message(z, c)

...
register_handler(foo, a.type)

a在b之前处理，b在z之前处理。 x 可以与 a、b 和 z 并行处理。一旦处理了对话中的所有消息，对话就会被销毁。

Answer 1

我会说做一些不同的事情。不要发送工作到线程。有线程拉当他们完成以前的工作时工作。

保持固定数量的工作线程(最佳数量等于系统中 CPU 内核的数量)，并让每个工作线程在完成前一个任务后从全局队列中按顺序拉取下一个要执行的任务。显然，您需要跟踪消息之间的依赖关系，以推迟对消息的处理，直到其依赖关系得到完全处理。

这可以通过非常小的同步开销来完成 - 可能只使用原子操作，没有像互斥锁或信号量这样的重型原语。

此外，如果您将消息传递给每个处理程序 引用 ，而不是复制，让不同 CPU 内核上的不同处理程序同时处理相同的消息实际上可以 改善缓存性能，因为更高级别的缓存(通常从 L2 向上)通常为 共享 CPU 内核之间 - 因此，当一个处理程序将消息读入缓存时，第二个内核上的另一个处理程序将在 L2 中拥有该消息。所以仔细想想——你真的需要复制这些信息吗？