取消不同类型的异步时行为不一致

Question

我在取消不同类型的异步时遇到了看似不一致的行为问题。

为了重现问题，假设有一个函数接受“作业”列表（Async<_> 列表），等待它们完成并打印它们的结果。该函数还获取一个取消令牌，因此可以取消它：

let processJobs jobs cancel =
  Async.Start(async {
    try
      let! results = jobs |> Async.Parallel
      printfn "%A" results
    finally
      printfn "stopped"
  }, cancel)

函数是这样调用的：

let jobs = [job1(); job2(); job3(); job4(); job5()]
use cancel = new CancellationTokenSource()

processJobs jobs cancel.Token

稍后它被取消了：

Thread.Sleep(1000)
printfn "cancelling..."
cancel.Cancel()

当取消令牌源被取消时，该函数应该执行finally-block并打印“stopped”。

这适用于 job1、2 和 3，但当列表中有 job4 或 job5 时不起作用。

Job1 只是 Async.Sleeps：

let job1() = async {
  do! Async.Sleep 1000000
  return 10
}

Job2 启动一些异步子节点并等待它们：

let job2() = async {
  let! child1 = Async.StartChild(async {
    do! Async.Sleep 1000000
    return 10
  })

  let! child2 = Async.StartChild(async {
    do! Async.Sleep 1000000
    return 10
  })

  let! results = [child1; child2] |> Async.Parallel
  return results |> Seq.sum
}

Job3 等待由更丑陋的线程设置的丑陋等待句柄：

let job3() = async {
  use doneevent = new ManualResetEvent(false)

  let thread = Thread(fun () -> Thread.Sleep(1000000); doneevent.Set() |> ignore)
  thread.Start()

  do! Async.AwaitWaitHandle(doneevent :> WaitHandle) |> Async.Ignore

  return 30
}

Job4 向 MailboxProcessor 发帖并等待回复：

let job4() = async {
  let worker = MailboxProcessor.Start(fun inbox -> async {
    let! (msg:AsyncReplyChannel<int>) = inbox.Receive()
    do! Async.Sleep 1000000
    msg.Reply 40
  })

  return! worker.PostAndAsyncReply (fun reply -> reply) // <- cannot cancel this
}

Job5 等待任务（或 TaskCompletionSource）：

let job5() = async {
  let tcs = TaskCompletionSource<int>()

  Async.Start(async {
    do! Async.Sleep 1000000
    tcs.SetResult 50
  })

  return! (Async.AwaitTask tcs.Task) // <- cannot cancel this
}

为什么 Job1、2 和 3 可以取消（“stopped”被打印），而 Job4 和 5 使函数“永远”挂起？

到目前为止，我一直依赖 F# 来处理幕后取消 - 只要我在异步块中并使用 !s（让！，做！，返回！，...）一切都应该没问题。 . 但似乎并非一直如此。

Quote:

在 F# 异步工作流中，传递 CancellationToken 对象 自动绕到盖下。这意味着我们不必 做任何特别的事情来支持取消。异步运行时 工作流程，我们可以给它取消令牌，一切都会正常工作 自动。

完整的代码在这里：http://codepad.org/euVO3xgP

编辑

我注意到通过 Async.StartAsTask 和 Async.AwaitTask 管道异步使其在所有情况下都可以取消。

即对于 Job4，这意味着换行：

return! worker.PostAndAsyncReply (fun reply -> reply)

到：

return! cancelable <| worker.PostAndAsyncReply (fun reply -> reply)

具有可取消的存在：

let cancelable (x:Async<_>) = async {
  let! cancel = Async.CancellationToken
  return! Async.StartAsTask(x, cancellationToken = cancel) |> Async.AwaitTask
}

使 Job5 可取消也是如此。

但是.. 这只是一种解决方法，我几乎无法在每次调用未知 Async<_> 时使用它。

Answer 1

只有异步。方法本身使用默认的 CancellationToken 进行处理。

在您的 MailboxProcessor 示例中，取消应该在 Start 方法上进行

let! ct= Async.CancellationToken
use worker := MailboxProcessor.Start( theWork, ct) 

在 TaskCompletionSource 示例中，您必须注册一个回调来取消它。

let! ct = Async.CancellationToken
use canceler = ct.Register( fun () -> tcs.TrySetCanceled() |> ignore )