asynchronous - 如何确保 Async.StartChild 在继续之前已启动？

Question

我正在尝试等待超时事件。我在函数 startAwaitEventWithTimeout 后面对此进行了抽象。目前我的代码看起来像这样(包括一些调试输出消息):

let startAwaitEventWithTimeout timeoutMs event =
  async {
    Console.WriteLine("Starting AwaitEvent in eventAwaiter")
    let! eventWaiter = Async.StartChild(Async.AwaitEvent event, timeoutMs)
    try
      Console.WriteLine("Awaiting event in eventAwaiter")
      let! res = eventWaiter
      return Ok res
    with :? TimeoutException ->
      return Error ()
  } |> Async.StartChild

这是一个测试:

let testEvent = Event<string>()

[<EntryPoint>]
let run _ =
  async {
    Console.WriteLine("Starting event awaiter in main")
    let! eventAwaiter = testEvent.Publish |> startAwaitEventWithTimeout 1000

    Console.WriteLine("Triggering event")
    testEvent.Trigger "foo"
    Console.WriteLine("Awaiting event awaiter in main")
    let! result = eventAwaiter

    match result with
    | Ok str -> Console.WriteLine("ok: " + str)
    | Error () -> Console.WriteLine("TIMEOUT")
  } |> Async.RunSynchronously
  0

不幸的是，尽管据我所知一切都在“等待”，但 run 函数似乎在 Async.AwaitEvent 有一个事件之前继续触发事件有机会订阅事件。简而言之，这是我得到的输出:

Starting event awaiter in main
Starting AwaitEvent in eventAwaiter
Triggering event
Awaiting event awaiter in main
Awaiting event in eventAwaiter
TIMEOUT

这是我所期望的:

Starting event awaiter in main
Starting AwaitEvent in eventAwaiter
Awaiting event in eventAwaiter  <-- this is moved up
Triggering event
Awaiting event awaiter in main
ok foo

我可以通过添加例如做! Async.Sleep 100 在调用 startAwaitEventWithTimeout 和触发事件之间，但这当然不太理想。

我是否做错了什么，有什么方法可以可靠地确保在触发事件之前调用了 AwaitEvent？

(旁注:我这样做是因为我们正在通过 TCP 调用远程进程，并且来自远程的所有通信都是通过事件完成的。)

Answer 1

可能我遗漏了一些要求，但您的代码可以很容易地使用延续进行重构，并且可以自行修复错误。

let testEvent = Event<unit>()

let run _ =
  let ts = new CancellationTokenSource(TimeSpan.FromSeconds(float 1))
  let rc r =  Console.WriteLine("ok")
  let ec _ =  Console.WriteLine("exception")
  let cc _ =  Console.WriteLine("cancelled")
  Async.StartWithContinuations((Async.AwaitEvent testEvent.Publish), rc , ec,  cc, ts.Token  )
  testEvent.Trigger()
run()

编辑:如果您有使用异步工作流的特定要求，您可以使用 TPL 中的 TaskCompletionSource 进行转换。

let registerListener  timeout event= 
  let tcs = TaskCompletionSource()
  let ts = new CancellationTokenSource(TimeSpan.FromSeconds(timeout))
  let er _ =  tcs.SetResult (Error())
  Async.StartWithContinuations(Async.AwaitEvent event, tcs.SetResult << Ok , er , er , ts.Token)
  Async.AwaitTask tcs.Task

let run _ =
  let testEvent = Event<int>()
  async {
       let listener = registerListener (float 1) testEvent.Publish
       testEvent.Trigger 2
       let! ta  = listener 
       match ta with
         | Ok n -> printfn "ok: %d" n
         | Error () -> printfn "error"
  } |> Async.RunSynchronously

run()

请注意，尽管它比产生/等待多个子计算更容易理解，但大部分代码仍然是样板代码，我相信必须有更简单的解决方案来设置简单的超时值。