In a previous posting, available here, I discussed a pattern for running long running processes synchronously and in parallel, but terminating processing when a certain condition was met. As an F# exercise I decided to write a similar code pattern using PSeq, an F#-style API for parallel operations on sequences that are part in .NET 4.0 as System.Linq.ParallelEnumerable class, available in the F# Power Pack.
A follow-up article demonstrating this code without the need for the Command patter can be found here.
Once again, as this is only a small amount of code, here is a full listing:
A little explanation of this code is warranted. As in the previous C# implementation, the operation input definitions (based on the Command pattern) allow for the definition of an operation to be performed with associated parameters.
The ForEachPartial extension method once again takes in a seq of operations, and a results collector operation. It is this results collector operation that informs the parallel processing whether any further iterations are required.
One of the big differences in this implementation is the use of PSeq, which is based on PLINQ, and as a result a loop state is unavailable. This means the cancellation token is used to terminate future scheduled iterations. The other difference is the definition of the collectorState and operation functions. The collectorState function takes the result of the operation execution and passes it into the collector, with the necessary locking. The return from the collector operation determines if further iterations are cancelled. The operation function merely wraps the call to the long running process.
To perform some testing here are some simple definitions:
The operation merely returns the square of a number, after sleeping for 2 seconds, and the collector terminates when the collected results top a given value.
So to call this code, in an object based fashion one can use:
However this does not really follow the F# calling schematics. To support this the F# code includes the following extension to PSeq:
Using this function one can now write, in a much more functional style:
To finalize the testing code, one of the testing tasks is to record the operation timings and output the result to the console. Using the power of F# function I have defined the following two functions:
Thus to call the operations in parallel, record and display the timing we can write:
If you are using F# hopefully you will find this code useful.
Written by Carl Nolan