Parallel Programming in F# (IV.): Financial dashboard example

In the fourth part of the Parallel Programming in F# series, we'll take a look at the Adatum Financial Dashboard sample. The sample is a part of Parallel Programming with Microsoft .NET, which is a guide that introduces common parallel programming patterns on .NET 4.0. The C# version of the sample is in details discussed in the guide, but the F# translation contains several additional interesting aspects that I'll introduce in this article.

The sample simulates a financial modelling application that performs processing of market data in several steps that can be partly executed in parallel. In this article we'll compare two implementations. The first one (similar to the C# version) uses the Task<'T> type from .NET 4.0 and chains steps using the ContinueWith method. The second version is F# specific and it implements step as a sequential computation wrapped in asynchronous workflows. Partial results are reported to the user interface using events.

For discussion about other F# examples from Parallel Programming with Microsoft .NET, you can refer to other parts of this article series. So far, the series covers the following topics:

• Parallel Programming in F# (I.): Introducing the samples
• Parallel Programming in F# (II.): Using PLINQ and Tasks
• Parallel Programming in F# (III.): Aggregating data
• Parallel Programming in F# (IV.): Financial dashboard example

Adatum financial dashboard

(Click on the image for a larger version)

The example shows how to develop a financial analysis application that performs several tasks in parallel. The application is created in WPF and uses the MVVM pattern, which separates the declarative user interface (View), a type that exposes data for WPF (ViewModel) and the actual analysis engine (Model).

The application performs analysis of a historical data and yields a market recommendation (to buy or not to buy). It starts by reading data from three data sources, which is an I/O bound task. Next, it merges data from two sources and analyzes them. In parallel, it analyzes data from the third source. The data-flow is displayed on the diagram on the right (adapted from Chapter 5 of Parallel Programming with Microsoft .NET).

The file AnalysisEngine.fs contains both of the parallel implementations of the calculations (using .NET 4.0 Task<'T> type and using F# asynchronous workflows) as well as a single sequential reference solution. Let's start by looking at the Task-based solution.

Financial Dashboard using Tasks

The key part of the solution is the DoAnalysisParallel method. It constructs tasks that call long-running functions that simulate doing some work. For example, loadNyseData is I/O intensive and mergeMarketData, analyzeData and runModel are CPU intensive. In the left branch of the processing (see the diagram), we first start two I/O bound operations:

1: /// Utility function that creates a long running task from F# function
2: let taskLong (f:unit -> 'T) = 
3:     Task<'T>.Factory.StartNew(new Func<'T>(f), TaskCreationOptions.LongRunning)
4: 
5: // Load & process NYSE and NASDAQ data
6: let loadNyseDataTask = taskLong loadNyseData
7: let loadNasdaqDataTask = taskLong loadNasdaqData

The taskLong function is a simple wrapper that starts a specified F# function as a task with the LongRunning flag, which provides a hint to the task scheduler that the task is I/O bound. We use the function to create two tasks. The first one loads data from New York Stock Exchange and the second from the NASDAQ Stock Market. Both of the tasks are values of type Task<StockDataCollection>. This means that the task will start running in background and will eventually produce a collection of stock data.

Chaining of tasks

As a next step, we want to create a task that will wait for both of the sources to load and merge the data into a single collection. When this step is done, we want to continue by running data normalization using normalizeData. This can be written as follows:

 1: // Create task that merges data when both sources are loaded
 2: let mergeMarketDataTask = 
 3:   Task.Factory.ContinueWhenAll
 4:     ( [| loadNyseDataTask; loadNasdaqDataTask |],
 5:       fun (tasks:Task<_>[]) -> 
 6:           mergeMarketData [ for t in tasks -> t.Result ])
 7: 
 8: // Create task that normalizes data when they are merged
 9: let normalizeMarketDataTask =
10:     mergeMarketDataTask.ContinueWith(fun (t:Task<_>) ->
11:         normalizeData(t.Result))

The ContinueWhenAll method creates a task that will be started when all tasks in the specified array complete. When that happens, it runs the specified lambda function. Inside the lambda, we select results of all tasks using a simple list comprehension and merge them.

When the merged task finishes, we want to perform another step - normalize the merged data. Because we're creating a task that waits only for a single other task, we can use the ContinueWith method. The lambda function will be called when the merging is done and the merged stock data can be accessed via t.Result. In the complete example, the normalization of data is followed by several other steps (see the diagram above), but the implementation follows the pattern presented here.

Handling exceptions

One interesting aspect of the example is error handling. When an exception occurs anywhere in the code, we want to run a callback provided by the ViewModel component (to update the GUI). To do that, we need to wait for all the tasks that form the computation (using ContinueWhenAll) and then call Task.WaitAll. This method throws an exception when any of the awaited tasks throws, so we can use it to get the error message and log it. In the example, this functionality is wrapped in createErrorHandler function which is used as follows:

1: // Construct error handler and return tasks as result
2: let errorHandler = 
3:   // Merge all tasks, so that we can handle all exceptions
4:   createErrorHandler 
5:     [ loadNyseDataTask; loadNasdaqDataTask; loadFedHistoricalDataTask; 
6:       mergeMarketDataTask; normalizeHistoricalDataTask; (...) ] 

The combined task named errorHandler will finish when all the tasks of the computation finish. If an error occurred, the status of the task will be TaskStatus.Faulted. We use this fact in the code that registers an error handler. When the task failed, we call the provided function to report the failure to the user interface. It is also worth noting that we use the FromCurrentSynchronizationContext method to create a TaskScheduler that will invoke the body of the task on the GUI thread, so that the provided function can safely access WPF components.

Financial Dashboard using Asyncs

The dashboard application combines I/O and CPU intensive tasks, so F# asynchronous workflows may be actually a better choice than tasks. I/O intensive operation in the task-based solution such as loadNyseData blocks the calling thread, but do not fully utilize the thread. When using asynchronous workflows, we'll use asyncLoadNyseData, which is a non-blocking version of the function.

We'll write the computation as a single asynchronous workflow that will produce the final decision (to buy or not to buy). However, the application also requires some intermediate results, so we'll need to report results of various steps in the computation. To do that, we'll use events...

Notifying user interface

As already mentioned, we'll use events to notify the user interface of intermediate results, so we need to start by declaring several events. The following snippet includes only those that will be used in the larger sample below (implementing the left branch of the above diagram). In addition, we'll need two utility functions. The function triggerGuiEvent uses WPF Dispatcher type to invoke the specified event on the main GUI thread, so that the listener can safely access user interface controls. The function triggerWhenCompleted is used to create asynchronous workflow that triggers the specified event on completion:

1: /// Trigger event on the GUI thread using a captured dispatcher
2: let triggerGuiEvent (e:Event<_>) v = 
3:     guiDispatch.Invoke(new Action(fun () -> e.Trigger(v) ), [| |]) |> ignore
4: 
5: /// Constructs workflow that triggers the specified event 
6: /// on the GUI thread when the wrapped async completes 
7: let triggerWhenCompletes (e:Event<_>) (a:Async<_>) = async {
8:     let! res = a
9:     triggerGuiEvent e res
10:    return res }

1: // Create events that are used to report progress to the GUI
2: let loadNyseDataEvt = new Event<_>()
3: let loadNasdaqDataEvt = new Event<_>()
4: let mergeMarketDataEvt = new Event<_>()
5: let normalizeMarketDataEvt = new Event<_>()
6: let analyzeMarketDataEvt = new Event<_>()
7: let modelMarketDataEvt = new Event<_>()
8: let compareModelsEvt = new Event<_>()
9: let errorHandlerEvt = new Event<_>()

The implementation of triggerGuiEvent uses an instance guiDispatch that is captured by the AnalysisEngine when it is created on the main thread. The second helper function is slightly more interesting - it is implemented as an asynchronous workflow that calls the provided workflow, then triggers the given event and finally returns the original result. The function actually just wraps a specified workflow into another workflow that also triggers the event.

Implementing workflow

Now that we have all the utility functions and events, we can look at the most important part of the code - the workflow that performs the computation and reports results to the user interface. This part of the code corresponds to the code that constructs and chains tasks in the previous implementation.

The workflow first needs to download data from the two data sources in parallel. This is done using the Async.StartChild function, which starts the workflow as a task in background and give us another async that can be used to retrieve the result. Once we get both of the inputs, we can run the rest of the processing as a sequential asynchronous code. In the complete application, this will run in parallel with the other branch.

 1: // Load & process NYSE and NASDAQ data
 2: let marketModel = async {
 3:     // Start loading of data from two sources in background (both I/O
 4:     // operations are perfomed asynchronously without blocking threads)
 5:     let nyseWork = asyncLoadNyseData() |> triggerWhenCompletes loadNyseDataEvt
 6:     let nasdaqWork = asyncLoadNasdaqData() |> triggerWhenCompletes loadNasdaqDataEvt 
 7:     let! nyse = nyseWork |> Async.StartChild
 8:     let! nasdaq = nasdaqWork |> Async.StartChild
 9:     
10:     // Wait for both tasks to complete and continue
11:     let! nyseData = nyse
12:     let! nasdaqData = nasdaq
13:     let merged = mergeMarketData [ nyseData; nasdaqData ] 
14:     triggerGuiEvent mergeMarketDataEvt merged
15:     
16:     // Perform analysis of the merged data
17:     let normalized = normalizeData merged
18:     triggerGuiEvent normalizeMarketDataEvt normalized
19:     let analyzed = analyzeData normalized 
20:     triggerGuiEvent analyzeMarketDataEvt analyzed
21:     let res = runModel analyzed 
22:     triggerGuiEvent modelMarketDataEvt res
23:     return res }

In the first part of the workflow, we first create two workflows that will be executed in parallel to download stock data from NYSE and NASDAQ. We use the asynchronous version of a function for loading data and wrap the resulting workflow into a workflow that triggers an event when the loading completes (using triggerWhenCompletes). Then we start both of the workflows in parallel using Async.StartChild and wait until both of them complete (using let! to get the result asynchronously).

The second part of the workflow is fully sequential. We perform all the remaining steps (merge the data, normalize the result, analyze the data and run the data model). After each step, we report the result to the user interface by triggering one of the events (we use triggerGuiEvent to schedule the triggering via message queue). Note that the second part of the workflow does not contain any yield points (e.g. let!), so it will run as a CPU-intensive computation.

Comparing models and exception handling

In the task-based version, we spent some time discussing code that implements exception handling, because we need to write it explicitly by waiting for all tasks. When using asynchronous workflows in F#, the situation is a lot simpler - we can simply use try ... with construct to handle any exceptions. Even when the exception is thrown by another workflow that we use (or invoke using Async.StartChild), the exception will be automatically propagated.

In the next example, we'll look at the last piece of code that builds the complete workflow for the processing. We'll use the marketModel workflow declared above and a workflow historicalModel that represents the other branch. We'll want to run these two workflows in parallel to get two models for a comparison, compare them and report the result to the user interface. When an exception occurs, we want to report it to the GUI using the same mechanism:

1: let compare = async {
2:     try
3:         // Run both of the models and compare them to get recommendation
4:         let! models = Async.Parallel [ marketModel; historicalModel ]
5:         let res = compareModels models 
6:         triggerGuiEvent compareModelsEvt res 
7:     with e ->
8:         // Report error to the user interface
9:         triggerGuiEvent errorHandlerEvt ()  }

To run the two models in parallel, we use Async.Parallel to create a single workflow that runs both of them. Then we wait for the result using let! As a result, we'll get an array of models called models, which can be directly passed to the compareModels function. As you can see, adding exception handling is very straightforward, because we just wrapped the whole code inside the try ... with block and we trigger the error handler event in the with clause.

Summary

In this article, we looked at two F# implementations of the Adatum financial dashboard sample application from Parallel Programming with Microsoft .NET. The first version was a direct translation of a C# sample and used .NET 4.0 tasks. The second version was built using F# asynchronous workflows. The notable differences between the two versions are following:

• Asynchronous workflows offer better scalability for I/O bound operations. When loading data from stock exchanges, asynchronous workflows use thread pool threads for non-blocking I/O, while the task based solution requires one thread for each of the operations.
• Exception handling is easier in the version based on asynchronous workflows. The exception is automatically propagated and can be handled using the natural try ... with construct.
• Reporting results to the user interface is slightly easier in the Task-based solution. The result of the task can be accessed at any time after the task finished, while asynchronous workflows invoke the notification once (and the result would have to be cached by the caller).

The choice between the two possible implementations depends on many factors. Asynchronous workflows were designed specifically for F#, so they more naturally fit with the language. They offer better performance for I/O bound tasks and provide more convenient exception handling. Moreover, the sequential syntax is quite convenient. On the other hand, tasks are optimized for CPU bound calculations and make it easier to access the result of calculation from other places of the application without explicit caching.

Downloads and References

• Parallel Programming with Microsoft .NET - Book homepage at CodePlex
• F# Code Samples - Parallel Programming with Microsoft .NET