模拟 ReadAsync答案

【问题标题】：Emulating ReadAsync模拟 ReadAsync
【发布时间】：2012-09-27 23:32:59
【问题描述】：

在 .Net 4.5 中，Stream 类中存在一个方法，该方法在监视取消令牌的同时从流中异步读取。

ReadAsync : 
       buffer:byte[] * 
       offset:int * 
       count:int * 
       cancellationToken:CancellationToken -> Task<int>

如果另一个线程应该先触发取消令牌然后关闭流，那么是否保证读取线程在ReadAsync抛出异常之前被取消？

我能否使用 .Net 4.0 框架和 F# 异步工作流而不使用 ReadAsync（它没有接受要监视的取消令牌的重载）以某种方式实现此保证？

【问题讨论】：

标签： .net asynchronous f#

【解决方案1】：

除非您将多个任务链接在一起，否则这种特殊的重载基本上是无用的 - cancellationToken 仅在进入 ReadAsync 方法调用时检查，而不是在底层 Stream.BeginRead 调用正在执行时检查。

从 ILSpy 转储的代码：

public virtual Task<int> ReadAsync(byte[] buffer,
                                   int offset,
                                   int count,
                                   CancellationToken cancellationToken)
{
    if (!cancellationToken.IsCancellationRequested)
        return this.BeginEndReadAsync(buffer, offset, count);
    return Task.FromCancellation<int>(cancellationToken);
}

如你所见，cancellationToken 没有被转发到BeginEndReadAsync 调用中，而BeginEndReadAsync 只是在Stream.BeginRead 方面实现：

private Task<int> BeginEndReadAsync(byte[] buffer, int offset, int count)
{
    return TaskFactory<int>.FromAsyncTrim<Stream, Stream.ReadWriteParameters>(
        this,
        new Stream.ReadWriteParameters
        {
            Buffer = buffer,
            Offset = offset,
            Count = count
        },
        (Stream stream, Stream.ReadWriteParameters args, AsyncCallback callback, object state) =>
            stream.BeginRead(args.Buffer, args.Offset, args.Count, callback, state),
        (Stream stream, IAsyncResult asyncResult) =>
            stream.EndRead(asyncResult)
    );
}

此时，您拥有的唯一保证是派生流类型做出的保证，这些类型因类型而异。

^{请注意，这是基于当前的.Net 4.5位，实现当然会在未来发生变化。}

【讨论】：

【解决方案2】：

您可以在使用Result/RunSynchronously 运行任务之前检查Task.IsCanceled。

下面是一些示例代码：

use stream = new MemoryStream(Array.init 1000 (fun i -> byte (i % int Byte.MaxValue)))
use waitHandle = new EventWaitHandle(false, EventResetMode.ManualReset)
use cts = new CancellationTokenSource()
let thread = Thread(fun () -> 
  Thread.Sleep(1000)
  let buf = Array.zeroCreate 100
  let task = stream.ReadAsync(buf, 0, buf.Length, cts.Token)
  if not task.IsCanceled then task.RunSynchronously()
  waitHandle.Set() |> ignore)
thread.Start()
cts.Cancel()
waitHandle.WaitOne() |> ignore

但是一旦ReadAsync 开始，它将抛出一个AggregateException，说明任务已被取消。

【讨论】：

【解决方案3】：

AFAIK，您必须同时使用互斥锁来保护流并使用取消令牌来检查线程是否有待取消。 Stream 上没有异步原语可以为您处理这个问题。

【讨论】：