【问题标题】:Waiting on the cancellation of an asynchronous workflow等待取消异步工作流
【发布时间】:2012-07-23 08:42:31
【问题描述】:

CancellationTokenSource 对象的 Cancel 成员“传达取消请求”,我认为这意味着它是触发并忘记并且不会等到取消完成(例如,所有异常处理程序都已运行)。这很好,但我需要等到一个未完成的异步完全取消后再创建另一个异步。有没有简单的方法可以做到这一点?

【问题讨论】:

    标签: asynchronous concurrency f#


    【解决方案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) }
    

    EDIT 将结果包装在 TryFinally 中,以按照 Jon 的建议处理 CancellationTokenSource。我认为这足以确保正确处理它。

    这是一个使用该方法的示例。 loop 函数是我用于测试的简单工作流程。其余代码启动它,等待 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

    【讨论】:

    • 你应该处理CancellationTokenSource吗?
    • 我认为这很重要。我曾经在FSharp.Core 中写过一个泄漏,我认为这是由完全相同的问题引起的,而不是处理 CTS:t0yv0.blogspot.com/2011/12/…
    • @JonHarrop 这是一个很好的观点。我不确定在这种情况下是否会导致泄漏,但最好致电Dispose。在计算被取消(并且取消完成)之后,我在终结器中编辑了调用 Dispose 的答案。
    • @TomasPetricek 我认为如果您在循环中调用代码可能会发生泄漏,例如在服务器中一遍又一遍地调用 StartCancellable。使用 Dispose 应该没问题,感谢您的更正。
    • @TomasPetricek 一个可以说是代码不直观的方面是,如果工作代码正常完成或在取消发生之前出现异常,则等待取消会无限期地等待。这就是我更喜欢我的答案的原因,尽管它可能与你的答案相同。
    【解决方案2】:

    由于易于使用的同步原语,这应该不难。我特别喜欢只写一次的“逻辑”变量:

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

    很容易封装一个 Async 来在完成时设置一个逻辑变量,然后等待它,例如:

    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
    

    【讨论】:

      猜你喜欢
      • 2020-04-23
      • 1970-01-01
      • 1970-01-01
      • 2014-06-06
      • 2021-10-16
      • 2019-01-13
      • 1970-01-01
      • 2019-06-30
      • 1970-01-01
      相关资源
      最近更新 更多