等待取消异步工作流程

Question

CancellationTokenSource对象的CancellationTokenSource成员“传递取消请求”，我认为这意味着它是火并忘记，并且不会等到取消完成为止（例如，所有异常处理程序已经运行）。这很好，但我需要等到一个出色的异步完全取消之后再创建另一个异步。有没有简单的方法来完成这个？

Answer 1

我不认为有任何直接的办法，使用标准库函数从F＃异步库。其中，当工作流运行的回调最近的操作我们Async.TryCancelled（实际上可以）取消，但发送从回调的消息到开始工作流的代码必须由手工完成。

这是比较容易的使用事件和从F＃异步扩展，我写（也包含在FSharpX包）的扩展来解决 - 扩展名是GuardedAwaitObservable可用于等待一个事件的发生（这可通过一些操作立即触发）。

以下Async.StartCancellable方法采用异步工作流程，并返回Async<Async<unit>>。当您绑定在外工作流程，它开始参数（如Async.StartChild），当您绑定在返回的内部工作流程，它取消了计算和等待，直到它实际上是取消：

open System.Threading 

module Async = 
    /// Returns an asynchronous workflow 'Async<Async<unit>>'. When called 
    /// using 'let!', it starts the workflow provided as an argument and returns 
    /// a token that can be used to cancel the started work - this is an 
    /// (asynchronously) blocking operation that waits until the workflow 
    /// is actually cancelled 
    let StartCancellable work = async { 
    let cts = new CancellationTokenSource() 
    // Creates an event used for notification 
    let evt = new Event<_>() 
    // Wrap the workflow with TryCancelled and notify when cancelled 
    Async.Start(Async.TryCancelled(work, ignore >> evt.Trigger), cts.Token) 
    // Return a workflow that waits for 'evt' and triggers 'Cancel' 
    // after it attaches the event handler (to avoid missing event occurrence) 
    let waitForCancel = Async.GuardedAwaitObservable evt.Publish cts.Cancel 
    return async.TryFinally(waitForCancel, cts.Dispose) } 

编辑包裹结果TryFinally由Jon的建议处置CancellationTokenSource的。我认为这应该足以确保它被正确处置。

下面是一个使用该方法的例子。功能是一个简单的工作流程，我用于测试。的代码的其余部分启动它，等待5.5秒，然后取消它：

/// Sample workflow that repeatedly starts and stops long running operation 
let loop = async { 
    for i in 0 .. 9999 do 
    printfn "Starting: %d" i 
    do! Async.Sleep(1000) 
    printfn "Done: %d" i } 

// Start the 'loop' workflow, wait for 5.5 seconds and then 
// cancel it and wait until it finishes current operation 
async { let! cancelToken = Async.StartCancellable(loop) 
     printfn "started" 
     do! Async.Sleep(5500) 
     printfn "cancelling" 
     do! cancelToken 
     printfn "done" } 
|> Async.Start 

为了完整起见，从用FSharpX必要的定义中，样品是here on F# snippets。

Answer 2

这应该不难给予使用方便同步原语。我特别喜欢一次性写入“逻辑”的变量：

type Logic<'T> = 
    new : unit -> Logic<'T> 
    member Set : 'T -> unit 
    member Await : Async<'T> 

这是很容易包裹一个异步设置完成时逻辑变量，然后等待就可以了，例如：

type IWork = 
    abstract member Cancel : unit -> Async<unit> 

let startWork (work: Async<unit>) = 
    let v = Logic<unit>() 
    let s = new CancellationTokenSource() 
    let main = async.TryFinally(work, fun() -> s.Dispose(); v.Set()) 
    Async.Start(main, s.Token) 
    { 
     new IWork with 
      member this.Cancel() = s.Cancel(); v.Await 
    } 

一种可能的逻辑变量的实现可能是：

type LogicState<'T> = 
    | New 
    | Value of 'T 
    | Waiting of ('T -> unit) 

[<Sealed>] 
type Logic<'T>() = 
    let lockRoot = obj() 
    let mutable st = New 
    let update up = 
     let k = 
      lock lockRoot <| fun() -> 
       let (n, k) = up st 
       st <- n 
       k 
     k() 

    let wait (k: 'T -> unit) = 
     update <| function 
      | New -> (Waiting k, ignore) 
      | Value value as st -> (st, fun() -> k value) 
      | Waiting f -> (Waiting (fun x -> f x; k x), ignore) 

    let await = 
     Async.FromContinuations(fun (ok, _, _) -> wait ok) 

    member this.Set<'T>(value: 'T) = 
     update <| function 
      | New -> (Value value, ignore) 
      | Value _ as st -> (st, ignore) 
      | Waiting f as st -> (Value value, fun() -> f value) 

    member this.Await = await