learning-tpl-dataflow.cs

using System;
using System.Collections.Generic;
using System.Net.Http;
using System.Threading.Tasks.Dataflow;
using System.Threading.Tasks;
using System.Collections.Concurrent;

namespace Pipelining
{
    public class WorkItem
    {
        public string Payload { get; set; } = "";
        public double Priority { get; set; } = 0;
    };
    
    public static class Counters
    {
        public static int TaskDroppedFrontDoor = 0;
        public static int TaskDroppedPriorityQueue = 0;
    }

    public class PipelineProcessor
    {
        // Dataflow blocks for scoring and processing.
        private readonly TransformBlock<WorkItem, WorkItem> _scoringBlock;
        private readonly TransformBlock<WorkItem, WorkItem> _throttlingBlock;
        private readonly PriorityBufferBlock<WorkItem, double> _priorityBufferBlock;
        private readonly BatchBlock<WorkItem> _batchBlock;
        private readonly ActionBlock<WorkItem[]> _sendingBlock;
        private readonly HttpClient _httpClient = new HttpClient();

        public ITargetBlock<WorkItem> InputBlock => _scoringBlock;

        public PipelineProcessor()
        {
            _scoringBlock = new TransformBlock<WorkItem, WorkItem>(async item =>
            {
                item.Priority = await EvaluatePriorityAsync(item);
                return item;
            }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2, BoundedCapacity = 500 });

            //_throttlingBlock = new TransformBlock<WorkItem, WorkItem>(async workItem =>
            //{
            //    await Task.Delay(500);
            //    return workItem;
            //},
            //new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4, BoundedCapacity = 100 });

            _priorityBufferBlock = new PriorityBufferBlock<WorkItem, double>( boundedCapacity: 1000 );

            _batchBlock = new BatchBlock<WorkItem>(32, new GroupingDataflowBlockOptions
            {
                Greedy = true,
            });

            _sendingBlock = new ActionBlock<WorkItem[]>(
                async batch =>
                {
                    // Convert array to a list (if needed).
                    await SendBatchAsync(new List<WorkItem>(batch));
                },
                new ExecutionDataflowBlockOptions
                {
                    MaxDegreeOfParallelism = 1,
                }
            );

            _scoringBlock.LinkTo(_priorityBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });
            // _throttlingBlock.LinkTo(_priorityBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });
            _priorityBufferBlock.LinkTo(_batchBlock, new DataflowLinkOptions { PropagateCompletion = true });
            _batchBlock.LinkTo(_sendingBlock, new DataflowLinkOptions { PropagateCompletion = true });

            // Start the background batching process.
            StartBatchTrigger();
        }

        private async void StartBatchTrigger()
        {
            while (true)
            {
                await Task.Delay(TimeSpan.FromMilliseconds(10));
                _batchBlock.TriggerBatch();
            }
        }

        // Simulated method to evaluate priority.
        private Task<double> EvaluatePriorityAsync(WorkItem item)
        {
            return Task.FromResult<double>(new Random().NextDouble() * 100);
        }

        // Simulated method to send a batch via HttpClient.
        private async Task SendBatchAsync(List<WorkItem> batch)
        {
            try
            {
                //string json = JsonSerializer.Serialize(batch);
                //var content = new StringContent(json, Encoding.UTF8, "application/json");
                //HttpResponseMessage response = await _httpClient.PostAsync("https://example.com/api/tasks", content);
                //response.EnsureSuccessStatusCode();
                await Task.Delay(100); // pretend to do the work
                var now = DateTime.Now;
                Console.WriteLine($"[{now}] Sent batch of {batch.Count} items.");
                Console.WriteLine($"[{now}] Up until now, {Counters.TaskDroppedFrontDoor} has been dropped at front door and {Counters.TaskDroppedPriorityQueue} has been droped on priority queue.");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Error sending batch: {ex.Message}");
            }
        }

        // Signal that no more items will be posted.
        public void Complete()
        {
            _scoringBlock.Complete();
        }

        // Expose overall completion via the buffer (which completes when the pipeline is done).
        public Task Completion => _sendingBlock.Completion;
    }

    // Propagates data in a sliding window fashion.
    public class PriorityBufferBlock<T, TPriority> : IPropagatorBlock<T, T>, IDisposable
    {
        private readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
        private readonly IReceivableSourceBlock<T> _source;
        private ITargetBlock<T> _target = null;
        private readonly Task _pusher;
        private int _boundedCapacity;
        private bool _targetCompleted = false;

        // Constructs a SlidingWindowBlock object.
        public PriorityBufferBlock(int boundedCapacity)
        {
            var source = new BufferBlock<T>();
            _boundedCapacity = boundedCapacity;
            // _source = source;
            _pusher = Task.Run(async () =>
            {
                while (!(_targetCompleted && _queue.IsEmpty))
                {
                    if (_queue.TryDequeue(out var item))
                    {
                        if (!source.Post(item))
                        {
                            _queue.Enqueue(item);
                            await Task.Delay(100);
                        }
                    }
                    else
                    {
                        await Task.Delay(100);
                    }
                }
                Counters.TaskDroppedPriorityQueue++;
                source.Complete();
            });
        }

        #region ISourceBlock<TOutput> members

        // Links this dataflow block to the provided target.
        public IDisposable LinkTo(ITargetBlock<T> target, DataflowLinkOptions linkOptions)
        {
            // return _source.LinkTo(target, linkOptions);
            _target = target;
            return this;
        }

        // Called by a target to reserve a message previously offered by a source
        // but not yet consumed by this target.
        bool ISourceBlock<T>.ReserveMessage(DataflowMessageHeader messageHeader,
           ITargetBlock<T> target)
        {
            return _source.ReserveMessage(messageHeader, target);
        }

        // Called by a target to consume a previously offered message from a source.
        T ISourceBlock<T>.ConsumeMessage(DataflowMessageHeader messageHeader,
           ITargetBlock<T> target, out bool messageConsumed)
        {
            return _source.ConsumeMessage(messageHeader,
               target, out messageConsumed);
        }

        // Called by a target to release a previously reserved message from a source.
        void ISourceBlock<T>.ReleaseReservation(DataflowMessageHeader messageHeader,
           ITargetBlock<T> target)
        {
            _source.ReleaseReservation(messageHeader, target);
        }

        #endregion

        #region ITargetBlock<TInput> members

        private int itemsDropped = 0;

        // Asynchronously passes a message to the target block, giving the target the
        // opportunity to consume the message.
        DataflowMessageStatus ITargetBlock<T>.OfferMessage(DataflowMessageHeader messageHeader,
           T messageValue, ISourceBlock<T> source, bool consumeToAccept)
        {
            if (_queue.Count == _boundedCapacity)
            {
                var _ = _queue.TryDequeue(out var result);
                _queue.Enqueue(messageValue);
                itemsDropped++;
                return DataflowMessageStatus.Accepted;
            }
            else
            {
                _queue.Enqueue(messageValue);
                return DataflowMessageStatus.Accepted;
            }
        }

        #endregion

        #region IDataflowBlock members

        // Gets a Task that represents the completion of this dataflow block.
        public Task Completion { get { return _source.Completion; } }

        // Signals to this target block that it should not accept any more messages,
        // nor consume postponed messages.
        public void Complete()
        {
            _targetCompleted = true;
        }

        public void Fault(Exception error)
        {
            throw error;
        }

        public void Dispose()
        {
            _target = null;
        }

        #endregion
    }

    class Program
    {
        public static async Task Main(string[] args)
        {
            var processor = new PipelineProcessor();
            await Task.Delay(500);
            var totalItems = 0;
            for (int t = 0; t < 100; t++)
            {
                for (int i = 0; i < 100; i++)
                {
                    var workItem = new WorkItem { Payload = $"Payload {t * 10000 + i}" };
                    var queued = processor.InputBlock.Post(workItem);

                    if (!queued)
                    {
                        Counters.TaskDroppedFrontDoor++;
                        // await Task.Delay(1);
                    }
                    totalItems += 1000;
                }
                await Task.Delay(1);
            }

            // Signal completion so the pipeline will eventually shut down.
            processor.Complete();

            // Wait for the entire pipeline to finish.
            await processor.Completion;

            Console.WriteLine($"{totalItems} items processed. {Counters.TaskDroppedFrontDoor} items dropped due to back pressure.");
        }
    }
}