Using Task and Async together for Asynchronous operations on seperate threads in F#

Question

I'm being a little adventurous with my code for the amount of experience I have with F# and I am a little worried about cross threading issues.

Background:

I have a number of orders where I need to validate the address. Some of the orders can be validated against google maps geocoding API which allows 50/ second. the rest are Australian PO Boxes which we don't have many of - but I need to validate them against a different API that only allows 1 call per second.

I have switched over most of my code from async{} functions to task{} functions and I am assuming to get something on several threads at the same time it needs to be in an async{} function or block and be piped to Async.Parallel

Question: Is this the right way to do this or will it fall over? I am wondering if I am fundamentally thinking about this the wrong way.

Notes:

I am passing a database context into the async function and updating the database within that function
I will call this from a C# ( WPF ) Application and report the progress

Am I going to have cross threading issues?

let validateOrder
        (
            order: artooProvider.dataContext.``dbo.OrdersEntity``,
            httpClient: HttpClient,
            ctx: artooProvider.dataContext,
            isAuPoBox: bool
        ) =
        async {

            // Validate Address
            let! addressExceptions = ValidateAddress.validateAddress (order, httpClient, ctx, isAuPoBox) |> Async.AwaitTask

            // SaveExceptions
            do! ctx.SubmitUpdatesAsync()

            // return Exception count
            return ""
        }

    let validateGMapOrders(httpClient: HttpClient, ctx: artooProvider.dataContext, orders: artooProvider.dataContext.``dbo.OrdersEntity`` list) =
    async {
            let ordersChunked = orders |> List.chunkBySize 50

            for fiftyOrders in ordersChunked do
                let! tasks = 
                    fiftyOrders 
                    |> List.map (fun (order) -> validateOrder (order, httpClient, ctx, false) )
                    |> Async.Parallel
                do! Async.Sleep(2000)
        }


    let validateOrders (ctx: artooProvider.dataContext, progress: IProgress) =
    task {
        let unvalidatedOrders =
            query {
                for orders in ctx.Dbo.Orders do
                    where (orders.IsValidated.IsNone)
                    select (orders)
            }
            |> Seq.toList

        let auPoBoxOrders =
            unvalidatedOrders
            |> List.filter (fun order -> isAUPoBox(order) = true )

        let gMapOrders =
            unvalidatedOrders
            |> List.filter (fun order -> isAUPoBox(order) = false )


        let googleHttpClient = new HttpClient()
        let auspostHttpclient = Auspost.AuspostApi.getApiClient ()

        // Google maps validations
        do! validateGMapOrders(googleHttpClient,ctx,gMapOrders)

        // PO Box Validations
        for position in 0 .. auPoBoxOrders.Length - 1 do
            let! result = validateOrder (gMapOrders[position], auspostHttpclient, ctx, true)
            do! Task.Delay(1000)

        return true
    }

tranquillity · Answer

When I have had to deal with rate-limited API problems I hide that API behind a MailboxProcessor that maintains an internal time to comply with the rate limit but appears as a normal async API from the outside.

Since you have two API's with different rate limits I'd parameterise the time delay and processing action then create one object for each API.

open System

type Request = string
type Response = string

type RateLimitedProcessor() =
    // Initialise 1s in past so ready to start immediately.
    let mutable lastCall = DateTime.Now - TimeSpan(0, 0, 1)
    let mbox = new MailboxProcessor>((fun mbox ->
        let rec f () = 
            async {
                let! (req, reply) = mbox.Receive()
                let msSinceCall = (DateTime.Now - lastCall).Milliseconds
                // wait 1s between requests
                if msSinceCall < 1000 then
                    do! Async.Sleep (1000 - msSinceCall)

                lastCall <- DateTime.Now
                reply.Reply "Response"

                // Call self recursively to process the next incoming message
                return! f()
            }
        f()
        ))

    do mbox.Start()
    
    member __.Process(req:Request): Async =
        async {
            return! mbox.PostAndAsyncReply(fun reply -> req, reply)
            }

    interface IDisposable with
        member this.Dispose() = (mbox :> IDisposable).Dispose()

Using Task and Async together for Asynchronous operations on seperate threads in F#

1 Answers