multithreading - 加入第一个完成的线程？

Question

我正在用 F# 编写一系列图形搜索算法，并认为利用并行化会很好。我想并行执行多个线程并获取第一个线程的结果以完成。我有一个实现，但它并不漂亮。

两个问题:这种函数有标准名称吗？不是 Join 或 JoinAll，而是 JoinFirst？其次，有没有更惯用的方法来做到这一点？

//implementation
let makeAsync (locker:obj) (shared:'a option ref) (f:unit->'a) =
    async {
        let result = f()
        Monitor.Enter locker
        shared := Some result
        Monitor.Pulse locker
        Monitor.Exit locker
    }

let firstFinished test work =
    let result = ref Option.None
    let locker = new obj()
    let cancel = new CancellationTokenSource()    
    work |> List.map (makeAsync locker result) |> List.map (fun a-> Async.StartAsTask(a, TaskCreationOptions.None, cancel.Token)) |> ignore
    Monitor.Enter locker
    while (result.Value.IsNone || (not <| test result.Value.Value)) do
        Monitor.Wait locker |> ignore
    Monitor.Exit locker
    cancel.Cancel()
    match result.Value with
    | Some x-> x
    | None -> failwith "Don't pass in an empty list"
//end implentation

//testing
let delayReturn (ms:int) value = 
    fun ()->
        Thread.Sleep ms
        value

let test () =
    let work = [ delayReturn 1000 "First!"; delayReturn 5000 "Second!" ]
    let result = firstFinished (fun _->true) work
    printfn "%s" result

Answer 1

是否可以将 CancellationTokenSource 和 test 传递给每个异步并让第一个计算出有效结果的取消其他的？

let makeAsync (cancel:CancellationTokenSource) test f =
  let rec loop() =
    async {
      if cancel.IsCancellationRequested then 
        return None
      else
        let result = f()
        if test result then
          cancel.Cancel()
          return Some result
        else return! loop()
    }
  loop()

let firstFinished test work =
  match work with
  | [] -> invalidArg "work" "Don't pass in an empty list"
  | _ ->
    let cancel = new CancellationTokenSource()    
    work
    |> Seq.map (makeAsync cancel test) 
    |> Seq.toArray
    |> Async.Parallel
    |> Async.RunSynchronously
    |> Array.pick id

这种方法有几处改进:1)它只使用async(它没有与Task混合，后者是做同样事情的替代方案-- async 在 F# 中更为惯用； 2) 除了专为此目的而设计的 CancellationTokenSource 之外，没有共享状态； 3) 干净的函数链方法可以轻松地向管道添加额外的逻辑/转换，包括简单地启用/禁用并行性。