F # jak uruchomić kilka asynchrous zadań i czekać na wynik po raz pierwszy zakończone?

Question

Istnieje wiele przykładów, jak wykonać zadania async w F # jak

[dowork 1; work 2] 
|> Async.Parallel 
|> Async.RunSynchronously 

Ale jak mogę asynchronicznie czekać tylko pierwszy wynik?

Na przykład, jeśli chcę uruchomić kilka równoległych zadań wyszukiwania i chcę przeszukać głębiej, gdy zostanie uzyskany pierwszy pomyślny wynik.

Answer 1

Najprostsza możliwa realizacja mogłem pomyśleć wyglądać tak:

open FSharp.Control 

let getOneOrOther() = 
    let queue = BlockingQueueAgent(1) 
    let async1 = async { 
      do! Async.Sleep (System.Random().Next(1000, 2000)) 
      do! queue.AsyncAdd(1) } |> Async.Start 
    let async2 = async { 
      do! Async.Sleep (System.Random().Next(1000, 2000)) 
      do! queue.AsyncAdd(2) } |> Async.Start 

    queue.Get() 

for i in 1..10 do 
    printfn "%d" <| getOneOrOther() 

Console.ReadLine() |> ignore 

Opiera się on realizacji kolejki blokowanie z projektu FSharpx, które prawdopodobnie zechcesz z innych powodów. Ale jeśli nie chcesz żadnych zależności, zawiera także kolejkę blokującą, z nieco mniej ładnym interfejsem.

Aby uzyskać bardziej ogólną wersję z wbudowanym anulowaniem, poniższa wersja pobiera Seq<unit -> Async<'T>> i zwraca pierwszy wynik, przez który przechodzi, anulując wszystkie pozostałe.

open FSharp.Control 
open System.Threading 

let async1() = async { 
     do! Async.Sleep (System.Random().Next(1000, 2000)) 
     return 1 } 
let async2() = async { 
     do! Async.Sleep (System.Random().Next(1000, 2000)) 
     return 2 } 

let getFirst asyncs = 
    let queue = BlockingQueueAgent(1) 
    let doWork operation = async { 
      let! result = operation() 
      do! queue.AsyncAdd(result) } 
    let start work = 
     let cts = new CancellationTokenSource() 
     Async.Start(work, cts.Token) 
     cts 
    let cancellationTokens = 
     asyncs 
     |> Seq.map doWork 
     |> Seq.map start 
    let result = queue.Get() 
    cancellationTokens 
    |> Seq.iter (fun cts -> cts.Cancel(); cts.Dispose()) 
    result 

for i in 1..10 do 
    printfn "%A" <| getFirst [async1;async2] 

Console.ReadLine() |> ignore 

Answer 2

Roztwór ogólnego zastosowania można znaleźć w poniższym przykładzie: http://fssnip.net/dN

Async.Choice mogą być osadzone w każdym asynchronicznego obiegu, podobnie jak Async.Parallel. Opcjonalny typ wyjścia koduje możliwość wykonania obliczeń podrzędnych bez satysfakcjonującego wyniku.

Answer 3

Wydaje się, że takie rozwiązanie jest dość proste, nonblocking i działa na moim przypadku

let any (list: Async<'T>[])= 
    let tcs = new TaskCompletionSource<'T>() 

    list |> Array.map (fun wf->Async.Start (async{ 
       let! res=wf 
       tcs.TrySetResult (res) |> ignore 
      })) 
     |> ignore 

    Async.AwaitTask tcs.Task 

let async1 = async { 
     do! Async.Sleep (System.Random().Next(1000, 2000)) 
     return 1 } 
let async2 = async { 
     do! Async.Sleep (System.Random().Next(1000, 2000)) 
     return 2 } 

printfn "%d" <| ([|async1;async2|] |> any |> Async.RunSynchronously) 

Answer 4

innym przykładzie wykonania w oparciu o zdarzenie:

let Choice (asyncs: seq<Async<'T>>) : Async<'T> = 
    async { 
     let e = Event<'T>() 
     let cts = new System.Threading.CancellationTokenSource() 
     do Async.Start(
      asyncs 
      |> Seq.map (fun a -> async { let! x = a in e.Trigger x }) 
      |> Async.Parallel 
      |> Async.Ignore, 
      cts.Token) 
     let! result = Async.AwaitEvent e.Publish 
     cts.Cancel() 
     return result 
    } 

Answer 5

użyję coś takiego:

let any asyncs = 
    async { 
     let t = 
      asyncs 
      |> Seq.map Async.StartAsTask 
      |> System.Threading.Tasks.Task.WhenAny 
     return t.Result.Result }