Create a pipeline that doesn't overload memory

Question

I'm creating a pipeline where you need to:

• read from files and make transformations [TRANSFORM_1];
• with the transformed data I have to do 2 procedures:
  • do another transformation (very expensive for the CPU) [TRANSFORM_2]
  • perform an action that I will need to do other things not important for the purpose of the question [ACTION_1].

Below is the map with how I thought of doing:

ReadingFiles() -> TRANSFORM1_ -> BROADCASTBLOCK -> TRANSFORM2_ -> ... -> _______________________________________________\ -> ACTION_1

Requirements

two points must be met:

• Do not overload the available memory;
• All messages sent must arrive at the final block.

Half Solution

To fulfill the first requirement I simply have set ExecutionDataflowBlockOptions with BoundedCapacity = n in the various blocks. However, by using a back pressure with BroadcastBlock there is no guarantee that all messages will be sent.

Question

if there are no other solutions, how is it possible to implement a class that creates a custom BroadCast Block? I would make a class that works like a GuaranteedBroadCastBlock but implements the IPropagatorBlock interface

at the moment I have only read a few examples where they create an ActionBlock that acts as BroadCast through a method, but I think the best thing is to create a custom class rather than a method

N.B. About first part of the pipeline (when reading from file and sending to TRANSFORM_1) I already know that you need to use await TRANSFORM_1.SendAsync () to ensure that all messages are sent to the first block. The problem is in the BroadCast Block which is sending the most recent.

Answer 1

I thought about writing this class by taking a cue from other examples of classes and modifying something.

• Opinions?
• Is there something that would not fit?
• More than inheriting the ITargetBlock interface, could I inherit the IPropagatorBlock <T, U> interface and improve the connection with other TargetBlocks?
• is this class thread safe? if I created multiple instances of this class what could happen?

public class GuaranteedBroadcastBlock : ITargetBlock

{
    private readonly List<ITargetBlock<T>> _targets;
    private readonly ITargetBlock<T> block;

    public GuaranteedBroadcastBlock(DataflowBlockOptions options)
    {
        var task = new List<Task>();
        block = new ActionBlock<T>(
            async item =>
            {
                foreach (var target in _targets)
                {
                    task.Add(target.SendAsync(item));
                }
                await Task.WhenAll(task);
            }, new ExecutionDataflowBlockOptions
                   {
                       BoundedCapacity = options.BoundedCapacity,
                       CancellationToken = options.CancellationToken
                   });
        
    }

    public Task Completion => block.Completion.ContinueWith(task =>
    {
        foreach (var target in _targets)
        {
            if (task.Exception != null)
                target.Fault(task.Exception);
            else
                target.Complete();
        }
    });

    public void AddTarget(ITargetBlock<T> target) => _targets.Add(target);

    public void Complete() => block.Complete();

    public void Fault(Exception exception)
    {
        Console.WriteLine("Error in GuaranteedBroadcastBlock");
    }

    public DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
    {
        return block.OfferMessage(messageHeader, messageValue, source, consumeToAccept);
    }

}