5
\$\begingroup\$

I needed BatchBlock with support for timeout (act greedy on what it received at that time) and by some search and study, this is the outcome.

What are drawbacks or rooms for improvement in this code?

class BatchedQueue
{
    static readonly Logger SlotClassLogger = LogManager.GetCurrentClassLogger();
    protected static Logger ClassLogger { get { return SlotClassLogger; } }
}

class BatchedQueue<T> : BatchedQueue
{
    readonly int _interval;
    readonly TimeSpan _loopWait;
    readonly BatchBlock<T> _queue;
    readonly Timer _timer;
    readonly BufferBlock<T> _source;

    public event Action<T[]> Received;

    protected virtual void OnReceived(T[] obj)
    {
        var handler = Received;
        if (handler != null) handler(obj);
    }

    public BatchedQueue(int interval, int size)
    {
        _interval = interval;

        _loopWait = TimeSpan.FromMilliseconds(2.19 * (1.3 + _interval)); //study

        _queue = new BatchBlock<T>(size, new GroupingDataflowBlockOptions { Greedy = true });
        _timer = new Timer(_ => _queue.TriggerBatch());

        var transformer = new TransformBlock<T, T>(v =>
        {
            _timer.Change(_interval, Timeout.Infinite);
            return v;
        });
        transformer.LinkTo(_queue);
        _source = new BufferBlock<T>();
        _source.LinkTo(transformer);

        Utl.SafeTask(Loop, t =>
        {
            if (t == null || t.Exception == null) return;
            ClassLogger.Error(t.Exception);
        }, TaskCreationOptions.LongRunning);
    }

    public virtual void Post(T t) { _source.Post(t); }
    void Loop()
    {
        while (true)
        {
            var buffer = _queue.Receive(_loopWait);
            if (buffer != null) { OnReceived(buffer); }
        }
    }
}
\$\endgroup\$
  • 3
    \$\begingroup\$ You can accomplish the same with much less code using TPL Dataflow. See this related post: stackoverflow.com/questions/9419442/… \$\endgroup\$ – Nathan Feb 26 '16 at 8:51
  • \$\begingroup\$ I built a PCQueue with Batching and Timeout using the TPL Dataflow Library and it's very simple and might be what you're looking for: stackoverflow.com/q/25656391/392184 Scroll to the bottom of the OP question for my final solution. \$\endgroup\$ – Lee Grissom Dec 3 '17 at 20:30
3
\$\begingroup\$
class BatchedQueue

What's the reason for having this separate class? Is it so that there is only one instance of SlotClassLogger? If that's the case, that sounds like premature optimization to me.


class BatchedQueue<T> : BatchedQueue

You're not implementing (or exposing in other way) any of the dataflow interfaces. That means this class can't easily be used as a part of a dataflow network.


readonly int _interval;

When you have a value that is of some unit of measure, you should always very clearly specify what unit is uses. (You don't want your Mars probe to crash, right?) You can do that either in the name of variable or in a comment.

Though it seems you're not using any common unit. That should also be document very clearly and it should also have a very good reason. If the unit is something specific to your domain, then I don't think it belongs here, a class like this should be reusable.

If the condition for triggering it not actually some time elapsing, then consider using that condition directly, instead of approximating it using a timer.


transformer.LinkTo(_queue);
_source = new BufferBlock<T>();
_source.LinkTo(transformer);

I don't see any reason for the _source buffer block here. You could send items directly into transformer.


while (true)
{
    var buffer = _queue.Receive(_loopWait);
    if (buffer != null) { OnReceived(buffer); }
}

This code is blocking a thread unnecessarily when there is no work to do. The simplest solution would be to use ActionBlock to execute OnReceived.


There is no way to tell the queue to stop processing or for your to wait until its processing is done. Consider adding the Complete()/Completed pair used in dataflow blocks.


When dealing with time, Rx is often better than Dataflow. A behavior similar to yours could be achived by using Buffer():

_source = new Subject<T>();
_source.Buffer(_loopWait, size)
       .Subscribe(OnReceived);

Though this measures the timeout since the last batch, not since the last element. I think you could achieve that using something like (based on this SO answer):

_source.Buffer(() => 
    Observable.Merge(_source.Throttle(_loopWait).Take(1), _source.Skip(size - 1).Take(1)))
    .Subscribe(OnReceived);
\$\endgroup\$
  • \$\begingroup\$ I think you are right on that Loop() is blocking. But how could I be notified about new data that is available at that time? And how by merging Throttle with _source I can force it to trigger eagerly? And is it safe to combine TPL Dataflow with Rx? \$\endgroup\$ – Kaveh Shahbazian Oct 11 '14 at 17:54
  • 1
    \$\begingroup\$ @KavehShahbazian For that notification, you can use ActionBlock as I suggested. Or, if you want to do it manually, await OutputAvailableAsync(). And I'm not suggesting you combine the two, I'm suggesting to switch to Rx completely. \$\endgroup\$ – svick Oct 11 '14 at 18:00

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.