OnceAsync：正確に一度f＃async関数を実行する

Question

私は、非同期関数がサーバー（つまりマルチスレッド環境）上で一度しか実行されないようにする関数（OnceAsync f）を記述しようとしています。 （ロック、ビジー待機!!）私はそれが簡単だろうと思ったが、それはすぐに複雑になった

これが私の解決策ですが、私はそれが設計上-だと思います。より良い方法が必要です。これは、FSIで動作するはずです：OnceAsyncが正しいかどう

let locked_counter init = 
    let c = ref init 
    fun x -> lock c <| fun() -> 
     c := !c + x 
     !c 
let wait_until finished = async { 
    while not(finished()) do 
     do! Async.Sleep(1000) 
} 

let OnceAsync f = 
    // - ensure that the async function, f, is only called once 
    // - this function always returns the value, f() 
    let mutable res = None 
    let lock_inc = locked_counter 0 

    async { 
     let count = lock_inc 1 

     match res, count with 
     | None, 1 -> // 1st run 
      let! r = f 
      res <- Some r 
     | None, _ -> // nth run, wait for 1st run to finish 
      do! wait_until (fun() -> res.IsSome) 
     | _ ->()  // 1st run done, return result 

     return res.Value 
    } 

あなたがテストするには、このコードを使用することができます：それは収まる場合

let test() = 
    let mutable count = 0 

    let initUser id = async { 
     do! Async.Sleep 1000 // simulate work 
     count <- count + 1 
     return count 
    } 

    //let fmem1 = (initUser "1234") 
    let fmem1 = OnceAsync (initUser "1234") 

    async { 
     let ps = Seq.init 20 (fun i -> fmem1) 
     let! rs = ps |> Async.Parallel 
     printfn "rs = %A" rs  // outputs: [|1; 1; 1; 1; 1; ....; 1|] 
    } 

test() |> Async.Start 

Answer 1

は、最も簡単な方法は、全体的なAsync.StartChildを使用することです。ソリューションとは異なり、たとえ結果が実際には使用されない場合でも、関数は実行されます。 Seq.init 0の場合

//let fmem1 = OnceAsync (initUser "1234") 

async { 
    let! fmem1 = Async.StartChild (initUser "1234") 
    let ps = Seq.init 20 (fun i -> fmem1) 
    let! rs = ps |> Async.Parallel 
    printfn "rs = %A" rs  // outputs: [|1; 1; 1; 1; 1; ....; 1|] 
} |> Async.RunSynchronously 

あなたに最も類似した最も簡単な方法は、次のようにTaskCompletionSourceを使用することです：

let OnceAsync f = 
    let count = ref 0 
    let tcs = TaskCompletionSource<_>() 
    async { 
    if Interlocked.Increment(count) = 1 then 
     let! r = f 
     tcs.SetResult r 
    return! Async.AwaitTask tcs.Task 
    } 

より機能的なアプローチはMailboxProcessorを使用して、それが最初の実行後に結果をキャッシュだろう、とすべての後続の要求に応答します。

let OnceAsync f = 
    let handler (agent: MailboxProcessor<AsyncReplyChannel<_>>) = 
    let rec run resultOpt = 
     async { 
     let! chan = agent.Receive() 
     let! result = 
      match resultOpt with 
      | None -> f 
      | Some result -> async.Return result 
     chan.Reply result 
     return! run (Some result) 
     } 
    run None 
    let mbp = MailboxProcessor.Start handler 
    async { return! mbp.PostAndAsyncReply id }