Capturing common control patterns such as cancellation and progress reporting is an absolutely essential part of mastering concurrent programming. However, one of the problems with .NET classes such as BackgroundWorker is that they're often more imperative than you may want, and they force other common patterns to be captured by using mutable data structures shared between threads. This leads to the more difficult topic of shared-memory concurrency, discussed later in the chapter. Furthermore, the way BackgroundWorker handles cancellation means that you must insert flag-checks and early-exit paths in the executing background process. Finally, BackgroundWorker isn't useful for background threads that perform asynchronous operations, because the background thread exits too early, before the callbacks for the asynchronous operations have executed.

For this reason, it can often be useful to build abstractions that are similar to BackgroundWorker but that capture richer or different control patterns, preferably in a way that doesn't rely on the use of mutable state and that interferes less with the structure of the overall computation. Much of the rest of this chapter looks at various techniques to build these control structures.

You start with a case study in which you build a type IterativeBackgroundWorker that represents a variation on the BackgroundWorker design pattern. Listing 13-2 shows the code.

open System.ComponentModel
open System.Windows.Forms

/// An IterativeBackgroundWorker follows the BackgroundWorker design pattern
/// but instead of running an arbitrary computation it iterates a function
/// a fixed number of times and reports intermediate and final results.
/// The worker is paramaterized by its internal state type.
///
/// Percentage progress is based on the iteration number. Cancellation checks
/// are made at each iteration. Implemented via an internal BackgroundWorker.
type IterativeBackgroundWorker<'T>(oneStep:('T -> 'T),
                                   initialState:'T,
                                   numIterations:int) =

    let worker =
        new BackgroundWorker(WorkerReportsProgress=true,
                             WorkerSupportsCancellation=true)


    // Create the events that we will later trigger
    let completed = new Event<_>()
    let error     = new Event<_>()
    let cancelled = new Event<_>()
    let progress  = new Event<_>()

    do worker.DoWork.Add(fun args ->

// This recursive function represents the computation loop.
        // It runs at "maximum speed", i.e. is an active rather than
        // a reactive process, and can only be controlled by a
        // cancellation signal.
        let rec iterate state i =
            // At the end of the computation terminate the recursive loop
            if worker.CancellationPending then
               args.Cancel <- true
            elif i < numIterations then
                // Compute the next result
                let state' = oneStep state

                // Report the percentage computation and the internal state
                let percent = int ((float (i+1)/float numIterations) * 100.0)
                do worker.ReportProgress(percent, box state);

                // Compute the next result
                iterate state' (i+1)
            else
                args.Result <- box state

        iterate initialState 0)

    do worker.RunWorkerCompleted.Add(fun args ->
        if args.Cancelled       then cancelled.Trigger()
        elif args.Error <> null then error.Trigger args.Error
        else completed.Trigger (args.Result :?> 'T))

    do worker.ProgressChanged.Add(fun args ->
        progress.Trigger (args.ProgressPercentage,(args.UserState :?> 'T)))

    member x.WorkerCompleted  = completed.Publish
    member x.WorkerCancelled  = cancelled.Publish
    member x.WorkerError      = error.Publish
    member x.ProgressChanged  = progress.Publish

    // Delegate the remaining members to the underlying worker
    member x.RunWorkerAsync()    = worker.RunWorkerAsync()
    member x.CancelAsync()       = worker.CancelAsync()

The types inferred for the code in Listing 13-2 are as follows:

type IterativeBackgroundWorker<'State> =
    new : ('State -> 'State) * 'State * int -> IterativeBackgroundWorker<'State>
    member RunWorkerAsync : unit -> unit
    member CancelAsync : unit -> unit
    member ProgressChanged : IEvent<int * 'State>
    member WorkerCompleted : IEvent<'State>
    member WorkerCancelled : IEvent<unit>
    member WorkerError     : IEvent<exn>

Let's look at this signature first, because it represents the design of the type. The worker constructor is given a function of type 'State -> 'State to compute successive iterations of the computation, plus an initial state and the number of iterations to compute. For example, you can compute the Fibonacci numbers using the following iteration function:

let fibOneStep (fibPrevPrev:bigint,fibPrev) = (fibPrev, fibPrevPrev+fibPrev);;

The type of this function is as follows:

val fibOneStep : bigint * bigint -> bigint * bigint

The RunWorkerAsync and CancelAsync members follow the BackgroundWorker design pattern, as do the events, except that you expand the RunWorkerCompleted event into three events to correspond to the three termination conditions and modify the ProgressChanged to include the state. You can instantiate the type as follows:

> let worker = new IterativeBackgroundWorker<_>( fibOneStep,(1I,1I),100);;
val worker : IterativeBackgroundWorker<bigint * bigint>

> worker.WorkerCompleted.Add(fun result ->
      MessageBox.Show(sprintf "Result = %A" result) |> ignore);;
val it : unit = ()

> worker.ProgressChanged.Add(fun (percentage, state) ->
    printfn "%d%% complete, state = %A" percentage state);;
val it : unit = ()

> worker.RunWorkerAsync();;
1% complete, state = (1I, 1I)
2% complete, state = (1I, 2I)
3% complete, state = (2I, 3I)
4% complete, state = (3I, 5I)
...
98% complete, state = (135301852344706746049I, 218922995834555169026I)
99% complete, state = (218922995834555169026I, 354224848179261915075I)
100% complete, state = (354224848179261915075I, 573147844013817084101I)
val it : unit = ()

One difference here is that cancellation and percentage progress reporting are handled automatically based on the iterations of the computation. This is assuming each iteration takes roughly the same amount of time. Other variations on the BackgroundWorker design pattern are possible. For example, reporting percentage completion of fixed tasks such as installation is often performed by timing sample executions of the tasks and adjusting the percentage reports appropriately.

Often, you need to raise additional events from objects that follow the BackgroundWorker design pattern. For example, let's say you want to augment IterativeBackgroundWorker to raise an event when the worker starts its work and for this event to pass the exact time that the worker thread started as an event argument. Listing 13-3 shows the extra code you need to add to the IterativeBackgroundWorker type to make this happen. You use this extra code in the next section.

open System
open System.Threading

// Pseudo-code for adding event-raising to this object
type IterativeBackgroundWorker<'T>(...) =

    let worker = ...

    // The constructor captures the synchronization context. This allows us to post
    // messages back to the GUI thread where the BackgroundWorker was created.
    let syncContext = SynchronizationContext.Current
    do if syncContext = null then failwith "no synchronization context found"

    let started = new Event<_>()

    // Raise the event when the worker starts. This is done by posting a message
    // to the captured synchronization context.
    do worker.DoWork.Add(fun args ->
        syncContext.Post(SendOrPostCallback(fun _ -> started.Trigger(DateTime.Now)),
                         state=null)
        ...

    /// The Started event gets raised when the worker starts. It is
    /// raised on the GUI thread (i.e. in the synchronization context of
    /// the thread where the worker object was created).
    // It has type IEvent<DateTime>
    member x.Started             = started.Publish

The simple way to raise additional events is often wrong. For example, it's tempting to create an event, arrange for it to be triggered, and publish it, as you would do for a GUI control. However, if you do that, you end up triggering the event on the background worker thread, and its event handlers run on that thread. This is dangerous, because most GUI objects (and many other objects) can be accessed only from the thread they were created on; this is a restriction enforced by most GUI systems.

To round off this section on the BackgroundWorker design pattern, Listing 13-4 shows the full code required to build a small application with a background worker task that supports cancellation and reports progress.

open System.Drawing
open System.Windows.Forms

let form = new Form(Visible=true,TopMost=true)

let panel = new FlowLayoutPanel(Visible=true,
                                Height = 20,
                                Dock=DockStyle.Bottom,
                                BorderStyle=BorderStyle.FixedSingle)

let progress = new ProgressBar(Visible=false,
                               Anchor=(AnchorStyles.Bottom ||| AnchorStyles.Top),
                               Value=0)

let text = new Label(Text="Paused",
                     Anchor=AnchorStyles.Left,
                     Height=20,
                     TextAlign= ContentAlignment.MiddleLeft)

panel.Controls.Add(progress)
panel.Controls.Add(text)
form.Controls.Add(panel)

let fibOneStep (fibPrevPrev:bigint,fibPrev) = (fibPrev, fibPrevPrev+fibPrev)

// Run the iterative algorithm 500 times before reporting intermediate results
let rec repeatMultipleTimes n f s =

if n <= 0 then s else repeatMultipleTimes (n-1) f (f s)

// Burn some additional cycles to make sure it runs slowly enough
let rec burnSomeCycles n f s =
    if n <= 0 then f s else ignore (f s); burnSomeCycles (n-1) f s

let step = (repeatMultipleTimes 500 (burnSomeCycles 1000 fibOneStep))

// Create the iterative worker.
let worker = new IterativeBackgroundWorker<_>(step,(1I,1I),80)

worker.ProgressChanged.Add(fun (progressPercentage,state)->
    progress.Value <- progressPercentage)

worker.WorkerCompleted.Add(fun (_,result) ->
    progress.Visible <- false;
    text.Text <- "Paused";
    MessageBox.Show(sprintf "Result = %A" result) |> ignore)

worker.WorkerCancelled.Add(fun () ->
    progress.Visible <- false;
    text.Text <- "Paused";
    MessageBox.Show(sprintf "Cancelled OK!") |> ignore)

worker.WorkerError.Add(fun exn ->
    text.Text <- "Paused";
    MessageBox.Show(sprintf "Error: %A" exn) |> ignore)

form.Menu <- new MainMenu()
let workerMenu = form.Menu.MenuItems.Add("&Worker")

workerMenu.MenuItems.Add(new MenuItem("Run",onClick=(fun _ args ->
    text.Text <- "Running";
    progress.Visible <- true;
    worker.RunWorkerAsync())))


workerMenu.MenuItems.Add(new MenuItem("Cancel",onClick=(fun _ args ->
    text.Text <- "Cancelling";
    worker.CancelAsync())))

form.Closed.Add(fun _ -> worker.CancelAsync())

When you run the code in F# Interactive, a window appears, as shown in Figure 13-1.

Figure 13.1. A GUI window with a BackgroundWorker reporting progress percentage

One of the most intuitive asynchronous tasks is fetching a web page; we all use web browsers that can fetch multiple pages simultaneously. The samples in Chapter 2 showed how to fetch pages synchronously. This is useful for many purposes, but browsers and high-performance web crawlers have tens or thousands of connections in flight at once.

The type Microsoft.FSharp.Control.Async<'T> lies at the heart of F# asynchronous workflows. A value of type Async<'T> represents a program fragment that will generate a value of type 'T at some point in the future. Listing 13-5 shows how to use asynchronous workflows to fetch several web pages simultaneously. (This example uses a method AsyncReadToEnd defined in the open source library called the F# Power Pack. If you like, you can replace this with a call to the synchronous method ReadToEnd, defined in the standard .NET libraries. Doing so may increase the thread count used by executing this code.)

#r "FSharp.PowerPack.dll" // contains the definition for AsyncReadToEnd
open System.Net
open System.IO

let museums = ["MOMA",           "http://moma.org/";
               "British Museum", "http://www.thebritishmuseum.ac.uk/";
               "Prado",          "http://museoprado.mcu.es"]

let fetchAsync(nm,url:string) =
    async { printfn "Creating request for %s..." nm
            let req  = WebRequest.Create(url)

            let! resp  = req.AsyncGetResponse()

            printfn "Getting response stream for %s..." nm
            let stream = resp.GetResponseStream()

            printfn "Reading response for %s..." nm
            let reader = new StreamReader(stream)
            let! html = reader.AsyncReadToEnd()

            printfn "Read %d characters for %s..." html.Length nm }

for nm,url in museums do
    Async.Start (fetchAsync(nm,url))

The types of these functions and values are as follows:

val museums : (string * string) list
val fetchAsync : string * string -> Async<unit>

When run on one of our machines via F# Interactive, the output of the code from Listing 13-5 is as follows:

Creating request for MOMA...
Creating request for British Museum...
Creating request for Prado...
Getting response for MOMA...
Reading response for MOMA...
Getting response for Prado...
Reading response for Prado...
Read 188 characters for Prado...
Read 41635 characters for MOMA...
Getting response for British Museum...
Reading response for British Museum...
Read 24341 characters for British Museum...

The heart of the code in Listing 13-5 is the construct introduced by async { ... }. This is an application of the workflow syntax introduced in Chapter 9.

Let's take a closer look at Listing 13-5. The key operations are the two let! operations within the workflow expression:

async { ...
        let! resp  = req.AsyncGetResponse()
        ...
        let! html = reader.AsyncReadToEnd()
        ... }

Within asynchronous workflow expressions, the language construct let! var = expr in body means "perform the asynchronous operation expr and bind the result to var when the operation completes. Then, continue by executing the rest of the computation body."

With this in mind, you can now see what fetchAsync does:

Finally, you use the method Async.Start to initiate the execution of a number of asynchronous computations. This works by queuing the computations in the .NET thread pool. The following section explains the .NET thread pool in more detail.

Asynchronous computations are different from normal, synchronous computations: an asynchronous computation tends to hop between different underlying .NET threads. To see this, let's augment the asynchronous computation with diagnostics that show the ID of the underlying .NET thread at each point of active execution. You can do this by replacing uses of printfn in the function fetchAsync with uses of the following function:

let tprintfn fmt =
    printf "[.NET Thread %d]" System.Threading.Thread.CurrentThread.ManagedThreadId;
    printfn fmt

After doing this, the output changes to the following:

[.NET Thread 12]Creating request for MOMA...
[.NET Thread 13]Creating request for British Museum...
[.NET Thread 12]Creating request for Prado...
[.NET Thread 8]Getting response for MOMA...
[.NET Thread 8]Reading response for MOMA...
[.NET Thread 9]Getting response for Prado...
[.NET Thread 9]Reading response for Prado...
[.NET Thread 9]Read 188 characters for Prado...

[.NET Thread 8]Read 41635 characters for MOMA...
[.NET Thread 8]Getting response for British Museum...
[.NET Thread 8]Reading response for British Museum...
[.NET Thread 8]Read 24341 characters for British Museum...

Note how each individual Async program hops between threads; the MOMA request started on .NET thread 12 and finished life on .NET thread 8. Each asynchronous computation in Listing 13-5 executes in the following way:

open System.Threading

ThreadPool.QueueUserWorkItem(fun _ -> printf "Hello!") |> ignore

Async<'T> values are essentially a way of writing continuation-passing or callback programs explicitly. Continuations themselves were described in Chapter 8 along with techniques to pass them explicitly. Async<'T> computations call a success continuation when the asynchronous computation completes and anexception continuation if it fails. They provide a form of managed asynchronous computation, where managed means that several aspects of asynchronous programming are handled automatically:

If you put aside the question of cancellation, values of type Async<'T> are effectively identical to the following type:

type Async<'T> = Async of ('T -> unit) * (exn -> unit) -> unit

Here, the functions are the success continuation and exception continuations, respectively. Each value of type Async<'T> should eventually call one of these two continuations. The async object is of type AsyncBuilder and supports the following methods, among others:

type AsyncBuilder with
    member Return : 'T -> Async<'T>
    member Delay : (unit -> Async<'T>) -> Async<'T>
    member Using: 'T * ('T -> Async<'U>) -> Async<'U> when 'T :> System.IDisposable
    member Bind: Async<'T> * ('T -> Async<'U>) -> Async<'U>

The full definition of Async<'T> values and the implementations of these methods for the async object are given in the F# library source code. As you saw in Chapter 9, builder objects such as async containing methods like those shown previously mean that you can use the syntax async { ... } as a way of building Async<'T> values.

Table 13-2 shows the common constructs used in asynchronous workflow expressions. For example, the following asynchronous workflow

async { let req  = WebRequest.Create("http://moma.org/")
        let! resp  = req.AsyncGetResponse()
        let stream = resp.GetResponseStream()
        let reader = new StreamReader(stream)
        let! html = reader.AsyncReadToEnd()
        html }

is shorthand for the following code:

async.Delay(fun () ->
    let req = WebRequest.Create("http://moma.org/")
    async.Bind(req.AsyncGetResponse(), (fun resp ->
        let stream = resp.GetResponseStream()
        let reader = new StreamReader(stream)
        async.Bind(reader.AsyncReadToEnd(), (fun html ->
            async.Return html)))

As you saw in Chapter 9, the key to understanding the F# workflow syntax is always to understand the meaning of let!. In the case of async workflows, let! executes one asynchronous computation and schedules the next computation for execution after the first asynchronous computation completes. This is syntactic sugar for the Bind operation on the async object.

This section shows a slightly longer example of asynchronous I/O processing. The running sample is an application that must read a large number of image files and perform some processing on them. This kind of application may be compute bound (if the processing takes a long time and the file system is fast) or I/O bound (if the processing is quick and the file system is slow). Using asynchronous techniques tends to give good overall performance gains when an application is I/O bound and can also give performance improvements for compute-bound applications if asynchronous operations are executed in parallel on multicore machines.

Listing 13-6 shows a synchronous implementation of the image-transformation program.

open System.IO
let numImages = 200
let size = 512
let numPixels = size * size

let makeImageFiles () =
    printfn "making %d %dx%d images... " numImages size size

let pixels = Array.init numPixels (fun i -> byte i)
    for i = 1 to numImages  do
        System.IO.File.WriteAllBytes(sprintf "Image%d.tmp" i, pixels)
    printfn "done."

let processImageRepeats = 20

let transformImage (pixels, imageNum) =
    printfn "transformImage %d" imageNum;
    // Perform a CPU-intensive operation on the image.
    for i in 1 .. processImageRepeats do
        pixels |> Array.map (fun b -> b + 1uy) |> ignore
    pixels |> Array.map (fun b -> b + 1uy)

let processImageSync i =
    use inStream =  File.OpenRead(sprintf "Image%d.tmp" i)
    let pixels = Array.zeroCreate numPixels
    let nPixels = inStream.Read(pixels,0,numPixels);
    let pixels' = transformImage(pixels,i)
    use outStream =  File.OpenWrite(sprintf "Image%d.done" i)
    outStream.Write(pixels',0,numPixels)

let processImagesSync () =
    printfn "processImagesSync...";
    for i in 1 .. numImages do
        processImageSync(i)

> System.Environment.CurrentDirectory <- __SOURCE_DIRECTORY__;;
val it : unit = ()

> makeImageFiles();;
val it : unit = ()

You leave the transformation on the image largely unspecified, such as the function transformImage. By changing the value of processImageRepeats, you can adjust the computation from compute bound to I/O bound.

The problem with this implementation is that each image is read and processed sequentially, when in practice multiple images can be read and transformed simultaneously, giving much greater throughput. Listing 13-7 shows the implementation of the image processor using an asynchronous workflow.

let processImageAsync i =
    async { use inStream = File.OpenRead(sprintf "Image%d.tmp" i)
            let! pixels = inStream.AsyncRead(numPixels)
            let  pixels' = transformImage(pixels,i)
            use outStream = File.OpenWrite(sprintf "Image%d.done" i)
            do! outStream.AsyncWrite(pixels')  }

let processImagesAsync() =
    printfn "processImagesAsync...";
    let tasks = [ for i in 1 .. numImages -> processImageAsync(i) ]
    Async.RunSynchronously (Async.Parallel tasks)  |> ignore
    printfn "processImagesAsync finished!";

On one of our machines, the asynchronous version of the code ran up to three times as fast as the synchronous version (in total elapsed time), when processImageRepeats is 20 and numImages is 200. A factor of 2 was achieved consistently for any number of processImageRepeats because this machine had two CPUs.

Let's take a closer look at this code. The call Async.Run (Async.Parallel ...) executes a set of asynchronous operations in the thread pool, collects their results (or their exceptions), and returns the overall array of results to the original code. The core asynchronous workflow is introduced by the async { ... } construct. Let's look at the inner workflow line by line:

async { use inStream = File.OpenRead(sprintf "Image%d.tmp" i)
        ... }

This line opens the input stream synchronously using File.OpenRead. Although this is a synchronous operation, the use of use indicates that the lifetime of the stream is managed over the remainder of the workflow. The stream is closed when the variable is no longer in scope: that is, at the end of the workflow, even if asynchronous activations occur in between. If any step in the workflow raises an uncaught exception, then the stream is also closed while handling the exception.

The next line reads the input stream asynchronously using inStream.AsyncRead:

async { use inStream =  File.OpenRead(sprintf "Image%d.tmp" i)
        let! pixels = inStream.AsyncRead(numPixels)
        ... }

Stream.AsyncRead is an extension method added to the .NET System.IO.Stream class defined in the F# library and generates a value of type Async<byte[]>. The use of let! executes this operation asynchronously and registers a callback. When the callback is invoked, the value pixels is bound to the result of the operation, and the remainder of the asynchronous workflow is executed. The next line transforms the image synchronously using transformImage:

async { use inStream =  File.OpenRead(sprintf "Image%d.tmp" i)
        let! pixels = inStream.AsyncRead(numPixels)
        let  pixels' = transformImage(pixels,i)
        ... }

Like the first line, the next line opens the output stream. Using use guarantees that the stream is closed by the end of the workflow regardless of whether exceptions are thrown in the remainder of the workflow:

async { use inStream =  File.OpenRead(sprintf "Image%d.tmp" i)
        let! pixels = inStream.AsyncRead(numPixels)
        let  pixels' = transformImage(pixels,i)
        use outStream =  File.OpenWrite(sprintf "Image%d.done" i)
        ... }

The final line of the workflow performs an asynchronous write of the image. Once again, AsyncWrite is an extension method added to the .NET System.IO.Stream class defined in the F# library:

async { use inStream =  File.OpenRead(sprintf "Image%d.tmp" i)
        let! pixels = inStream.AsyncRead(numPixels)
        let  pixels' = transformImage(pixels,i)
        use outStream =  File.OpenWrite(sprintf "Image%d.done" i)
        do! outStream.AsyncWrite(pixels')  }

If you now return to the first part of the function, you can see that the overall operation of the function is to create numImages individual asynchronous operations, using a sequence expression that generates a list:

let tasks = [ for i in 1 .. numImages -> processImageAsync(i) ]

You can compose these tasks in parallel using Async.Parallel and then run the resulting process using Async.Run. This waits for the overall operation to complete and returns the result:

Async.Run (Async.Parallel tasks)

Table 13-3 shows some of the primitives and combinators commonly used to build asynchronous workflows. Take the time to compare Listings 13-7 and 13-6. Notice the following:

Values of type Async<'T> are usually run using the functions listed in Table 13-4. You can build Async<'T> values using functions and members in the F# libraries.

Asynchronous programming is becoming more widespread because of the use of multicore machines and networks in applications, and many .NET APIs now come with both synchronous and asynchronous versions of their functionality. For example, all web service APIs generated by .NET tools have asynchronous versions of their requests. A quick scan of the .NET API documentation on the Microsoft website reveals the asynchronous operations listed in Table 13-5. These all have equivalent Async<'T> operations defined in the F# libraries as extensions to the corresponding .NET types.

Sometimes, you may need to write a few primitives to map .NET asynchronous operations into the F# asynchronous framework. You see some examples later in this section and in Chapter 14.

Async.Parallel can appear magical. Computation tasks are created, executed, and resynchronized almost without effort. Listing 13-9 shows that a basic implementation of this operator is simple and again helps you see how Async<'T> values work under the hood.

let forkJoinParallel(taskSeq) =
    Async.FromContinuations (fun (cont,econt,ccont) ->
        let tasks = Seq.toArray taskSeq
        let count = ref tasks.Length
        let results = Array.zeroCreate tasks.Length
        tasks |> Array.iteri (fun i p ->
            Async.Start
               (async { let! res = p
                        results.[i] <- res;
                        let n = System.Threading.Interlocked.Decrement(count)
                        if n=0 then cont results })))

This basic implementation first converts the input task sequence to an array and then creates mutable state count and results to record the progress of the parallel computations. It then iterates through the tasks and queues each for execution in the .NET thread pool. Upon completion, each writes its result and decrements the counter using an atomic Interlocked.Decrement operator, discussed further in the section "Using Shared-Memory Concurrency" at the end of this chapter. The last process to finish calls the continuation with the collected results.

In practice, Async.Parallel is implemented more efficiently and takes into account exceptions and cancellation; again, see the F# library code for full details.

One of the great advantages of F# async programming is that it can be used for both CPU and I/O parallel programming tasks. For example, you can use it for many CPU parallelism tasks that don't perform any I/O but rather carry out straight CPU-bound computations.

For optimized, partitioned CPU parallelism, this is often done by using Async.Parallel with a number of tasks that exactly matches the number of physical processors on a machine. For example, the following code shows parallel initialization of an array where each cell is filled by running the input function. The implementation of this function makes careful use of shared memory primitives (a topic discussed later in this book) and is highly efficient:

open System.Threading
open System

// Initialize an array by a parallel init using all available processors
// Note, this primitive doesn't support cancellation.
let parallelArrayInit n f =
   let currentLine = ref −1
   let res = Array.zeroCreate n
   let rec loop () =
       let y = Interlocked.Increment(&currentLine.contents)
       if y < n then res.[y] <- f y; loop()

   // Start just the right number of tasks, one for each physical CPU
   Async.Parallel [ for i in 1 .. Environment.ProcessorCount -> async { do loop()} ]
      |> Async.Ignore
      |> Async.RunSynchronously

   res
> let rec fib x = if x < 2 then 1 else fib (x - 1) + fib (x - 2)
> parallelArrayInit 25 (fun x -> fib x);;
val it : int [] =
  [|1; 1; 2; 3; 5; 8; 13; 21; 34; 55; 89; 144; 233; 377; 610; 987; 1597; 2584;
    4181; 6765; 10946; 17711; 28657; 46368; 75025; 121393; 196418; 317811;
    514229; 832040|]

Two recurring topics in asynchronous programming are exceptions and cancellation. Let's first explore some of the behavior of asynchronous programs with regard to exceptions:

> let failingTask = async { do failwith "fail" };;
val failingTask: Async<unit>

> Async.RunSynchronously failingTask;;
Microsoft.FSharp.Core.FailureException: fail
stopped due to error

> let failingTasks = [ async { do failwith "fail A" };
                       async { do failwith "fail B" }; ];;

val failingTasks: Async<unit>

> Async.RunSynchronously (Async.Parallel failingTasks);;
Microsoft.FSharp.Core.FailureException: fail A
stopped due to error

> Async.RunSynchronously (Async.Parallel failingTasks);;
Microsoft.FSharp.Core.FailureException: fail B
stopped due to error

From these examples, you can see the following:

You can wrap a task using the Async.Catch combinator. This has the following type:

static member Catch : Async<'T> -> Async<Choice<'T,exn>>

For example:

> Async.RunSynchronously (Async.Catch failingTask);;
val it : Choice<unit,exn> = Choice2_2 (FailureException ())

You can also handle errors by using try/finally in an async { ... } workflow.

In a sense, you've already seen a good deal of message passing in this chapter. For example:

This section covers a simple kind of message processing called mailbox processing that's popular in languages such as Erlang. A mailbox is a message queue that you can scan for a message particularly relevant to the message-processing agent you're defining. Listing 13-10 shows a concurrent agent that implements a simple counter by processing a mailbox as messages arrive. The type MailboxProcessor is defined in the F# library.

let counter =
    new MailboxProcessor<_>(fun inbox ->
        let rec loop n =
            async { printfn "n = %d, waiting..." n
                    let! msg = inbox.Receive()
                    return! loop (n+msg) }
        loop 0)

The type of counter is MailboxProcessor<int>, where the type argument indicates that this object expects to be sent messages of type int:

val counter : MailboxProcessor<int>

The "Message Processing and State Machines" sidebar describes the general pattern of Listing 13-10 and the other MailboxProcessor examples in this chapter, all of which can be thought of as state machines. With this in mind, let's take a closer look at Listing 13-10. First, let's use counter on some simple inputs:

> counter.Start();;
n = 0, waiting...

> counter.Post(1);;
n = 1, waiting...

> counter.Post(2);;
n = 3, waiting...

> counter.Post(1);;
n = 4, waiting...

Looking at Listing 13-10, note that calling the Start method causes the processing agent to enter loop with n = 0. The agent then performs an asynchronous Receive request on the inbox for the MailboxProcessor; that is, the agent waits asynchronously until a message has been received. When the message msg is received, the program calls loop (n+msg). As additional messages are received, the internal counter (actually an argument) is incremented further.

You post messages to the agent using counter.Post. The type of inbox.Receive is as follows:

member Receive: unit -> Async<'Message>

Using an asynchronous receive ensures that no real threads are blocked for the duration of the wait. This means the previous techniques scale to many thousands of concurrent agents.

let agent =
    MailboxProcessor.Start(fun inbox ->

        // The states of the state machine
        let rec state1(args) =  async { ... }
        and     state2(args) =  async { ... }
        ...
        and     stateN(args) =  async { ... }

        // Enter the initial state
        state1(initialArgs))

Often, it's wise to hide the internals of an asynchronous computation behind an object, because the use of message passing can be seen as an implementation detail. Listing 13-10 doesn't show you how to retrieve information from the counter, except by printing it to the standard output. Furthermore, it doesn't show how to ask the processing agent to exit. Listing 13-11 shows how to implement an object wrapping an agent that supports Increment, Stop, and Fetch messages.

/// The internal type of messages for the agent
type internal msg = Increment of int | Fetch of AsyncReplyChannel<int> | Stop

type CountingAgent() =
    let counter = MailboxProcessor.Start(fun inbox ->
             // The states of the message-processing state machine...
             let rec loop n =
                async { let! msg = inbox.Receive()
                        match msg with
                        | Increment m ->
                            // increment and continue...
                            return! loop(n+m)
                        | Stop ->
                            // exit
                            return ()
                        | Fetch  replyChannel  ->
                            // post response to reply channel and continue
                            do replyChannel.Reply n
                            return! loop n }

             // The initial state of the message-processing state machine...
             loop(0))

    member a.Increment(n) = counter.Post(Increment n)
    member a.Stop() = counter.Post Stop
    member a.Fetch() = counter.PostAndReply(fun replyChannel -> Fetch replyChannel)

The inferred public types indicate how the presence of a concurrent agent is successfully hidden by the use of an object:

type CountingAgent =
     new : unit -> CountingAgent
     member Fetch : unit -> int
     member Increment : n:int -> unit
     member Stop : unit -> unit

Here, you can see an instance of this object in action:

> let counter = new CountingAgent();;
val counter : CountingAgent

> counter.Increment(1);;
val it : unit = ()

> counter.Fetch();;
val it : int = 1

> counter.Increment(2);;
val it : unit = ()

> counter.Fetch();;
val it : int = 3

> counter.Stop();;
val it : unit = ()

Listing 13-11 shows several important aspects of message passing and processing using the mailbox-processing model:

Table 13-6 summarizes the most important members available on the MailboxProcessor type.

It's common for a message-processing agent to end up in a state where it's not interested in all messages that may appear in a mailbox but only a subset of them. For example, you may be awaiting a reply from another agent and aren't interested in serving new requests. In this case, it's essential that you use MailboxProcessor.Scan rather than MailboxProcessor.Receive. Table 13-6 shows the signatures of both of these. The former lets you choose between available messages by processing them in order, whereas the latter forces you to process every message. Listing 13-12 shows an example of using MailboxProcessor.Scan.

type Message =
    | Message1
    | Message2 of int
    | Message3 of string

let agent =
    MailboxProcessor.Start(fun inbox ->
        let rec loop() =
            inbox.Scan(function
                | Message1 ->
                   Some (async { do printfn "message 1!"
                                 return! loop() })
                | Message2 n ->
                   Some (async { do printfn "message 2!"
                                 return! loop() })
                | Message3 _ ->
                   None)
        loop())

You can now post these agent messages, including messages of the ignored kind Message3:

> agent.Post(Message1) ;;
message 1!
val it : unit = ()

> agent.Post(Message2(100));;
message 2!
val it : unit = ()

> agent.Post(Message3("abc"));;
val it : unit = ()

> agent.Post(Message2(100));;
message 2!
val it : unit = ()

> agent.CurrentQueueLength;;
val it : int = 1

When you send Message3 to the message processor, the message is ignored. However, the last line shows that the unprocessed Message3 is still in the message queue, which you examine using the backdoor property UnsafeMessageQueueContents.

At the start of this chapter, we mentioned that the rise of the Web and other forms of networks is a major reason for the increasing importance of concurrent and asynchronous programming. Listing 13-13 shows an implementation of a web crawler using asynchronous programming and mailbox-processing techniques.

open System.Collections.Generic
open System.Net
open System.IO
open System.Threading
open System.Text.RegularExpressions

let limit = 50
let linkPat = "href=s*"[^"h]*(http://[^&"]*)""
let getLinks (txt:string) =
    [ for m in Regex.Matches(txt,linkPat)  -> m.Groups.Item(1).Value ]

// A type that helps limit the number of active web requests
type RequestGate(n:int) =
    let semaphore = new Semaphore(initialCount=n,maximumCount=n)
    member x.AsyncAcquire(?timeout) =
        async { let! ok = Async.AwaitWaitHandle(semaphore,
                                                ?millisecondsTimeout=timeout)
                if ok then
                   return
                     { new System.IDisposable with
                         member x.Dispose() =
                             semaphore.Release() |> ignore }
                else
                   return! failwith "couldn't acquire a semaphore" }

// Gate the number of active web requests
let webRequestGate = RequestGate(5)

// Fetch the URL, and post the results to the urlCollector.
let collectLinks (url:string) =
    async { // An Async web request with a global gate
            let! html =
                async { // Acquire an entry in the webRequestGate. Release
                        // it when 'holder' goes out of scope
                        use! holder = webRequestGate.AsyncAcquire()

                        let req = WebRequest.Create(url,Timeout=5)

                        // Wait for the WebResponse
                        use! response = req.AsyncGetResponse()

// Get the response stream
                        use reader = new StreamReader(response.GetResponseStream())

                        // Read the response stream (note: a synchronous read)
                        return reader.ReadToEnd()  }

            // Compute the links, synchronously
            let links = getLinks html

            // Report, synchronously
            do printfn "finished reading %s, got %d links" url (List.length links)

            // We're done
            return links }

/// 'urlCollector' is a single agent that receives URLs as messages. It creates new
/// asynchronous tasks that post messages back to this object.
let urlCollector =
    MailboxProcessor.Start(fun self ->

        // This is the main state of the urlCollector
        let rec waitForUrl (visited : Set<string>) =

           async { // Check the limit
                   if visited.Count < limit then

                       // Wait for a URL...
                       let! url = self.Receive()
                       if not (visited.Contains(url)) then
                           // Start off a new task for the new url. Each collects
                           // links and posts them back to the urlCollector.
                           do! Async.StartChild
                                   (async { let! links = collectLinks url
                                            for link in links do
                                               self.Post link }) |> Async.Ignore

                       // Recurse into the waiting state
                       return! waitForUrl(visited.Add(url)) }

        // This is the initial state.
        waitForUrl(Set.empty))

You can initiate a web crawl from a particular URL as follows:

> urlCollector <-- "http://news.google.com";;
finished reading http://news.google.com, got 191 links
finished reading http://news.google.com/?output=rss, got 0 links
finished reading http://www.ktvu.com/politics/13732578/detail.html, got 14 links
finished reading http://www.washingtonpost.com/wp-dyn/content/art..., got 218 links
finished reading http://www.newsobserver.com/politics/story/646..., got 56 links
finished reading http://www.foxnews.com/story/0,2933,290307,0...l, got 22 links
...

The key techniques shown in Listing 13-13 are as follows:

Listing 13-13 shows that it's relatively easy to create sophisticated, scalable asynchronous programs using a mix of message passing and asynchronous I/O techniques. Modern web crawlers have thousands of outstanding open connections, indicating the importance of using asynchronous techniques in modern scalable web-based programming.

This chapter has avoided showing how to work with threads directly, instead relying on abstractions such as BackgroundWorker and the .NET thread pool. If you want to create threads directly, here is a short sample:

open System.Threading
let t = new Thread(ThreadStart(fun _ ->
                printfn "Thread %d: Hello" Thread.CurrentThread.ManagedThreadId));
t.Start();
printfn "Thread %d: Waiting!" Thread.CurrentThread.ManagedThreadId
t.Join();
printfn "Done!"

When run, this gives the following:

val t : Thread

Thread 1: Waiting!
Thread 10: Hello
Done!

Many multithreaded applications use mutable data structures shared between multiple threads. Without synchronization, these data structures will almost certainly become corrupt: threads may read data that has been only partially updated (because not all mutations are atomic), or two threads may write to the same data simultaneously (a race condition). Mutable data structures are usually protected by locks, although lock-free mutable data structures are also possible.

Shared-memory concurrency is a difficult and complicated topic, and a considerable amount of good material on .NET shared-memory concurrency is available on the Web. All this material applies to F# when you're programming with mutable data structures such as reference cells, arrays, and hash tables and the data structures can be accessed from multiple threads simultaneously. F# mutable data structures map to .NET memory in fairly predictable ways; for example, mutable references become mutable fields in a .NET class, and mutable fields of word size can be assigned atomically.

On modern microprocessors, multiple threads can see views of memory that aren't consistent; that is, not all writes are propagated to all threads immediately. The guarantees given are called a memory model and are usually expressed in terms of the ordering dependencies between instructions that read/write memory locations. This is, of course, deeply troubling, because you have to think about a huge number of possible reorderings of your code, and it's one of the main reasons why shared mutable data structures are difficult to work with. You can find further details on the .NET memory model at www.expert-fsharp.net/topics/MemoryModel.

Locks are the simplest way to enforce mutual exclusion between two threads attempting to read or write the same mutable memory location. Listing 13-14 shows an example of code with a race condition.

type MutablePair<'T,'U>(x:'T,y:'U) =
    let mutable currentX = x
    let mutable currentY = y
    member p.Value = (currentX,currentY)
    member p.Update(x,y) =
        // Race condition: This pair of updates is not atomic
        currentX <- x;
        currentY <- y

let p = new MutablePair<_,_>(1,2)
do Async.Start (async { do (while true do p.Update(10,10)) })
do Async.Start (async { do (while true do p.Update(20,20)) })

Here is the definition of the F# lock function:

open System.Threading
let lock (lockobj : obj) f  =
    Monitor.Enter lockobj;
    try
        f()
    finally
        Monitor.Exit lockobj

The pair of mutations in the Update method isn't atomic; that is, one thread may have written to currentX, another then writes to both currentX and currentY, and the final thread then writes to currentY, leaving the pair holding the value (10,20) or (20,10). Mutable data structures are inherently prone to this kind of problem if shared between multiple threads. Luckily, F# code tends to have fewer mutations than imperative languages, because functions normally take immutable values and return a calculated value. However, when you do use mutable data structures, they shouldn't be shared between threads, or you should design them carefully and document their properties with respect to multithreaded access.

Here is one way to use the F# lock function to ensure that updates to the data structure are atomic. Locks are also required on uses of the property p.Value:

do Async.Start (async { do (while true do lock p (fun () -> p.Update(10,10))) })
do Async.Start (async { do (while true do lock p (fun () -> p.Update(20,20))) })

It's common for mutable data structures to be read more than they're written. Indeed, mutation is often used only to initialize a mutable data structure. In this case, you can use a .NET ReaderWriterLock to protect access to a resource. For example, consider the following two functions:

open System.Threading

let readLock (rwlock : ReaderWriterLock) f  =
  rwlock.AcquireReaderLock(Timeout.Infinite)
  try
      f()
  finally
      rwlock.ReleaseReaderLock()

let writeLock (rwlock : ReaderWriterLock) f  =
  rwlock.AcquireWriterLock(Timeout.Infinite)
  try
      f();
      Thread.MemoryBarrier()
  finally
      rwlock.ReleaseWriterLock()

Listing 13-15 shows how to use these functions to protect the MutablePair class.

type MutablePair<'T,'U>(x:'T,y:'U) =
    let mutable currentX = x
    let mutable currentY = y
    let rwlock = new ReaderWriterLock()
    member p.Value =
        readLock rwlock (fun () ->
            (currentX,currentY))
    member p.Update(x,y) =
        writeLock rwlock (fun () ->
            currentX <- x;
            currentY <- y)

Table 13-7 shows some of the other shared-memory concurrency primitives available in the .NET Framework.