asynchronous - 选择第一个异步结果

Question

是否有异步运算符获取两个异步值(Async<_>)首先返回的值？

例如，给定两个 Async<_> 值，其中一个 A1 在 1 秒后返回，A2 在 2 秒后返回，那么我想要 A1 的结果。

原因是我想为异步序列实现一个交错函数，这样如果有两个像这样“定义”的异步序列(用空格表示时间，就像大理石图一样):

S1 = -+-----+------------+----+
S2 = ---+-------+----------+-----+

然后我想生成一个新的异步序列，其行为如下:

S3 = -+-+---+---+--------+-+--+--+

交织 S1 S2 = S3

但是有两个这样做，我可能需要一种异步选择运算符来选择选择值。

我认为这就像 Go 中的“select”，您可以从两个 channel 中获取第一个可用值。

TPL 有一个名为 Task.WhenAny 的函数 - 我可能需要类似的东西。

Answer 1

我认为该运算符在 F# 库中不可用。要将此与现有操作相结合，您可以使用 Async.StartAsTask，然后使用现有的 Task.WhenAny 运算符。但是，我不确定在取消方面会有什么表现。

另一种选择是使用在 F# Snippets web site 上实现的 Async.Choose 运算符.这不是特别优雅，但应该可以解决问题!为了使答案独立，代码附在下面。

/// Creates an asynchronous workflow that non-deterministically returns the 
/// result of one of the two specified workflows (the one that completes
/// first). This is similar to Task.WaitAny.
static member Choose(a, b) : Async<'T> = 
    Async.FromContinuations(fun (cont, econt, ccont) ->
      // Results from the two 
      let result1 = ref (Choice1Of3())
      let result2 = ref (Choice1Of3())
      let handled = ref false
      let lockObj = new obj()
      let synchronized f = lock lockObj f

      // Called when one of the workflows completes
      let complete () = 
        let op =
          synchronized (fun () ->
            // If we already handled result (and called continuation)
            // then ignore. Otherwise, if the computation succeeds, then
            // run the continuation and mark state as handled.
            // Only throw if both workflows failed.
            match !handled, !result1, !result2 with 
            | true, _, _ -> ignore
            | false, (Choice2Of3 value), _ 
            | false, _, (Choice2Of3 value) -> 
                handled := true
                (fun () -> cont value)
            | false, Choice3Of3 e1, Choice3Of3 e2 -> 
                handled := true; 
                (fun () -> 
                    econt (new AggregateException
                                ("Both clauses of a choice failed.", [| e1; e2 |])))
            | false, Choice1Of3 _, Choice3Of3 _ 
            | false, Choice3Of3 _, Choice1Of3 _ 
            | false, Choice1Of3 _, Choice1Of3 _ -> ignore )
        op() 

      // Run a workflow and write result (or exception to a ref cell
      let run resCell workflow = async {
        try
          let! res = workflow
          synchronized (fun () -> resCell := Choice2Of3 res)
        with e ->
          synchronized (fun () -> resCell := Choice3Of3 e)
        complete() }

      // Start both work items in thread pool
      Async.Start(run result1 a)
      Async.Start(run result2 b) )