取消不接受CancellationToken的异步操作的正确方法是什么？

Question

取消以下内容的正确方法是什么？

var tcpListener = new TcpListener(connection);
tcpListener.Start();
var client = await tcpListener.AcceptTcpClientAsync();

简单地调用tcpListener.Stop()似乎导致a ObjectDisposedException并且该AcceptTcpClientAsync方法不接受CancellationToken结构.

我完全错过了一些明显的东西吗

Answer 1

假设您不想在类上调用该Stop方法,这里没有完美的解决方案.TcpListener

如果您在某个时间范围内未完成操作时收到通知,但允许原始操作完成,那么您可以创建扩展方法,如下所示:

public static async Task<T> WithWaitCancellation<T>( 
    this Task<T> task, CancellationToken cancellationToken) 
{
    // The tasck completion source. 
    var tcs = new TaskCompletionSource<bool>(); 

    // Register with the cancellation token.
    using(cancellationToken.Register( s => ((TaskCompletionSource<bool>)s).TrySetResult(true), tcs) ) 
    {
        // If the task waited on is the cancellation token...
        if (task != await Task.WhenAny(task, tcs.Task)) 
            throw new OperationCanceledException(cancellationToken); 
    }

    // Wait for one or the other to complete.
    return await task; 
}

以上内容来自Stephen Toub的博客文章"如何取消不可取消的异步操作？" .

这里需要说明的值得重复,这实际上并没有取消操作,因为没有了过载AcceptTcpClientAsync方法,需要一个CancellationToken,它不能够被取消.

这意味着如果扩展方法指示取消确实发生,则取消原始回调的等待Task,而不是取消操作本身.

为此,这就是为什么我改名的方法,从WithCancellation到WithWaitCancellation,以表明您要取消的等待,而不是实际行动.

从那里,它很容易在您的代码中使用:

// Create the listener.
var tcpListener = new TcpListener(connection);

// Start.
tcpListener.Start();

// The CancellationToken.
var cancellationToken = ...;

// Have to wait on an OperationCanceledException
// to see if it was cancelled.
try
{
    // Wait for the client, with the ability to cancel
    // the *wait*.
    var client = await tcpListener.AcceptTcpClientAsync().
        WithWaitCancellation(cancellationToken);
}
catch (AggregateException ae)
{
    // Async exceptions are wrapped in
    // an AggregateException, so you have to
    // look here as well.
}
catch (OperationCancelledException oce)
{
    // The operation was cancelled, branch
    // code here.
}

请注意,OperationCanceledException如果等待被取消,您必须为客户端包装调用以捕获抛出的实例.

我也抛出一个AggregateExceptioncatch,因为从异步操作中抛出异常(在这种情况下你应该自己测试).

这就留下了一个问题,即在采用类似Stop方法的方法时,哪种方法是更好的方法(基本上,任何事情都会猛烈地撕下一切,无论发生什么事情),这当然取决于你的情况.

如果你没有共享你正在等待的资源(在这种情况下,那么TcpListener),那么可能更好地利用资源来调用abort方法并吞下来自你正在等待的操作的任何异常(当你调用停止并监视你正在等待某个操作的其他区域中的那个位时,你将不得不翻转一下.这会增加代码的复杂性,但如果您担心资源利用率和尽快清理,并且您可以选择此选项,那么这就是您要做的.

如果资源利用率不是问题,并且您对更合作的机制感到满意,并且您没有共享资源,那么使用该WithWaitCancellation方法就可以了.这里的优点是代码更清晰,更易于维护.

Answer 2

虽然casperOne的答案是正确的,但实现相同目标的WithCancellation(或WithWaitCancellation)扩展方法有更清晰的潜在实现:

static Task<T> WithCancellation<T>(this Task<T> task, CancellationToken cancellationToken)
{
    return task.IsCompleted
        ? task
        : task.ContinueWith(
            completedTask => completedTask.GetAwaiter().GetResult(),
            cancellationToken,
            TaskContinuationOptions.ExecuteSynchronously,
            TaskScheduler.Default);
}

• 首先,我们通过检查任务是否已完成来进行快速路径优化.
• 然后我们只需将延续注册到原始任务并传递CancellationToken参数.
• 如果可能(TaskContinuationOptions.ExecuteSynchronously),则继续提取原始任务的结果(如果有的话,则为异常),ThreadPool如果不是(TaskScheduler.Default)则使用线程,同时观察CancellationToken取消.

如果原始任务在CancellationToken取消之前完成,则返回的任务将存储结果,否则任务将被取消并TaskCancelledException在等待时抛出.