AsyncParallelStats.fsx

// Change this to your profile folder for nuget packages
#I @"C:\Users\Isaac\.nuget\packages\"
#r @"XPlot.GoogleCharts\2.0.0\lib\netstandard2.0\XPlot.GoogleCharts.dll"
#r @"Newtonsoft.Json\12.0.3\lib\netstandard2.0\Newtonsoft.Json.dll"
#r @"Google.DataTable.Net.Wrapper\4.0.0\lib\netstandard2.0\Google.DataTable.Net.Wrapper.dll"

open System
open XPlot.GoogleCharts

/// Executes an asynchronous workflow and provides some simple statistics on the results.
let exec (values:Async<int array>) =
    async {
        let sw = Diagnostics.Stopwatch.StartNew()
        let! values = values
        return
            {| Runtime = sw.Elapsed.TotalMilliseconds |> int
               WorkerTime = Seq.sum values
               Max = Seq.max values |}
    }
    |> Async.RunSynchronously

module Async =
    /// A simplistic Async throttling implementation which batches workflows
    /// into groups, executing each batch in sequence.
    let Batch size workflow = async {
        let! batchResults =
            workflow
            |> Seq.chunkBySize size
            |> Seq.map Async.Parallel
            |> Async.Sequential
        return Array.concat batchResults }
    /// A wrapper around the throttling overload of Parallel to allow easier
    /// pipelining.
    let ParallelThrottle throttle workflows =
            Async.Parallel(workflows, throttle)
        
type Simulator() =
    let sleepTimes =
        let r = Random()
        let createSleepTime() =
            (float (r.Next(1, 5))
            + r.NextDouble())
            * 1000.
            |> int
        [ for _ in 1 .. 10 -> createSleepTime() ]
    /// Generates a collection of 10 Async workflows, each that sleep for between
    /// 1 and 5 seconds. Then, executes them all using a provided "fork join" function
    /// such as Async.Parallel and returns some statistics on them
    member __.Run forkJoiner =
        let events = ResizeArray()
        let normaliseTime =
            let baseline = float DateTime.UtcNow.Second
            fun (date:DateTime) -> date.AddSeconds(-baseline)
        let results =
            sleepTimes
            |> List.mapi(fun n sleepTime -> 
                async {
                    printfn "Starting %d (%dms)" n sleepTime
                    let stopwatch = Diagnostics.Stopwatch.StartNew()
                    do! Async.Sleep sleepTime
                    printfn "Done %d" n
                    stopwatch.Stop()
                    events.Add {| Index = n; Stopped = normaliseTime DateTime.UtcNow; Duration = stopwatch.Elapsed |}
                    return sleepTime
                })
            |> forkJoiner
            |> exec
        {| results with Events = events.ToArray() |}

// Try out the different versions here
let simulator = Simulator()
let workSeq = simulator.Run Async.Sequential
let workParFull = simulator.Run Async.Parallel
let workParX = simulator.Run (Async.ParallelThrottle 3)
let workBatch = simulator.Run (Async.Batch 3)

// Visualise them here
workBatch.Events
|> Seq.sortBy(fun a -> a.Index)
|> Seq.map(fun a ->
    let start = a.Stopped - a.Duration
    sprintf "Worker %d" a.Index, sprintf "%.2fs" a.Duration.TotalSeconds,start, a.Stopped)
|> Seq.toArray
|> Chart.Timeline
|> Chart.WithHeight 1200
|> Chart.WithWidth 600
|> Chart.WithXTitle "Time in Seconds"
|> Chart.WithLabels [ "Start"; "End" ]
|> Chart.Show