多线程管道中的维护顺序

假设如下

• 听众
• 输出消息应包含等待 句柄和指向工作/输出数据的指针
• 可能有任意数量的 工作者线程

我将考虑以下流程:

输入队列侦听器执行以下步骤:

1. 提取输入消息;
2. 创建输出消息:
   1. 初始化工作程序数据结构
   2. 重置等待句柄
3. 将指向输出消息的指针排入 工作 队列
4. 将指向输出消息的指针排入 输出 队列

工作线程执行以下操作:

1. 等待一个 工作队列 到 提取指向输出的指针 来自它的信息
2. 根据给定的数据处理消息,并在完成时设置事件

消费者执行以下操作:

1. 等待n 输出队列 到 提取指向输出的指针 来自它的信息
2. 等待句柄,直到输出数据就绪
3. 对数据做些什么

让一个线程读取输入队列,在输出队列上发布一个占位符,然后将该项移交给工作线程进行处理。当数据准备就绪时,工作线程将更新占位符。当需要输出队列中的值的线程读取占位符时,它可以阻塞,直到关联的数据就绪。

因为只有一个线程读取输入队列,而这个线程立即将占位符放在输出队列中,所以输出队列中的顺序与输入中的顺序相同。工作线程可以是多个的,并且可以以任何顺序进行转换。

在支持未来的平台上,它们是理想的占位符。在其他系统上,您可以使用事件、监视器或条件变量。

这将是特定于实现的。一个通用的解决方案是对输入项进行编号并保留编号,以便以后对输出项进行排序。这可以在输出队列被填充后完成,也可以作为填充队列的一部分完成。换句话说,您可以将它们插入到适当的位置,并且只允许在下一个可用项是连续的时读取队列。

编辑

我将勾画出一个基本的方案,通过使用适当的原语来保持它的简单:

1. 而不是排队 Packet 在输入队列中,我们围绕它创建一个未来的值,并将其排队放入输入和输出队列。在C中,您可以这样写:

   var future = new Lazy<Packet>(delegate() { return Process(packet); }, LazyThreadSafetyMode.ExecutionAndPublication);
   

2. 工人池中的线程出列 future 从输入队列执行 future.Value 使委托运行JIT,并在委托处理完数据包后返回。

3. 一个或多个消费者退出 未来 从输出队列。每当他们需要包的值时,他们会调用 未来价值 ,如果工作线程已调用委托,则立即返回。

简单,但有效。

如果使用窗口方法(已知元素数),请为输出队列使用数组。例如,如果它是媒体流,而您丢弃的包处理速度不够快。

否则,使用 priority queue (特殊类型的堆,通常基于固定大小的数组实现)用于输出项。

您需要添加一个序列号或任何可以对每个数据包中的项目进行排序的数据。优先级队列是一个树形结构,确保插入/弹出时项目的顺序。