How to Achieve Parallel Fan-out processing in Reactive Extensions?

Question

We already have parallel fan-out working in our code (using ParallelEnumerable) which is currently running on a 12-core, 64G RAM server. But we would like to convert the code to use Rx so that we can have better flexibility over our downstream pipeline.

Current Workflow:

1. We read millions of records from a database (in a streaming fashion).

2. On the client side, we then use a custom OrderablePartitioner<T> class to group the database records into groups. Let’s call an instance of this class: partioner.

3. We then use partioner.AsParallel().WithDegreeOfParallelism(5).ForAll(group => ProcessGroupOfRecordsAsync(group));
   
   Note: this could be read as “Process all the groups, 5 at a time in parallel.” (I.e. parallel fan-out).

4. ProcessGroupOfRecordsAsync() – loops through all the records in the group and turns them into hundreds or even thousands of POCO objects for further processing (i.e. serial fan-out or better yet, expand).
   
   Depending on the client’s needs:
   
   

5. This new serial stream of POCO objects are evaluated, sorted, ranked, transformed, filtered, filtered by manual process, and possibly more parallel and/or serial fanned-out throughout the rest of the pipeline.

6. The end of the pipeline may end up storing new records into the database, displaying the POCO objects in a form or displayed in various graphs.

The process currently works just fine, except that point #5 and #6 aren’t as flexible as we would like. We need the ability to swap in and out various downstream workflows. So, our first attempt was to use a Func<Tin, Tout> like so:

partioner.AsParallel
         .WithDegreeOfParallelism(5)
         .ForAll(group =>ProcessGroupOfRecordsAsync(group, singleRecord =>
             NextTaskInWorkFlow(singleRecord));

And that works okay, but the more we flushed out our needs the more we realized we are just re-implementing Rx.

Therefore, we would like to do something like the following in Rx:

IObservable<recordGroup> rg = dbContext.QueryRecords(inputArgs)
    .AsParallel().WithDegreeOfParallelism(5)
    .ProcessGroupOfRecordsInParallel();

If (client1)
    rg.AnalizeRecordsForClient1().ShowResults();

if (client2)
    rg.AnalizeRecordsForClient2()
      .AsParallel()
      .WithDegreeOfParallelism(3)
      .MoreProcessingInParallel()
      .DisplayGraph()
      .GetUserFeedBack()
      .Where(data => data.SaveToDatabase)
      .Select(data => data.NewRecords)
      .SaveToDatabase(Table2);
...
using(rg.Subscribe(groupId =>LogToScreen(“Group {0} finished.”, groupId);

Answer 1

It sounds like you might want to investigate Dataflows in the Task Parallel Library - This might be a better fit than Rx for dealing with part 5, and could be extended to handle the whole problem.

In general, I don't like the idea of trying to use Rx for parallelization of CPU bound tasks; its usually not a good fit. If you are not too careful, you can introduce inefficiencies inadvertently. Dataflows can give you nice way to parallelize only where it makes most sense.

From MSDN:

The Task Parallel Library (TPL) provides dataflow components to help increase the robustness of concurrency-enabled applications. These dataflow components are collectively referred to as the TPL Dataflow Library. This dataflow model promotes actor-based programming by providing in-process message passing for coarse-grained dataflow and pipelining tasks. The dataflow components build on the types and scheduling infrastructure of the TPL and integrate with the C#, Visual Basic, and F# language support for asynchronous programming. These dataflow components are useful when you have multiple operations that must communicate with one another asynchronously or when you want to process data as it becomes available. For example, consider an application that processes image data from a web camera. By using the dataflow model, the application can process image frames as they become available. If the application enhances image frames, for example, by performing light correction or red-eye reduction, you can create a pipeline of dataflow components. Each stage of the pipeline might use more coarse-grained parallelism functionality, such as the functionality that is provided by the TPL, to transform the image.

Answer 2

Kaboo!

As no one has provided anything definite, I'll point out that the source code can be browsed at GitHub at Rx. Taking a quick tour around, it looks like at least some of the processing (all of it?) is done on the thread-pool already. So, maybe it's not possibly to explicitly control the parallelization degree besides implementing your own scheduler (e.g. Rx TestScheduler), but it happens nevertheless. See also the links below, judging from the answers (especially the one provided by James in the first link), the observable tasks are queued and processed serially by design -- but one can provide multiple streams for Rx to process.

See also the other questions that are related and visible on the left side (by default). In particular it looks like this one, Reactive Extensions: Concurrency within the subscriber, could provide some answers to your question. Or maybe Run methods in Parallel using Reactive.

<edit: Just a note that if storing objects to database becomes a problem, the Rx stream could push the save operations to, say, a ConcurrentQueue, which would then be processed separately. Other option would be to let Rx to queue items with a proper combination of some time and number of items and push them to the database by bulk insert.