F # come eseguire diverse operazioni asincrone e attendere il risultato del primo completamento?

Question

Ci sono molti esempi di come eseguire compiti asincroni in F # come

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

ma come posso in modo asincrono aspettare solo per il primo risultato?

Ad esempio, se si desidera eseguire alcune attività di ricerca parallela e si desidera eseguire una ricerca più approfondita quando viene ottenuto il primo risultato riuscito.

Answer 1

Il più semplice possibile implementazione mi veniva in mente simile a questa:

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 

Essa si basa sulla realizzazione di coda di blocco dal progetto FSharpx, che probabilmente si vuole per altri motivi. Ma se non si desidera alcuna dipendenza, System.Collections.Concurrent include anche una coda di blocco, con un'interfaccia leggermente meno piacevole.

Per una versione più generale con cancellazione incorporata, la versione di seguito prende uno Seq<unit -> Async<'T>> e restituisce il primo risultato a venire attraverso, annullando tutti gli altri.

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

Una soluzione polivalente può essere trovato nel seguente frammento: http://fssnip.net/dN

Async.Choice può essere integrato in qualsiasi workflow asincrono, come Async.Parallel. Il tipo di output facoltativo codifica la possibilità che un calcolo figlio possa essere completato senza un risultato soddisfacente.

Answer 3

sembra che questa soluzione è abbastanza semplice, non-blocking e lavora per il mio caso

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

Un'altra implementazione in base a un evento:

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

userei qualcosa di simile:

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