如何调整 DatagramSocket.MessageReceived 以用于 async/await？答案

【问题标题】：How do I adapt DatagramSocket.MessageReceived for use with async/await?如何调整 DatagramSocket.MessageReceived 以用于 async/await？
【发布时间】：2013-02-08 20:54:21
【问题描述】：

我想用 TPL 包装以下数据报套接字操作以清理 API，以便它可以很好地与 async 和 await 一起工作，就像 StreamSocket 类一样。

public static async Task<bool> TestAsync(HostName hostName, string serviceName, byte[] data)
{
    var tcs = new TaskCompletionSource<bool>();
    var socket = new DatagramSocket();
    socket.MessageReceived += (sender, e) =>
    {
        var status = false; // Status value somehow derived from e etc.
        tcs.SetResult(status);
    };
    await socket.ConnectAsync(hostName, serviceName);
    var stream = await socket.GetOutputStreamAsync();
    var writer = new DataWriter(stream);
    writer.WriteBytes(data);
    await writer.StoreAsync();
    return tcs.Task;
}

关键点是MessageReceived 事件，它将DatagramSocket 类变成了事件异步模式和新async 模式的奇怪混搭。无论如何，TaskCompletionSource<T> 允许我调整处理程序以符合后者，所以这并不太可怕。

这似乎工作得很好，除非端点从不返回任何数据。与MessageReceived 处理程序关联的任务永远不会完成，因此从TestAsync 返回的任务永远不会完成。

正确包装此操作以包含超时和取消的正确方法是什么？我想扩展这个函数来为后者使用CancellationToken 参数，但我该怎么做呢？我想出的唯一方法是使用Task.Delay 创建一个额外的“监控”任务，我将超时值和取消令牌传递给它，以支持以下两种行为：

public static async Task<bool> CancellableTimeoutableTestAsync(HostName hostName, string serviceName, byte[] data, CancellationToken userToken, int timeout)
{
    var tcs = new TaskCompletionSource<bool>();
    var socket = new DatagramSocket();
    socket.MessageReceived += (sender, e) =>
    {
        var status = false; // Status value somehow derived from e etc.
        tcs.SetResult(status);
    };
    await socket.ConnectAsync(hostName, serviceName);
    var stream = await socket.GetOutputStreamAsync();
    var writer = new DataWriter(stream);
    writer.WriteBytes(data);
    await writer.StoreAsync();

    var delayTask = Task.Delay(timeout, userToken);
    var t1 = delayTask.ContinueWith(t => { /* Do something to tcs to indicate timeout */ }, TaskContinuationOptions.OnlyOnRanToCompletion);
    var t2 = delayTask.ContinueWith(t => { tcs.SetCanceled(); }, TaskContinuationOptions.OnlyOnCanceled);

    return tcs.Task;
}

但是，这会带来各种问题，包括延迟任务和MessageReceived 处理程序之间的潜在竞争条件。我从来没有能够让这种方法可靠地工作，而且它看起来非常复杂，而且线程池的使用效率低下。这很繁琐，容易出错，而且让我头疼。

旁注：我是唯一对DatagramSocket API 感到困惑的人吗？它不仅看起来像是IAsyncAction WinRT 模型和 TPL 的丑陋组合，并带有一些棘手的 EAP，而且我对旨在表示基本无连接协议（例如 UDP）的 API 不太满意，其中包含名为的方法ConnectAsync 在其中。这对我来说似乎是一个矛盾。

【问题讨论】：

创建一个新方法，将第一个任务与 Task.Delay 结合起来，并使用 Task.WhenAny，而不是整个第二个方法。一旦它返回，您要么完成任务，要么发生超时。

标签： c# network-programming windows-runtime task-parallel-library async-await

【解决方案1】：

首先，我认为DatagramSocket 的接口之所以有意义，正是因为UDP 的性质。如果您有数据报流，则事件是表示它的适当方式。 WinRT IAsyncAction（或 .Net Task）只能表示拉模型，在这种模型中，您明确地请求每条数据（例如，可能有一个方法 ReadNextDatagramAsync()）。这对 TCP 来说是有意义的，因为它有流量控制，所以如果你慢慢地读取数据，发送者也会慢慢地发送它们。但对于 UDP，推送模型（由 WinRT 和 .Net 中的事件表示）更有意义。

我同意 Connect 这个名字没有 100% 的意义，但我认为它大部分是有道理的，尤其是让它与 StreamSocket 更加一致。而且您确实需要这样的方法，以便系统可以解析域名并为您的套接字分配一个端口。

对于你的方法，我同意@usr 你应该创建一个单独的方法来接收数据报。而且，如果您想将一个异步模型转换为另一个，同时添加原始模型本身不支持的功能，那将会很繁琐，我认为您无能为力。

如果你正确实现它也不会低效：你应该确保在Task完成后，MessageReceived事件被取消订阅，与Delay()关联的计时器被释放（你做通过取消您传递给Delay() 的令牌，并且使用传入的CancellationToken 注册的委托未注册（我认为您应该直接使用Register() 而不是（ab）为此使用Delay()） .

关于比赛条件，您当然必须考虑它们。但是这里有一个相对简单的方法来处理它：使用TaskCompletionSource 的Try 方法（例如TrySetResult()）。

【讨论】：

【解决方案2】：

超时：启动计时器并使用tcs.TrySetCancelled() 完成任务。对于取消，使用cancellationToken.Register 注册一个您也设置为取消的回调。小心处理计时器。

我建议您将计时器逻辑移到可重用的辅助方法中。这可以防止代码看起来像意大利面条，其中混杂着许多不相关的东西。

【讨论】：