1

I just started to learn about TPL Dataflow and have a question as described below: enter image description here

  1. "Block"s 1, 2, 3 holds references to states. They modify the states and send messages downstream each time they receive a message. The number of such blocks varies.

  2. The "Aggregator" receives messages from the Blocks and check all the messages for errors. After all source blocks are Completed and aggregator passes a single message to the "Releaser".

  3. "Releaser" holds a reference to the state. It will know from "Aggregator" whether the updating is done correctly and will send the state with a success message or a failure message downstream.

        public static void Run()
{
    var sourceBlock1 = new TransformBlock<int, int>(x => x * 2);
    var sourceBlock2 = new TransformBlock<int, int>(x => x * 3);

    //How to implement the aggregator that aggregates messages from an unknown number of sources and then return a message
    //when all sources are complete?
    var aggregater = new TransformBlock<int, int[]>(x => ?); 
    var releaser = new TransformBlock<int[], int>(xs => xs.Sum());

    sourceBlock1.LinkTo(aggregater);
    sourceBlock2.LinkTo(aggregater);
    aggregater.LinkTo(releaser);

    sourceBlock1.Post(10);
    sourceBlock2.Post(20);

    targetBlock.Completion.Wait();
}
DavidY
  • 357
  • 5
  • 15
  • I noticed the vote for close and have reduced the scope of the question accordingly. – DavidY Feb 10 '23 at 03:25
  • I assume that the Blocks 1, 2, and 3 are `TransformBlock`s, and you want to know how to implement a custom dataflow block named `StateModifyingBlock`. Can you describe the desirable behavior of this block in details? Should it accept all the messages it receives, or postpone/reject some messages, and under which conditions? Should it complete always successfully, or it can also complete as faulted? Is it completing manually or automatically, and how? At which point is it offering the single output message to linked blocks? – Theodor Zoulias Feb 10 '23 at 13:08
  • 1
    In case you are looking for general directions about how to extend the TPL Dataflow library with custom blocks, search for this document *"Guide to Implementing Custom TPL Dataflow Blocks"*. It is a PDF document written by Zlatko Michailov. – Theodor Zoulias Feb 10 '23 at 17:34
  • 1
    Thanks a lot Theodor, I am trying to understand how to use Dataflow to handle shared state. I will get sample coding later to clarify the question. – DavidY Feb 11 '23 at 00:05
  • 1
    David in general writing custom dataflow blocks is unlikely to be a path to happiness. This library is notoriously difficult to extend. For example you can see [here](https://stackoverflow.com/questions/52633346/tpl-dataflow-batching-on-duration-or-threshold/69148085#69148085) what you have to do to implement a `BatchBlock` with timeout. It requires a lot of code, that is not easy to verify its correctness by reading it. – Theodor Zoulias Feb 11 '23 at 02:54
  • Hi Theodor I updated the question and added some coding. Not sure if I am on the right track. – DavidY Feb 12 '23 at 21:03
  • 1
    Related: [TPL Dataflow create aggregated result array from all incoming nodes (multiple producer, 1 consumer)](https://stackoverflow.com/questions/52074457/tpl-dataflow-create-aggregated-result-array-from-all-incoming-nodes-multiple-pr). – Theodor Zoulias Feb 13 '23 at 05:10

1 Answers1

1

In this line:

sourceBlock1.LinkTo(aggregater);

...the aggregater receives no notification that it has become the linked target of the sourceBlock1. The ISourceBlock<TOutput>.LinkTo method changes only the state of the source, not the target. The target will only become aware that it has been linked when it is offered the first message, via the ITargetBlock<TInput>.OfferMessage method:

public DataflowMessageStatus OfferMessage (
    DataflowMessageHeader messageHeader,
    TInput messageValue,
    ISourceBlock<TInput> source,
    bool consumeToAccept);

Even then, it's not guaranteed that the source argument will be a reference to the sourceBlock1, since the sourceBlock1 can opt to intercept an internal proxy ISourceBlock<TInput> implementation between itself and its linked target. So I don't think that you can achieve what you want using solely the methods of the existing interfaces. Maybe you could equip your custom aggregator block with an API that allows for bidirectional link-awareness. For example:

aggregater.Observe(sourceBlock1);
aggregater.Observe(sourceBlock2);

As for how to propagate the completion of multiple blocks to a single target block, take a look at these links:

Theodor Zoulias
  • 34,835
  • 7
  • 69
  • 104
  • 1
    Thanks Theodor! This is very helpful. There are probably no a one-size fits all answers, and custom blocks seem the way to go. – DavidY Feb 15 '23 at 01:53
  • I am also trying to use Rx with Dataflow and have another question :) – DavidY Feb 15 '23 at 02:32