F# Parallel Extras (III.): Financial dashboard with cancellation

In this article we'll look at several improvements that can be done to the Financial dashboard example (originally from the Parallel Programming with Microsoft .NET book). When I was translating samples in the book from C# to F#, the sample struck me, because it looks like a perfect fit for F# asynchronous workflows (instead of the Task<T> type used in the C# version). I already talked about an alternative implementation based on asynchronous workflows. However that version was still following some of the programming patterns, from the original C# version, which are not idiomatic in F#. In this article, I'll talk about a few other improvements that we can make to the sample...

In the original version of the sample (in both C# and F#), we explicitly checked whether a cancellation token has been cancelled in every single operation. This was needed in C#, because tasks do not support cancellation automatically. However, F# asynchronous workflows make cancellation easier. They automatically check if the computation should be cancelled at the beginning and the end of every asynchronous call. Our first change will be to use this feature. Also, the original version propagates a null value when the computation is cancelling. In F# we don't need that and we'll only change the type of the overall result to option<T>, so that we can notify the user interface (but we don't need to propagate cancellation explicitly). Finally, the original version contained sequential implementation, but didn't provide any way of actually running it, so we'll do a bit of refactoring to make that sample actually useable.

Representing analysis process

Let's start with the refactoring. The original C# version (as well as the task-based F# implementation) returns a record with Task<T> values, which allows the user interface to wait for the result using ContinueWith method (the code is available on CodePlex, if you want to take a look). In the previous article, we used events instead, but only in the version based on asynchronous workflows. A more convenient option is to encapsulate the reporting of status (implemented using events) in an abstract class that will represent the computation (or analysis process). The class can then be created and returned by various implementations of the financial calculation. Even though abstract classes are very object oriented, it will work very well in this case, because it can hide some implementation and we'll be able to create it using object expressions.

The following snippet implements the class (the listing hides some of the event declarations, because they all follow the same pattern). The class simply exposes several events and allows the user of the class to trigger them. More interestingly, the class also provides method for starting and cancelling the computation. It is worth noting that the original version of the sample returned a computation that was already running. This is not perfect (at least when using events), because a user may want to setup handlers for the events first and start the computation after that, to make sure that no event occurrences are missed.

 1: [<AbstractClass>]
 2: type AnalysisProcess() = 
 3: 
 4:     // Trigger event on the GUI thread using a captured dispatcher
 5:     let guiDispatch = Windows.Threading.Dispatcher.CurrentDispatcher
 6:     let triggerGuiEvent (e:Event<_>) v = 
 7:         guiDispatch.Invoke(new Action(fun () -> 
 8:               e.Trigger(v) ), [| |]) |> ignore
 9: 
10:     // Create events that are used to notify the 
11:     // caller about partial results of the computation
12:     let loadNyseDataEvt = new Event<StockDataCollection>()
13:     let compareModelsEvt = new Event<MarketRecommendation option>()
14:     let errorHandlerEvt = new Event<unit>()
15:     (other events omitted)
16: 
17:     /// Start the analysis (to be used after handlers are registered)
18:     abstract Start : unit -> unit
19:     /// Try to cancel the process (if it has been already started)
20:     abstract TryCancel : unit -> unit
21: 
22:     // Expose events to the user interface layer
23:     member x.LoadNyseData = loadNyseDataEvt.Publish
24:     member x.CompareModels = compareModelsEvt.Publish
25:     member x.ErrorHandler = errorHandlerEvt.Publish 
26:     // Trigger event (called from the computation engine)
27:     member x.TriggerLoadNyseData(v) = triggerGuiEvent loadNyseDataEvt v
28:     member x.TriggerCompareModels(v) = triggerGuiEvent compareModelsEvt v
29:     member x.TriggerErrorHandler(v) = triggerGuiEvent errorHandlerEvt v
30:     (other members omitted)F# Web Snippets

The type declaration is marked with the AbstractClass attribute, which instructs the F# compiler to treat it as an abstract class. The abstract members of the class are Start and TryCancel. Both of the members are quite simple and don't take any arguments or return any result. They simply start the computation or trigger its cancellation.

Aside from abstract members, the type creates and exposes several event values. In the constructor, we capture the current WPF Dispatcher (assuming that the type is first constructed on the main GUI thread) and when triggering an event, we dispatch the call to the original thread. This is done using a simple helper function triggeGuiEvent that was already discussed in the previous article. For every event, we have a private event value (e.g. loadNyseDataEvt) and two public members (e.g. LoadNyseData and TriggerLoadNyseData). The first member exposes the event and is used from the user interface when registering event handlers. The other member is used from the computation engine to report results to the GUI. Note that the CompareModels event carries MarketRecommendation option as a value. This event will be triggered even if a computation is cancelled, in which case the value will be None (other events may not fire when we cancel the computation).

Creating sequential implementation

To demonstrate how a computation would construct and return an implementation of the above abstract class, let's start by looking at the sequential implementation. In the sequential version, we'll ignore cancellation (adding it would be simple - we'd just set a flag and check whether it was set after every step). Even though this is fully sequential, we'll do the actual work in background (to avoid blocking the GUI thread):

 1: // Implements market analysis using sequential computation
 2: let sequentialAnalysis() = 
 3: 
 4:   // Analysis engine that will be returned as the result.
 5:   let rec engine = 
 6:     { new AnalysisProcess() with
 7:         member x.Start() = Task.Factory.StartNew(analysis) |> ignore
 8:         member x.TryCancel() = () }
 9: 
10:   // Function that implements the sequential analysis
11:   and analysis() =
12:       (loading and processing omitted)
13:   
14:       // Run both models and compare them
15:       let modeledMarketData = runModel analyzedStockData
16:       engine.TriggerModelMarketData(modeledMarketData)
17:       let modeledHistoricalData = runModel analyzedHistoricalData
18:       engine.TriggerModelHistoricalData(modeledHistoricalData)
19:       let recommendation = compareModels [ modeledMarketData; modeledHistoricalData ]
20:       engine.TriggerCompareModels(Some recommendation)
21:   engineF# Web Snippets

The snippet creates a recursive reference between the analysis function and the engine value. In this case, F# compiler inserts some runtime checks to verify that the recursion makes sense and it shows a warning that notifies us about this (we can safely ignore it by adding #nowarn "40"). The engine value is created using object expression and it adds implementation of the two abstract members of the AnalysisProcess class. In the Start member, it creates a task that will start running the analysis function. In the other way round, the analysis function calls members of the engine value to report intermediate progress of the computation (for example by calling the TriggercompareModels member at the end).

Supporting cancellation in asynchronous workflows

The asynchronous implementation of the computation will follow the same pattern. We'll create an engine value that implements the abstract class and we'll have a computation (asynchronous workflow instead of an ordinary function) that will be started from the Start method and will report the progress by calling engine.TriggerXyz members. The following snippet shows the overall structure (implementation of the computation is hidden, as we'll discuss it in the next section):

 1: // Implements market analysis using F# asynchronous workflows
 2: let asyncAnalysis() = 
 3:   let cts = new CancellationTokenSource()
 4: 
 5:   // Analysis engine that will be returned as the result.
 6:   let rec engine = 
 7:     { new AnalysisProcess() with
 8:         member x.Start() = 
 9:           // Start workflow with cancellation token
10:           Async.Start(compare, cancellationToken = cts.Token)
11:         member x.TryCancel() = cts.Cancel() }
12: 
13:   // Load & process NYSE and NASDAQ data
14:   and marketModel = (...)
15:   // Load & process FED data
16:   and historicalModel = (...)
17: 
18:   // Run both of the models and compare them to get recommendation
19:   and compare = (...)
20: 
21:   // Return the created engine
22:   engineF# Web Snippets

The computation is constructed in almost the same way as in the the previous article. We first create two workflows (marketModel and historicalModel) that construct two models and are called from compare computation in parallel. However, there is one important difference to enable automatic cancellation, but we'll discuss that in the next section.

To support cancellation, the function first creates an instance of CancellationTokenSource. When starting the workflow, we pass a token from the cancellation source to the Async.Start operation. When the user requests cancellation we call the Cancel method of the token source. As already discussed, we don't need to explicitly check whether the token was set anywhere in the workflow, because this is done automatically. However, we need to write the asynchronous computation in a way that makes this possible.

Creating computations with cancellation support

If we pass a cancellation token when starting asynchronous workflow, it will automatically check whether the cancellation was requested around every yield point. A yield point is when the control is transferred from the user code to the workflow to perform an asynchronous operation, which is done either using let!, do! or return! Asynchronous workflow does not attempt to cancel the computation actively (e.g. by calling Thread.Abort), because this is considered as dangerous. Instead it uses cooperative cancellation, which is deterministic. In practice, this means that if you have a workflow that should be cancelled, it should make all calls to non-trivial operations asynchronously (and so all the operations need to be written as asynchronous too). The following example demonstrates this practice (some parts are hidden, to make the code shorter, but you can see the idea):

 1: // Load & process NYSE and NASDAQ data
 2: and marketModel = async {
 3:     // Start loading data asynchronously in background
 4:     let! nyse = 
 5:       asyncLoadNyseData() 
 6:         |> Async.WithResult engine.TriggerLoadNyseData 
 7:         |> Async.StartChild
 8:     let! nasdaq = 
 9:         asyncLoadNasdaqData() 
10:         |> Async.WithResult engine.TriggerLoadNasdaqData 
11:         |> Async.StartChild
12: 
13:     (waiting for completion and normalization omitted)
14:     let! analyzed = asyncAnalyzeData normalized 
15:     engine.TriggerAnalyzeMarketData(analyzed)
16:     let! res = asyncRunModel analyzed 
17:     engine.TriggerModelMarketData(res)
18:     return res }
19: 
20: // Load & process FED data
21: and historicalModel = async {
22:     (loading and normalization omitted)
23:     let! analyzed = asyncAnalyzeData normalized
24:     engine.TriggerAnalyzeHistoricalData(analyzed)
25:     let! res = asyncRunModel analyzed 
26:     engine.TriggerModelHistoricalData(res)
27:     return res }F# Web Snippets

In order to write the code in this way, I needed to add a couple of functions to the main module that implements the financial analysis. As you can see, when performing some operation we call functions like asyncAnalyzeData or asyncRunModel instead of the usual analyzeData and runModel from the sequential version. These asynchronous versions simply wrap a call to the synchronous version, which means that the workflow can be cancelled only after the underlying operation completes. If we wanted to support cancellation in a more granular way, we'd need to modify these functions and insert additional yield points inside them.

When doing a call inside asynchronous workflow (e.g. using let! or Async.StartChild), the cancellation token is automatically propagated. However, if you needed to call a long-running operation that isn't itself implemented using asynchronous workflows (and you cannot change it), you can use the Async.CancellationToken primitive to get the current token. The following snippet shows a very basic example:

1: let work = async { 
2:   let! tok = Async.CancellationToken
3:   let res = longRunningOp tok 
4:   return! asynchronousOp res }F# Web Snippets

The computation starts by getting a cancellation token (which is an asynchronous operation, because it is available only in the asynchronous context) and then calls the longRunningOp with the token as an argument. The operation returns some value (and may return early if the cancellation was triggered). When the cancellation is triggered, the asynchronous workflow will be cancelled at the next yield point (in this case the return! call) and so the rest of the workflow will not be executed. Note that we're not notified about the cancellation inside the workflow in any way (because it just stops running). The reaction to the cancellation has to be provided more explicitly...

Handling cancellation

Let's get back to our example with financial modeling. When the workflow is cancelled, we want to call the TriggerCompareModels method with a None value as the argument, so that the user interface can react. We'll add cancellation handling to the main workflow compare, which also uses try...with construct to handle exceptions that may occur during the modeling. Unlike exception handling, there is no built-in construct to handle cancellation, so we'll need to write the handling using a method call. This can be done using Async.TryCancelled:

 1: // When the workflow is cancelled, return 'None' as the result
 2: and compare = 
 3:   Async.TryCancelled
 4:     (async {
 5:       try
 6:           let! models = Async.Parallel [ marketModel; historicalModel ]
 7:           let res = compareModels models 
 8:           engine.TriggerCompareModels(Some res)
 9:       with e ->
10:           engine.TriggerErrorHandler()  },
11:       fun ce -> 
12:           engine.TriggerCompareModels(None))F# Web Snippets

The TryCancelled operation takes an asynchronous workflow as the first argument and a compensation function as the second. It returns a new workflow that runs the wrapped workflow and when a cancellation occurs in the wrapped workflow, it calls the provided compensation function. In the example above, we combine the call with the try...with construct to handle all possible "wrong situations".

The overall behavior of the computation if nothing wrong happens is that it runs and compares the two financial models and report the result to the GUI. When some part of the computation fails (and throws an exception), the workflow calls the TriggerErrorHandler (to notify GUI about an error) and when the operation is cancelled (by the user) the workflow calls TriggerCompareModels with empty result as the argument.

Summary

In this article, we looked at an improved F# version of the Financial dashboard example from the Parallel Programming with Microsoft .NET book. We started with a version discussed in my earlier article , did a minor refactoring to make reporting progress to the user interface easier and used this change to write a sequential version of the code that can be actually started. More interestingly, we also discussed the problem of cancellation in F# asynchronous workflows. Unlike .NET 4.0 tasks, asynchronous workflows support automatic (cooperative) cancellation. The basic steps to support cancellation that we demonstrated are:

• Make all long-running calls asynchronous - asynchronous workflow can cancel itself only at yield points (e.g. let!, do! and return!), so we need to provide enough of these in our code, so that the workflow can check for cancellation after every primitive step. This essentially means that we should write most of our operations as asynchronous workflows.
• Create cancellation token and pass it to Async.Start - if we want to cancel a workflow, we just need to create a cancellation token (using the CancellationTokenSource class from .NET 4.0) and pass it as an argument to the Start method. To cancel the running workflow, we can just call Cancel method of the token source.
• Handle cancellation using TryCancelled - if we want to provide some code that will be executed after the workflow is cancelled, we can use the Async.TryCancelled method. This works in a similar way as try...with construct, but it has to be written explicitly, because there is no syntactic support for handling cancellation.

Source Code

• Download F# Parallel Extras samples (ZIP)
• Get the latest version of samples from F# Community Samples at CodePlex