等待取消异步工作流程
CancellationTokenSource
对象的CancellationTokenSource
成员“传递取消请求”,我认为这意味着它是火并忘记,并且不会等到取消完成为止(例如,所有异常处理程序已经运行)。这很好,但我需要等到一个出色的异步完全取消之后再创建另一个异步。有没有简单的方法来完成这个?等待取消异步工作流程
我不认为有任何直接的办法,使用标准库函数从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。
这应该不难给予使用方便同步原语。我特别喜欢一次性写入“逻辑”的变量:
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
你应该处理'CancellationTokenSource'吗? – 2012-07-23 13:23:19
我认为这很重要。我曾经写过关于'FSharp.Core'中的泄漏,我认为这是由于完全相同的问题导致的,而不是处置CTS:http://t0yv0.blogspot.com/2011/12/solving-f-asyncstartchild-leak- futures.html – t0yv0 2012-07-23 14:17:10
@JonHarrop这是一个很好的观点。我不确定在这种情况下是否会导致泄漏,但最好调用Dispose。在计算被取消(并且取消完成)之后,我编辑了在终结器中调用“Dispose”的答案。 – 2012-07-23 14:44:26