6
\$\begingroup\$

The following class implements a buffer which captures fast changing datapoints or frequently raised events and dispatches them in batches. It will asynchronously call a processing function for the buffered items under any of the following conditions:

  • A configurable timeout has expired since the last time an item was added
  • A configurable maximum number of items was added
  • The buffer is being disposed

There are several buffering strategies available which control how the items are accumulated. It was written in times of .NET 3.5 and still needs to work with it.

Looking for general improvements, potential pitfalls I missed, design improvements etc.

A couple if things which annoy me but I never got around changing:

  1. Violation of open/closed principle in regards of the accumulation strategies. Hasn't been an issue so far but should be fixed at some point I guess.
  2. Using a dedicated processing thread. Putting the main method on the thread pool is not really an option because it's long running. Doing it on a timer callback would be nice but it's a bit tricky with the different signals which can trigger a dispatch of the buffered items.

Code:

public class AccumulationBuffer<TKey, TValue> : IDisposable
{
    public enum RecordingStrategy
    {
        AllFifo,
        AllLifo,
        FirstWins,
        LastWins
    }

    private interface IStorage : IEnumerable<KeyValuePair<TKey, TValue>>
    {
        void Add(KeyValuePair<TKey, TValue> pair);
        void Clear();
    }

    private class StorageAllFifo : Queue<KeyValuePair<TKey, TValue>>, IStorage
    {
        public StorageAllFifo(int capacity)
            : base(capacity)
        {
        }

        public void Add(KeyValuePair<TKey, TValue> pair)
        {
            Enqueue(pair);
        }
    }

    private class StorageAllLifo : Stack<KeyValuePair<TKey, TValue>>, IStorage
    {
        public StorageAllLifo(int capacity)
            : base(capacity)
        {
        }

        public void Add(KeyValuePair<TKey, TValue> pair)
        {
            Push(pair);
        }
    }

    private class StorageMap : Dictionary<TKey, TValue>, IStorage
    {
        private readonly bool _SkipIfExists;

        public StorageMap(int capacity, bool skipIfExists)
            : base(capacity)
        {
            _SkipIfExists = skipIfExists;
        }

        public void Add(KeyValuePair<TKey, TValue> pair)
        {
            if (!_SkipIfExists || !ContainsKey(pair.Key))
                this[pair.Key] = pair.Value;
        }
    }


    private readonly Action<IEnumerable<KeyValuePair<TKey, TValue>>> _ItemProcessor;
    private readonly IStorage _Storage;
    private readonly RecordingStrategy _Strategy;
    private readonly object _StorageLock = new object();
    private Thread _MonitorThread;
    private volatile bool _Quit;
    private readonly AutoResetEvent _MonitorEvent;
    private DateTime _LastAddedTime;
    private readonly TimeSpan _InactivityTime;
    private readonly int _MaximumAccumulationCount;
    private int _CurrentAccumulationCount;

    public AccumulationBuffer(RecordingStrategy strategy, TimeSpan inactivityTime, int maximumAccumulationCount, Action<IEnumerable<KeyValuePair<TKey, TValue>>> itemProcessor)
    {
        if (itemProcessor == null)
            throw new ArgumentNullException("itemProcessor");

        _ItemProcessor = itemProcessor;
        _LastAddedTime = DateTime.MinValue;
        _InactivityTime = inactivityTime;
        _MaximumAccumulationCount = maximumAccumulationCount;
        _MonitorEvent = new AutoResetEvent(false);
        _Quit = false;
        _CurrentAccumulationCount = 0;

        _Strategy = strategy;
        switch (_Strategy)
        {
            case RecordingStrategy.AllFifo:
                _Storage = new StorageAllFifo(maximumAccumulationCount);
                break;
            case RecordingStrategy.AllLifo:
                _Storage = new StorageAllLifo(maximumAccumulationCount);
                break;
            case RecordingStrategy.FirstWins:
                _Storage = new StorageMap(maximumAccumulationCount, true);
                break;
            case RecordingStrategy.LastWins:
                _Storage = new StorageMap(maximumAccumulationCount, false);
                break;
            default:
                throw new ArgumentOutOfRangeException("strategy");
        }

        _MonitorThread = new Thread(MonitorLoop)
                             {
                                 IsBackground = true,
                                 Name = string.Format("Accumulation Buffer Monitor ({0}/{1})", typeof (TKey).Name, typeof (TValue).Name)
                             };
        _MonitorThread.Start();
    }

    public void Add(TKey key, TValue value)
    {
        lock (_StorageLock)
        {
            _Storage.Add(new KeyValuePair<TKey, TValue>(key, value));
            _LastAddedTime = DateTime.Now;
            _CurrentAccumulationCount++;
            _MonitorEvent.Set();
        }
    }

    private void MonitorLoop()
    {
        while (!_Quit)
        {
            bool timeout = false;
            while (!timeout && !_Quit)
            {
                TimeSpan wait = TimeSpan.FromMilliseconds(-1);
                lock (_StorageLock)
                {
                    if (_CurrentAccumulationCount >= _MaximumAccumulationCount)
                        break;

                    if (_LastAddedTime != DateTime.MinValue)
                    {
                        wait = _InactivityTime;
                    }
                }
                timeout = !_MonitorEvent.WaitOne(wait); // WaitOne returns false if it timed out
            }

            // if we are here we got either 
            // - woken up to quit OR
            // - the event has timed out after the inactivity timeout has passed OR
            // - the maximum number of items was accumulated

            lock (_StorageLock)
            {
                // always process pending items, whether we quit or not
                if (_CurrentAccumulationCount > 0)
                {
                    var toProcess = new List<KeyValuePair<TKey, TValue>>(_Storage);
                    _Storage.Clear();
                    _LastAddedTime = DateTime.MinValue;
                    _CurrentAccumulationCount = 0;
                    ThreadPool.QueueUserWorkItem(x => _ItemProcessor(toProcess));
                }
                if (_Quit) return;
            }
        }
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    protected virtual void Dispose(bool disposing)
    {
        if (disposing)
        {
            if (_MonitorThread != null)
            {
                _Quit = true;
                _MonitorEvent.Set();
                if (!_MonitorThread.Join(100))
                {
                    _MonitorThread.Abort();
                }
                _MonitorThread = null;
            }
        }
    }
}
\$\endgroup\$
4
\$\begingroup\$

Also, there isn't much to say (you have already mentioned the flaws) some nitpickings

  • You should be consistent in your coding style

    if (_CurrentAccumulationCount >= _MaximumAccumulationCount)
        break;
    
    if (_LastAddedTime != DateTime.MinValue)
    {
        wait = _InactivityTime;
    }  
    

    Sometimes you are using braces {} for single if statements (which i preffer) and sometimes not.

  • I would extract the retrieving of the "timeout TimeSpan" into a separate method,

    private TimeSpan GetWaitingTime()
    {
         lock (_StorageLock)
         {
             if(_CurrentAccumulationCount >= _MaximumAccumulationCount)
             {
                 return TimeSpan.Zero;
             }
             if (_LastAddedTime != DateTime.MinValue)
             {
                 return _InactivityTime;
             }
         }
         return TimeSpan.FromMilliseconds(-1);
    }
    

    and refactor the while (!timeout && !_Quit) loop to

    bool hasTimedOut = false;
    while (!hasTimedOut && !_Quit)
    {
        TimeSpan waitingTime = GetWaitingTime();
        if(waitingTime == TimeSpan.Zero)
        {
            break;
        }
    
        hasTimedOut = !_MonitorEvent.WaitOne(waitingTime); // WaitOne returns false if it timed out
    }
    
  • The constructor of StorageAllFifo and StorageAllLifo can be removed, as the base constructor is called anyway.

\$\endgroup\$
3
\$\begingroup\$
private int _CurrentAccumulationCount;

Why is this a separate field? I would instead let IStorage manage the count by adding it as a property to that interface.


if (!_SkipIfExists || !ContainsKey(pair.Key))
    this[pair.Key] = pair.Value;

I find this condition somewhat confusing. I think using De Morgan's laws makes it slightly better:

if (!(_SkipIfExists && ContainsKey(pair.Key)))
    this[pair.Key] = pair.Value;

_LastAddedTime = DateTime.Now;

I think you should use DateTime.UtcNow. Since Now uses local time, it behaves weirdly around DST changes.


 ThreadPool.QueueUserWorkItem(x => _ItemProcessor(toProcess));

This means that _ItemProcessor can execute multiple times concurrently. Is that okay?

(Also, if you're not going to use the parameter of a lambda, it's common to call it _, to make it clear.)

\$\endgroup\$
  • \$\begingroup\$ Good points, _CurrentAccumulationCount is kept separate because I didn't want each strategy having to implement it as it's the same behavior (and most strategies have a base class already). Regarding the condition, well if I read the original it reads as if i don't want to skip existing items or the item is no present then add it - the rephrased condition would be a bit more awkward to read (at least to me). Yes the processor needs to be re-entrant safe, that's fine. \$\endgroup\$ – ChrisWue Oct 19 '14 at 0:37
  • \$\begingroup\$ "I didn't want each strategy having to implement it" Since Queue, Stack and Dictionary already implement Count, it will work automatically if you just add the property to IStorage, just like you did it with Clear(). \$\endgroup\$ – svick Oct 19 '14 at 1:17
  • \$\begingroup\$ sure but that's not what that field represents. It represents the number of times Add was called effectively. For example this would allow to accumulate the occurrence of a single event (when using FirstWins/LastWins) without starving the consumer. I guess it should be renamed to express this more clearly. \$\endgroup\$ – ChrisWue Oct 19 '14 at 7:46

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.