TPL数据流使用异步操作

Question

我正在尝试使用TPL数据流，通过移植一些旧的套接字代码来使用TPL数据流和新的异步功能。尽管API感觉很稳定，但我的代码仍然感觉凌乱。我想知道我是否在这里错过了一些东西。

我的要求如下：一个套接字类公开：打开，关闭，发送和接收方法。所有返回一个任务，因此是异步的。打开和关闭是原子的。发送和接收可以彼此相邻，尽管两者一次只能处理1条命令。

逻辑上，这使我对内部控制的下一段代码：

// exposing an exclusive scheduler for connectivity related tasks and a parallel scheduler where send and receive can work with 
private readonly ConcurrentExclusiveSchedulerPair exclusiveConnectionSchedulerPair; 
private readonly ActionBlock<Action> connectionBlock; 
private readonly ActionBlock<Action> sendBlock; 
private readonly ActionBlock<Action> receiveBlock; 

// within the constructor: 
this.exclusiveConnectionSchedulerPair = new ConcurrentExclusiveSchedulerPair(); 
this.connectionBlock = new ActionBlock<Action>(action => action(), new ExecutionDataflowBlockOptions() { TaskScheduler = exclusiveConnectionSchedulerPair.ExclusiveScheduler }); 
this.sendBlock = new ActionBlock<Action>(action => action(), new ExecutionDataflowBlockOptions() { TaskScheduler = exclusiveConnectionSchedulerPair.ConcurrentScheduler }); 
this.receiveBlock = new ActionBlock<Action>(action => action(), new ExecutionDataflowBlockOptions() { TaskScheduler = exclusiveConnectionSchedulerPair.ConcurrentScheduler }); 

到目前为止都很好。我可以安全地将操作发送到发送和接收模块，而无需担心与此同时运行的连接相关操作。另外ActionBlock确保多个发送调用是同步的（同上接收，关闭和打开）。

问题是，没有简单的方法将任务传回给海报。现在我正在使用TaskCompletionSource传递结果。喜欢：

public Task Send(ArraySegment<byte> buffer, CancellationToken cancellationToken) 
{ 
    TaskCompletionSource<object> resultCompletionSource = new TaskCompletionSource<object>(); 

    sendBlock.Post(async() => 
    { 
     if (!tcpClient.Connected) 
      throw new InvalidOperationException("Cant send when not open"); 
     else 
     { 
      await sendStream.WriteAsync(buffer.Array, buffer.Offset, buffer.Count, cancellationToken); 
      resultCompletionSource.SetResult(null); 
     } 
    }); 

    return resultCompletionSource.Task; 
} 

这只是感觉丑陋和笨拙。我的问题是：有没有办法使用TPL同步工作流而不必使用TaskCompletionSource进行通信？

谢谢！

Answer 1

首先，您根本不需要TPL Dataflow，因为您实际上没有任何数据流。

其次，使用TaskScheduler这样的s也不是正确的解决方案。 TaskScheduler的日程安排代码，但是当你有一些东西时，没有代码在运行。因此，虽然WriteAsync()正在执行其异步工作，但代码Open()可以运行。

你真正需要的是像ReaderWriterLock，但与async效果很好。框架中没有类似的东西，但是您可以使用Stephen Toub's article Building Async Coordination Primitives, Part 7: AsyncReaderWriterLock中的代码，它完全符合您的需求。文章还详细解释了为什么使用TaskScheduler是错误的。

使用AsyncReaderWriterLock，你的代码可能是这样的：

public async Task Send(ArraySegment<byte> buffer, CancellationToken cancellationToken) 
{ 
    using (await readerWriterLock.ReaderLockAsync()) 
    { 
     if (!tcpClient.Connected) 
      throw new InvalidOperationException("Can't send when not open"); 

     await sendStream.WriteAsync(buffer.Array, buffer.Offset, buffer.Count, cancellationToken); 
    } 
}