How to call TriggerBatch automagically after a timeout if the number of queued items is less than the BatchSize?

Question

Using Dataflow CTP (in the TPL)

Is there a way to call BatchBlock.TriggerBatch automatically if the number of currently queued or postponed items is less than the BatchSize, after a timeout?

And better: this timeout should be reset to 0 each time the block receives a new item.

Answer 1

Yes, you can accomplish this rather elegantly by chaining together blocks. In this case you want to setup a TransformBlock which you link "before" the BatchBlock. That would look something like this:

Timer triggerBatchTimer = new Timer(() => yourBatchBlock.TriggerBatch());

TransformBlock<T, T> timeoutTransformBlock = new TransformBlock<T, T>((value) =>
{
    triggerBatchTimer.Change(5000, Timeout.Infinite);

    return value; 
});

timeoutTransformBlock.LinkTo(yourBatchBlock);

yourBufferBlock.LinkTo(timeoutTransformBlock);

Answer 2

Here is a policed version of Drew Marsh's idea. This implementation uses the DataflowBlock.Encapsulate method to create a dataflow block that encapsulates the timer+batch functionality. Beyond the new argument timeout, the CreateBatchBlock method also supports all options available to the normal BatchBlock constructor.

public static IPropagatorBlock<T, T[]> CreateBatchBlock<T>(int batchSize,
    int timeout, GroupingDataflowBlockOptions dataflowBlockOptions = null)
{
    dataflowBlockOptions = dataflowBlockOptions ?? new GroupingDataflowBlockOptions();
    var batchBlock = new BatchBlock<T>(batchSize, dataflowBlockOptions);
    var timer = new System.Threading.Timer(_ => batchBlock.TriggerBatch());
    var transformBlock = new TransformBlock<T, T>((T value) =>
    {
        timer.Change(timeout, Timeout.Infinite);
        return value;
    }, new ExecutionDataflowBlockOptions()
    {
        BoundedCapacity = dataflowBlockOptions.BoundedCapacity,
        CancellationToken = dataflowBlockOptions.CancellationToken,
        EnsureOrdered = dataflowBlockOptions.EnsureOrdered,
        MaxMessagesPerTask = dataflowBlockOptions.MaxMessagesPerTask,
        NameFormat = dataflowBlockOptions.NameFormat,
        TaskScheduler = dataflowBlockOptions.TaskScheduler
    });
    transformBlock.LinkTo(batchBlock, new DataflowLinkOptions()
    {
        PropagateCompletion = true
    });
    return DataflowBlock.Encapsulate(transformBlock, batchBlock);
}

Caution: As pointed by @Jeff in a comment, there is a race condition with this approach. In case the timeout is very small (in the range of milliseconds), the transformBlock will be racing with the timer to pass the data onto the batchBlock, and the timer may fire first before the batchBlock has anything in it yet. Worst case scenario, we hang indefinitely. No more messages end up in the queue because they are waiting on a handful of previous ones to complete, but there is one straggler sitting in the latest buffer that will never trigger.

---

Alternative: below is a BatchUntilInactiveBlock<T> class that offers the whole range of the BatchBlock<T> functionality. This implementation is a thin wrapper around a BatchBlock<T> instance. It has less overhead than the previous CreateBatchBlock implementation, while having a similar behavior. It's not affected by the race condition mentioned earlier.

/// <summary>
/// Provides a dataflow block that batches inputs into arrays.
/// A batch is produced when the number of currently queued items becomes equal
/// to BatchSize, or when a Timeout period has elapsed after receiving the last item.
/// </summary>
public class BatchUntilInactiveBlock<T> : IPropagatorBlock<T, T[]>,
    IReceivableSourceBlock<T[]>
{
    private readonly BatchBlock<T> _source;
    private readonly Timer _timer;
    private readonly TimeSpan _timeout;

    public BatchUntilInactiveBlock(int batchSize, TimeSpan timeout,
        GroupingDataflowBlockOptions dataflowBlockOptions)
    {
        _source = new BatchBlock<T>(batchSize, dataflowBlockOptions);
        _timer = new Timer(_ => _source.TriggerBatch());
        _timeout = timeout;
    }

    public BatchUntilInactiveBlock(int batchSize, TimeSpan timeout) : this(batchSize,
        timeout, new GroupingDataflowBlockOptions())
    { }

    public int BatchSize => _source.BatchSize;
    public TimeSpan Timeout => _timeout;
    public Task Completion => _source.Completion;
    public int OutputCount => _source.OutputCount;

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

    void IDataflowBlock.Fault(Exception exception)
        => ((IDataflowBlock)_source).Fault(exception);

    public IDisposable LinkTo(ITargetBlock<T[]> target,
        DataflowLinkOptions linkOptions)
            => _source.LinkTo(target, linkOptions);

    public void TriggerBatch() => _source.TriggerBatch();

    public bool TryReceive(Predicate<T[]> filter, out T[] item)
        => _source.TryReceive(filter, out item);

    public bool TryReceiveAll(out IList<T[]> items)
        => _source.TryReceiveAll(out items);

    DataflowMessageStatus ITargetBlock<T>.OfferMessage(
        DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source,
        bool consumeToAccept)
    {
        var offerResult = ((ITargetBlock<T>)_source).OfferMessage(messageHeader,
            messageValue, source, consumeToAccept);
        if (offerResult == DataflowMessageStatus.Accepted)
            _timer.Change(_timeout, System.Threading.Timeout.InfiniteTimeSpan);
        return offerResult;
    }

    T[] ISourceBlock<T[]>.ConsumeMessage(DataflowMessageHeader messageHeader,
        ITargetBlock<T[]> target, out bool messageConsumed)
            => ((ISourceBlock<T[]>)_source).ConsumeMessage(messageHeader,
                target, out messageConsumed);

    bool ISourceBlock<T[]>.ReserveMessage(DataflowMessageHeader messageHeader,
        ITargetBlock<T[]> target)
            => ((ISourceBlock<T[]>)_source).ReserveMessage(messageHeader, target);

    void ISourceBlock<T[]>.ReleaseReservation(DataflowMessageHeader messageHeader,
        ITargetBlock<T[]> target)
            => ((ISourceBlock<T[]>)_source).ReleaseReservation(messageHeader, target);
}

The _timer is scheduled immediately after the BatchBlock<T> has been offered and has accepted a message. There is no window of time between scheduling the timer and offering the message, so there is no race.

---

Disclaimer: The behavior of the above implementations is not ideal, in that they produce short batches even in cases that they shouldn't. The ideal behavior would be to produce a short batch only in case the batch can be propagated instantly to a consumer downstream. Producing short batches and just storing them in the block's output buffer doesn't make much sense. This deviation from the ideal behavior can only be observed if the CreateBatchBlock<T>/BatchUntilInactiveBlock<T> is not rigorously pumped, for example if the linked block downstream is bounded and has reached its maximum capacity.

Answer 3

Thanks to Drew Marsh for the idea of using a TransformBlock which greatly helped me with a recent solution. However, I believe that the timer needs to be reset AFTER the batch block (i.e. after it has either been triggered by the batch size being reached OR the TriggerBatch method being explicitly called within the timer callback). If you reset the timer every time you get a single item then it can potentially keep resetting several times without actually triggering a batch at all (constantly pushing the "dueTime" on the Timer further away).

This would make the code snippet look like the following:

Timer triggerBatchTimer = new Timer(() => yourBatchBlock.TriggerBatch(), null, 5000, Timeout.Infinite);

TransformBlock<T[], T[]> timeoutTransformBlock = new TransformBlock<T[], T[]>((value) =>
{
    triggerBatchTimer.Change(5000, Timeout.Infinite);

    return value; 
});

yourBufferBlock.LinkTo(yourBatchBlock);
yourBatchBlock.LinkTo(timeoutTransformBlock)
timeoutTransformBlock.LinkTo(yourActionBlock);

// Start the producer which is populating the BufferBlock etc.

Answer 4

Here is a solution which builds upon previous answers. This method encapsulates an exisiting BatchBlock with one the pushes out batches at least as ofter as the timeout.

The other answers doesn't handle if there are no items is the batch block when the timer is triggered. In that case the other solutions waits until the batch is full. We had that issue in our non-production environments which made testing harder. This solution make sure that after an item is posted to the BatchBlock, it is propagated after at most timeout seconds.

public static IPropagatorBlock<T, T[]> CreateTimeoutBatchBlock<T>(BatchBlock<T> batchBlock, int timeout)
{
    var timespan = TimeSpan.FromSeconds(timeout);
    var timer = new Timer(
        _ => batchBlock.TriggerBatch(),
        null,
        timespan,
        timespan);
    var transformBlock = new TransformBlock<T[], T[]>(
        value =>
        {
            // Reset the timer when a batch has been triggered
            timer.Change(timespan, timespan);
            return value;
        });
    batchBlock.LinkTo(transformBlock, new DataflowLinkOptions { PropagateCompletion = true });
    return DataflowBlock.Encapsulate(batchBlock, transformBlock);
}

Answer 5

You can use link options

_transformManyBlock.LinkTo(_batchBlock, new DataflowLinkOptions {PropagateCompletion = true});