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:
- 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.
- 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;
}
}
}
}
