取消F#异步工作流的子块

Question

我正在尝试创建一个异步工作流，其中有一个主异步循环，在每个循环中执行异步子块。 并且我希望这个异步子块可以被取消，但当它被取消时，我不希望主循环被取消。 我希望它在do! subBlock之后继续运行。

我只看到Async中有一个可接受签名的方法（使用CancellationToken，返回可转换为async的内容），即Async.StartAsTask，但似乎它在取消时会挂起； 在下面的代码中，它打印“已取消”，然后什么也没有。

open System
open System.Threading
open System.Threading.Tasks

// runs until cancelled
let subBlock = 
  async { 
    try 
      while true do
        printfn "doing it"
        do! Async.Sleep 1000
        printfn "did it"
    finally
      printfn "cancelled!"
  }

[<EntryPoint>]
let main argv = 
  let ctsRef = ref <| new CancellationTokenSource()      

  let mainBlock = 
    //calls subBlock in a loop
    async { 
      while true do
        ctsRef := new CancellationTokenSource()
        do! Async.StartAsTask(subBlock, TaskCreationOptions.None, (!ctsRef).Token) 
            |> Async.AwaitTask
        printfn "restarting"
    }
  Async.Start mainBlock

  //loop to cancel CTS at each keypress
  while true do
    Console.ReadLine() |> ignore
    (!ctsRef).Cancel()
  0

有没有办法做到这一点？

- Dax Fohl

Answer 1

无论启动和取消工作的调用者是否也是异步，都不会影响此问题，因为工作者是通过其明确指定的取消令牌来管理的。

异步操作具有三个连续性：正常的连续性，可以返回值；异常的连续性；以及取消的连续性。有多种方法可以向异步操作添加取消连续性，例如 Async.OnCancel、Async.TryCancelled 或通用的 Async.FromContinuations，其中包括异常情况。下面是一个具有所需输出的程序：

let rec doBlocks () = 
    async { printfn "doing it"
            do! Async.Sleep 1000
            printfn "did it"
            do! doBlocks () }

let rec runMain () =
    use cts = new CancellationTokenSource()
    let worker = Async.TryCancelled(doBlocks (), fun _ -> printfn "Cancelled")
    Async.Start(worker, cts.Token)
    let k = Console.ReadKey(true)
    cts.Cancel()
    if k.Key <> ConsoleKey.Q then runMain ()

如果runMain是异步的，这也同样有效。在这种简单情况下，您也可以让它自己打印“取消”消息。

希望这可以帮助您。我认为如何构建程序没有通用答案；这取决于具体的用例。

- Vandroiy

Answer 2

这里发生的情况是，当您的子任务被取消时，OperationCanceledException 也会使您的 mainBlock 停止。我通过使用以下代码使其正常工作：

let rec mainBlock = 
    async {
        ctsRef := new CancellationTokenSource()
        let task = Async.StartAsTask(subBlock, TaskCreationOptions.None, (!ctsRef).Token) |> Async.AwaitTask
        do! Async.TryCancelled(task, fun e -> 
            (!ctsRef).Dispose() 
            printfn "restarting" 
            Async.Start mainBlock)
    }

当任务被取消时，取消处理程序会明确地重新启动mainBlock。由于mainBlock在其定义内部使用，因此需要添加#nowarn "40"。还要注意标记源上的dispose。

您可以在这些 two threads 中找到关于此问题的更多信息（以及可能以StartCatchCancellation形式呈现的更好的解决方案）。

- scrwtp