F# Async actions with conflicts / ordering / mutex

user_1256041 0
Viewed 342

F# makes it easy to define asynchronous computations using the async builder. You can write an entire program and then pass it to Async.RunSynchronously.

The problem I am having is that some async actions must not be run at the same time; they should be forced to wait for other async actions to complete. This is something like a mutex. However, I do not want to just chain them in serial because this would be inefficient.

Concrete Example: Download Cache

Suppose I want to fetch some remote files using a local file cache. In my application I call fetchFile : Async<string> in many places, but there is a risk that if I call fetchFile on the same URL at the same time, the cache will get corrupted by multiple writes. Instead, the fetchFile command should have behaviour like this:

  • If there is no cache, download the file to the cache and then read the cache contents
  • If the cache is currently being written to, wait for the write to finish and then read the contents
  • If the cache is present and complete, just read the cache contents
  • fetchFile on two different URLs should work in parallel

I am imagining some kind of stateful DownloadManager class to which requests can be sent and ordered internally.

How do F# programmers usually implement such logic with async?

Imaginary usage: 

let dm = new DownloadManager()

let urls = [
  "https://www.google.com"; 
  "https://www.google.com"; 
  "https://www.wikipedia.org"; 
  "https://www.google.com"; 
  "https://www.bing.com"; 
]

let results = 
  urls
  |> Seq.map dm.Download
  |> Async.Parallel
  |> Async.RunSynchronously

Note: I previously asked this question before about how to run async actions in a semi-parallel fashion, but now I have realized that this approach is hard to compose.

Note: I do not have to worry about multiple instances of the application running at once. In-memory locking is sufficient.

async-await mutex f#
3 Answers

UPDATE

Better than the lazy value is the Async.StartChild suggested by Petricek so I changed lazyDownload to asyncDownload

You could use a MailboxProcessor as a download manager that handles the cache. The MailboxProcessor is a structure in F# that processes a queue of messages ensuring no collisions.

First you need a processor capable of maintaining a state:

let stateFull hndl initState =
    MailboxProcessor.Start(fun inbox ->
        let rec loop state : Async<unit> = async {
            try         let! f        = inbox.Receive()
                        let! newState = f state
                        return! loop newState
            with e ->   return! loop (hndl e state)
        }
        loop initState
    )

The first parameter is a handler for errors, the second is the initial state, in this case a Map<string, Async<string>>. This is our downloadManager:

let downloadManager = 
    stateFull (fun e s -> printfn "%A" e ; s) (Map.empty : Map<string, _>)

To invoke the MailBox we need to use PostAndReply:

let applyReplyS f (agent: MailboxProcessor<'a->Async<'a>>) = 
    agent.PostAndReply(fun (reply:AsyncReplyChannel<'r>) -> 
        fun v -> async {
            let st, r = f v
            reply.Reply r
            return st 
        })

This function expects a folder function that checks the cache and adds an Async<string> if none is found and returns the updated cache.

First the asyncDownload function:

let asyncDownload url = 
    async { 
        let started = System.DateTime.UtcNow.Ticks
        do! Async.Sleep 30
        let finished = System.DateTime.UtcNow.Ticks
        let r = sprintf "Downloaded  %A it took: %dms %s" (started / 10000L) ((finished - started) / 10000L) url
        printfn "%s" r
        return r
    }

Just a dummy function that returns a string and timing information.

Now the folder function that checks the cache:

let folderCache url cache  =
    cache 
    |> Map.tryFind url
    |> Option.map(fun ld -> cache, ld)
    |> Option.defaultWith (fun () -> 
        let ld = asyncDownload url |> Async.StartChild |> Async.RunSynchronously
        cache |> Map.add url ld, ld
    )

finally our download function:

let downloadUrl url =
    downloadManager 
    |> applyReplyS (folderCache url)

// val downloadUrl: url: string -> Async<string>

Testing

let s = System.DateTime.UtcNow.Ticks
printfn "started %A" (s / 10000L)
let res = 
    List.init 50 (fun i -> i, downloadUrl (string <| i % 5) )
    |> List.groupBy (snd >> Async.RunSynchronously)
    |> List.map (fun (t, ts) -> sprintf "%s - %A" t (ts |> List.map fst ) )

let f = System.DateTime.UtcNow.Ticks
printfn "finish  %A" (f / 10000L)

printfn "elapsed %dms" ((f - s) / 10000L)

res |> printfn "Result: \n%A"

produces the following output:

started 63676683215256L
Downloaded  63676683215292L it took: 37ms "2"
Downloaded  63676683215292L it took: 36ms "3"
Downloaded  63676683215292L it took: 36ms "1"
Downloaded  63676683215291L it took: 38ms "0"
Downloaded  63676683215292L it took: 36ms "4"
finish  63676683215362L
elapsed 106ms
Result: 
["Downloaded  63676683215291L it took: 38ms "0" - [0; 5; 10; 15; 20; 25; 30; 35; 40; 45]";
 "Downloaded  63676683215292L it took: 36ms "1" - [1; 6; 11; 16; 21; 26; 31; 36; 41; 46]";
 "Downloaded  63676683215292L it took: 37ms "2" - [2; 7; 12; 17; 22; 27; 32; 37; 42; 47]";
 "Downloaded  63676683215292L it took: 36ms "3" - [3; 8; 13; 18; 23; 28; 33; 38; 43; 48]";
 "Downloaded  63676683215292L it took: 36ms "4" - [4; 9; 14; 19; 24; 29; 34; 39; 44; 49]"]
user_45508980

I agree with @AMieres that mailbox processor is a good way to do this. My version of the code is somewhat less general - it uses mailbox processor directly for this one purpose and so it might be a bit simpler.

Our mailbox processor has only one message - you ask it to download a URL and it gives you back an async workflow that you can wait for to get your result:

type DownloadMessage = 
  | Download of string * AsyncReplyChannel<Async<string>>

We need a helper function to asynchronously download a URL:

let asyncDownload url = async {
  let wc = new System.Net.WebClient()
  printfn "Downloading: %s" url
  return! wc.AsyncDownloadString(System.Uri(url)) }

In the mailbox processor, we keep a mutable cache (this is fine, because the mailbox processor processes messages synchronously). When we get a download request, we check if we already have download in the cache - if no, we start the download as a child async and add it to the cache - so the cache contains async workflows that represent results of a running download.

let downloadCache = MailboxProcessor.Start(fun inbox -> async {
  let cache = System.Collections.Generic.Dictionary<_, _>()
  while true do
    let! (Download(url, repl)) = inbox.Receive()
    if not (cache.ContainsKey url) then 
      let! proc = asyncDownload url |> Async.StartChild
      cache.Add(url, proc)
    repl.Reply(cache.[url]) })

To actually download using cache, we just send a request to the mailbox processor and then wait for the result of the returned workflow (which may be shared by multiple requests).

let downloadUsingCache url = async {
  let! res = downloadCache.PostAndAsyncReply(fun ch -> Download(url, ch))
  return! res }
Tomas_Petricek_335180

I am offering you a simplified version based on @Tomas Petricek answer.

Lets assume that we have our download function that given a url returns an Async<string>. This is a dummy version:

let asyncDownload url = 
    async { 
        let started = System.DateTime.UtcNow.Ticks
        do! Async.Sleep 30
        let finished = System.DateTime.UtcNow.Ticks
        let r = sprintf "Downloaded  %A it took: %dms %s" (started / 10000L) ((finished - started) / 10000L) url
        printfn "%s" r
        return r
    }

Here we have some simple generic Mailbox helper functions in their own module:

module Mailbox =
    let iterA hndl f =
        MailboxProcessor.Start(fun inbox ->
            async {
                while true do
                    try       let!   msg = inbox.Receive()
                              do!  f msg
                    with e -> hndl e
            }
        )
    let callA hndl f = iterA hndl (fun ((replyChannel: AsyncReplyChannel<_>), msg) -> async {
        let! r = f msg
        replyChannel.Reply r
    })
    let call hndl f = callA hndl (fun msg -> async { return f msg } )

The purpose of this 'library' is to simplify the more typical uses of the MailboxProcessor. While it looks complicated and hard to understand the important thing is what the functions do and how to use them. In particular we are going to use Mailbox.call which returns a Mailbox agent capable of returning a value. It's signature is: 

val call: 
   hndl: exn -> unit ->
   f   : 'a -> 'b    
      -> MailboxProcessor<AsyncReplyChannel<'b> * 'a>

The first parameter is an exception handler and the second a function that returns a value. Here is how we define our downloadManager:

let downloadManager = 
    let dict = new System.Collections.Generic.Dictionary<string, _>()
    Mailbox.call (printfn "%A") (fun url ->         
        if dict.ContainsKey url then dict.[url] else
        let result = asyncDownload url |> Async.StartChild |> Async.RunSynchronously
        dict.Add(url, result)
        result
    )

Our cache is a Dictionary. If a url is not present we call asyncDownload and start it as a child process. By using Async.StartChild we do not have to wait until it finishes downloading, we just return an async that waits for it to finish.

To invoke the manager we use downloadManager.PostAndReply

let downloadUrl url = downloadManager.PostAndReply(fun reply -> reply, url)

And here is a test:

let s = System.DateTime.UtcNow.Ticks
printfn "started %A" (s / 10000L)
let res = 
    List.init 50 (fun i -> i, downloadUrl (string <| i % 5) )
    |> List.groupBy (snd >> Async.RunSynchronously)
    |> List.map (fun (t, ts) -> sprintf "%s - %A" t (ts |> List.map fst ) )

let f = System.DateTime.UtcNow.Ticks
printfn "finish  %A" (f / 10000L)

printfn "elapsed %dms" ((f - s) / 10000L)

res |> printfn "Result: \n%A"

That produces:

started 63676682503885L
Downloaded  63676682503911L it took: 34ms 1
Downloaded  63676682503912L it took: 33ms 2
Downloaded  63676682503911L it took: 37ms 0
Downloaded  63676682503912L it took: 33ms 3
Downloaded  63676682503912L it took: 33ms 4
finish  63676682503994L
elapsed 109ms
Result: 
["Downloaded  63676682503911L it took: 37ms 0 - [0; 5; 10; 15; 20; 25; 30; 35; 40; 45]";
 "Downloaded  63676682503911L it took: 34ms 1 - [1; 6; 11; 16; 21; 26; 31; 36; 41; 46]";
 "Downloaded  63676682503912L it took: 33ms 2 - [2; 7; 12; 17; 22; 27; 32; 37; 42; 47]";
 "Downloaded  63676682503912L it took: 33ms 3 - [3; 8; 13; 18; 23; 28; 33; 38; 43; 48]";
 "Downloaded  63676682503912L it took: 33ms 4 - [4; 9; 14; 19; 24; 29; 34; 39; 44; 49]"]
user_45508980
Related