C＃等待生产者/消费者中的多个事件

Question

我正在使用producer/consumer pattern实现数据链接层。数据链路层有自己的线程和状态机，用于通过线路（以太网，RS-232 ...）传输数据链路协议。物理层的接口表示为System.IO.Stream。另一个线程将消息写入数据链接对象并从中读取消息。

数据链路对象具有空闲状态必须等待的四个条件之一：

1. 一个字节被接收
2. 消息是可从网络螺纹
3. 保活定时器已过期
4. 所有通信都被

我有困难的时候figu网络层取消在没有将通信拆分为读/写线程的情况下（这会显着增加复杂性），最好的方法是实现这一点。以下是我能得到3开出4：

// Read a byte from 'stream'. Timeout after 10 sec. Monitor the cancellation token. 
stream.ReadTimeout = 10000; 
await stream.ReadAsync(buf, 0, 1, cts.Token); 

或

BlockingCollection<byte[]> SendQueue = new ...; 
... 
// Check for a message from network layer. Timeout after 10 seconds. 
// Monitor cancellation token. 
SendQueue.TryTake(out msg, 10000, cts.Token); 

我应该怎么做才能阻止线程，等待所有四个条件？所有建议都欢迎。我没有设置任何架构或数据结构。

编辑：********感谢大家的帮助。这是我的解决方案********

首先我不认为有一个生产者/消费者队列的异步实现。所以我实施了类似this stackoverflow post的东西。

我需要一个外部和内部的取消源来停止消费者线程并分别取消中间任务，similar to this article。

byte[] buf = new byte[1]; 
using (CancellationTokenSource internalTokenSource = new CancellationTokenSource()) 
{ 
    CancellationToken internalToken = internalTokenSource.Token; 
    CancellationToken stopToken = stopTokenSource.Token; 
    using (CancellationTokenSource linkedCts = 
     CancellationTokenSource.CreateLinkedTokenSource(stopToken, internalToken)) 
    { 
     CancellationToken ct = linkedCts.Token; 
     Task<int> readTask = m_stream.ReadAsync(buf, 0, 1, ct); 
     Task<byte[]> msgTask = m_sendQueue.DequeueAsync(ct); 
     Task keepAliveTask = Task.Delay(m_keepAliveTime, ct); 

     // Wait for at least one task to complete 
     await Task.WhenAny(readTask, msgTask, keepAliveTask); 

     // Next cancel the other tasks 
     internalTokenSource.Cancel(); 
     try { 
      await Task.WhenAll(readTask, msgTask, keepAliveTask); 
     } catch (OperationCanceledException e) { 
      if (e.CancellationToken == stopToken) 
       throw; 
     } 

     if (msgTask.IsCompleted) 
      // Send the network layer message 
     else if (readTask.IsCompleted) 
      // Process the byte from the physical layer 
     else 
      Contract.Assert(keepAliveTask.IsCompleted); 
      // Send a keep alive message 
    } 
} 

Answer 1

在这种情况下，我只会用取消标记注销。像保活计时器一样的重复超时可以更好地表示为计时器。

所以，我会把它建模为三个可取消任务。首先，将取消标记：

的所有通信是由网络层取消

CancellationToken token = ...; 

然后，三个并发操作：

一个字节被接收

var readByteTask = stream.ReadAsync(buf, 0, 1, token); 

保活定时器超时

var keepAliveTimerTask = Task.Delay(TimeSpan.FromSeconds(10), token); 

消息是从网线

这一个是有点棘手。您当前的代码使用BlockingCollection<T>，它不是异步兼容的。我建议切换到TPL Dataflow's BufferBlock<T>或my own AsyncProducerConsumerQueue<T>，它们中的任何一个都可以用作异步兼容的生产者/消费者队列（意味着生产者可以是同步或异步的，消费者可以是同步的或异步的）。

BufferBlock<byte[]> SendQueue = new ...; 
... 
var messageTask = SendQueue.ReceiveAsync(token); 

然后你可以使用Task.WhenAny，以确定这些任务的完成：

var completedTask = await Task.WhenAny(readByteTask, keepAliveTimerTask, messageTask); 

现在，你可以通过比较completedTask别人和await检索结果荷兰国际集团他们：

if (completedTask == readByteTask) 
{ 
    // Throw an exception if there was a read error or cancellation. 
    await readByteTask; 
    var byte = buf[0]; 
    ... 
    // Continue reading 
    readByteTask = stream.ReadAsync(buf, 0, 1, token); 
} 
else if (completedTask == keepAliveTimerTask) 
{ 
    // Throw an exception if there was a cancellation. 
    await keepAliveTimerTask; 
    ... 
    // Restart keepalive timer. 
    keepAliveTimerTask = Task.Delay(TimeSpan.FromSeconds(10), token); 
} 
else if (completedTask == messageTask) 
{ 
    // Throw an exception if there was a cancellation (or the SendQueue was marked as completed) 
    byte[] message = await messageTask; 
    ... 
    // Continue reading 
    messageTask = SendQueue.ReceiveAsync(token); 
} 

Answer 2

取消读取不会让您知道数据是否被读取。取消和阅读不是相对于原子的。只有在取消后关闭流，该方法才有效。

队列方法更好。您可以创建链接的CancellationTokenSource，该链接在您需要时即被取消。而不是通过cts.Token你传递一个你控制的令牌。

然后，您可以根据时间，另一个标记以及您喜欢的任何其他事件发出该标记的信号。如果使用内置超时，则队列将在内部执行相同的操作，以将传入的令牌与超时链接起来。

Answer 3

我会与您的选择两个，等待任何4种情况发生。假设你已经有4个任务已经awaitable方法：

var task1 = WaitForByteReceivedAsync(); 
var task2 = WaitForMessageAvailableAsync(); 
var task3 = WaitForKeepAliveTimerAsync(); 
var task4 = WaitForCommunicationCancelledAsync(); 

// now gather them 
IEnumerable<Task<bool>> theTasks = new List<IEnumerable<Task<bool>>>{ 
task1, task2, task3, task4 
}; 

// Wait for any of the things to complete 
var result = await Task.WhenAny(theTasks); 

上面的代码将立即恢复第一个任务完成后，而忽视了其他3。

注：

在the documentation for WhenAny，它说：

返回的任务将始终与它的结果的RanToCompletion状态设置为第一任务完成结束。即使完成的第一项任务以“已取消”或“已故”状态结束，也是如此。

相信发生了什么事之前，所以一定要做到这一点最后的检查：

if(result.Result.Result == true) 
... // First Result is the task, the second is the bool that the task returns