TPL数据流如何与异步操作配合使用

Question

TPL数据流如何与异步操作配合使用

c#.nettask-parallel-librarydataflowtpl-dataflow

4

我正在尝试使用TPL数据流将一些旧的套接字代码移植到使用TPL数据流和新的异步功能。虽然API感觉非常稳定，但我的代码仍然感觉杂乱无章。我想知道我是否漏掉了什么。

我的要求如下：一个套接字类公开了Open、Close、Send和Receive方法。所有方法都返回一个任务，并且都是异步的。Open和Close是原子性的。Send和Receive可以同时工作，但两者都只能处理一个命令。

从逻辑上讲，这将带我进入下一个用于内部控制的代码片段：

// 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来进行通信？

谢谢！

- Polity

1

我没有使用过TPL DataFlow，所以可能会有些偏差，但如果您需要知道操作何时完成，那么似乎您不希望使用Post的“fire and forget”，而是应该使用SendAsync，这样您就可以获得一个Task了。参见：http://msdn.microsoft.com/en-us/library/hh194681(v=vs.110) - James Manning

2

据我理解，当请求被接受或明确拒绝时，SendAsync返回一个任务。由于我正在使用无限制的操作块，因此请求将始终直接被接受。在我的情况下，SendAsync无法提供帮助。 - Polity

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- svick · Accepted Answer

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

其次，像这样使用TaskScheduler也不是正确的解决方案。 TaskScheduler调度代码，但在你await某些东西时，没有代码正在运行。因此，在WriteAsync()执行其异步工作时，Open()的代码可能会运行。

实际上，你需要的是类似于ReaderWriterLock的东西，但它可以很好地与async配合使用。框架中没有类似的东西，但是你可以使用Stephen Toub的文章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);
    }
}