10
\$\begingroup\$

I'm implementing a "live analytics" feature and I need to always keep track of the last N items that were added to a collection so I figured I'd make a circular buffer since I couldn't find one.

I tried to figure out to use Interlocked but couldn't find out how to use Increment or CompareExchange in an obviously correct way - so I took the lock.

Review/advice is appreciated.

public class CircularBuffer<T>
{
    private readonly T[] _data;
    private readonly ReaderWriterLockSlim _lock = new ReaderWriterLockSlim();
    private int _i = 0;
    private bool _isOneCycleFinished = false;
    private readonly int _size;

    public CircularBuffer(int size)
    {

        if (size < 0)
        {
            throw new ArgumentException($"{nameof(size)} cannot be negative");
        }
        _i = 0;
        _data = new T[size];
        _size = size;
    }

    public IEnumerable<T> Latest()
    {
        try
        {
            _lock.EnterReadLock();
            if (!_isOneCycleFinished)
            {
                return _data.Take(_i).ToList();
            }
            return _data.Concat(_data).Skip(_i).Take(_size).ToList();
        }
        finally
        {
            _lock.ExitReadLock();
        }
    }


    public void Add(T t)
    {
        try
        {
            _lock.EnterWriteLock();
            _i = (_i + 1)%_size;
            if (!_isOneCycleFinished && _i == 0)
            {
                _isOneCycleFinished = true;
            }
            _data[_i] = t;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }
}
\$\endgroup\$
  • 1
    \$\begingroup\$ I think you can use Interlocked.Exchange(ref _i, (_i + 1) % _size); and get rid of the local i variable. \$\endgroup\$ – Stefano d'Antonio Jul 5 '16 at 12:36
  • \$\begingroup\$ What's the point of _isOneCycleFinished? \$\endgroup\$ – 404 Jul 5 '16 at 12:38
  • \$\begingroup\$ @eurotrash I want to return the data so far - it determines if I need to only fetch the first _i items or I need to return the _size items starting at i. If the buffer is of ints and of size 4 - It starts as 0000, I add 1 and 2 and get 1200 - with Latest I expect to get 1,2. Now if I add 3 4 5 6 in - the buffer is at 5634, I expect to get 3456. \$\endgroup\$ – Benjamin Gruenbaum Jul 5 '16 at 12:39
  • \$\begingroup\$ Is this class meant to be thread safe? (It's not.) \$\endgroup\$ – 404 Jul 5 '16 at 12:42
  • 1
    \$\begingroup\$ @BenjaminGruenbaum Well that was before you edited it. Add()'s write to _isOneCycleFinished and Latest()'s read of _isOneCycleFinished were not synchronised, nor were the writing and reading of _i. \$\endgroup\$ – 404 Jul 5 '16 at 14:55
6
\$\begingroup\$

Like @Gareth mentioned in his comment it would be perfect to use the ConcurrentQueue<T>.

You answered @Gareth the queue needs to be bounded in size but where is the problem. We can just add a check to the Add() method and if the queue's size is equal to the bounded size we will just dequeue one item.

If the passed in size parameter of the constructor is 0 you will get a DivideByZeroException at _i = (_i + 1)%_size; and it doesn't really make sense to have a CircularBuffer with size == 0, does it ?

As you will notice in the adjusted code, I have placed the _lock.EnterWriteLock(); outside of the try..finally because what happens if the call to EnterWriteLock() fails ?

public class CircularBuffer<T>
{
    private readonly ConcurrentQueue<T> _data;
    private readonly ReaderWriterLockSlim _lock = new ReaderWriterLockSlim();
    private readonly int _size;

    public CircularBuffer(int size)
    {
        if (size < 1)
        {
            throw new ArgumentException($"{nameof(size)} cannot be negative or zero");
        }
        _data = new ConcurrentQueue<T>();
        _size = size;
    }

    public IEnumerable<T> Latest()
    {
        return _data.ToArray();
    }

    public void Add(T t)
    {
        _lock.EnterWriteLock();
        try
        {
            if (_data.Count == _size)
            {
                T value;
                _data.TryDequeue(out value);
            }

            _data.Enqueue(t);
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }
}

although using a ConcurrentQueue is IMO the cleaner way I will post a solution using a generic array as well like so

public class CircularBuffer<T>
{
    private readonly ReaderWriterLockSlim slimLock = new ReaderWriterLockSlim();
    private readonly int capacity;
    private readonly T[] buffer;
    private readonly int upperBound;

    private int currentIndex = -1;
    private int currentStart = 0;

    public CircularBuffer(int size)
    {
        if (size < 1)
        {
            throw new ArgumentException($"{nameof(size)} cannot be negative nor zero");
        }

        buffer = new T[size];
        capacity = size;
        upperBound = size - 1;
    }

    public void Add(T value)
    {
        slimLock.EnterWriteLock();
        try
        {

            if (IsFull || currentIndex == upperBound)
            {
                currentStart = FetchNextSlot(currentStart);
                IsFull = true;
            }

            currentIndex = FetchNextSlot(currentIndex);
            buffer[currentIndex] = value;

        }
        finally
        {
            slimLock.ExitWriteLock();
        }

    }

    public bool IsFull { get; private set; }

    private int FetchNextSlot(int value)
    {
        return (value + 1) % capacity;
    }

    public IEnumerable<T> Latest()
    {
        slimLock.EnterReadLock();
        try
        {
            return FetchItems().ToArray();
        }
        finally
        {
            slimLock.ExitReadLock();
        }
    }

    private IEnumerable<T> FetchItems()
    {
        IEnumerable<T> fetchedItems = Enumerable.Empty<T>();
        if (IsFull)
        {
            return FetchItems(currentStart, upperBound);
        }
        return fetchedItems.Concat(FetchItems(0, currentIndex));
    }
    private IEnumerable<T> FetchItems(int start, int end)
    {
        for (int i = start; i <= end; i++)
        {
            yield return buffer[i];
        }
    }
}  
\$\endgroup\$
  • \$\begingroup\$ Wouldn't that be backwards? Wouldn't I be relying on implementation detail here (that ConcurrentQueue uses a circular buffer internally)? I'm making the queue too big and then removing one item from it. \$\endgroup\$ – Benjamin Gruenbaum Jul 7 '16 at 10:43
  • 1
    \$\begingroup\$ Your circular buffer isn't actually a circular buffer, it's a bounded-in size queue. Since ConcurrentQueue is implemented using a circular buffer internally it happens to work but there are no guarantees it will continue to do so in the future. \$\endgroup\$ – Benjamin Gruenbaum Jul 7 '16 at 10:46
  • 3
    \$\begingroup\$ Right, but a CircularBuffer has other guarantees that are not explicitly preserved here - no dynamic memory allocations and fixed memory size. If ConcurrentQueue was implemented using a LinkedList for instance, or even as an array - these performance invariants would not be preserved. \$\endgroup\$ – Benjamin Gruenbaum Jul 7 '16 at 11:09
  • 1
    \$\begingroup\$ @Heslacher I think there is a problem in your CircularBuffer implementation. Latest will return double the length when currentStart=0 && IsFull. I've made a fiddle: dotnetfiddle.net/jnP2Lh \$\endgroup\$ – Ross Jun 25 '18 at 8:40
  • 1
    \$\begingroup\$ @Ross you are right. Edited answer. \$\endgroup\$ – Heslacher Jun 25 '18 at 8:54
3
\$\begingroup\$

You can make the usage of the ReaderWriteLockSlim a little bit nicer and get rid of the try/catch block with two disposable helpers:

public class ReaderLockSlim : IDisposable
{
    private readonly ReaderWriterLockSlim _lockSlim;

    public ReaderLockSlim(ReaderWriterLockSlim lockSlim) 
    { 
        _lockSlim = lockSlim; 
        _lockSlim.EnterReadLock();
    }

    public void Dispose()
    {
        _lockSlim.ExitReadLock();
    }
}

public class WriterLockSlim : IDisposable
{
    private readonly ReaderWriterLockSlim _lockSlim;

    public WriterLockSlim(ReaderWriterLockSlim lockSlim)
    {
        _lockSlim = lockSlim;
        _lockSlim.EnterWriteLock();
    }

    public void Dispose()
    {
        _lockSlim.ExitWriteLock();
    }
}

Example

before:

public IEnumerable<T> Latest()
{
    slimLock.EnterReadLock();
    try
    {
        return FetchItems().ToArray();
    }
    finally
    {
        slimLock.ExitReadLock();
    }
}

after:

public IEnumerable<T> Latest()
{
    using (new ReaderLockSlim(slimLock)) 
    {
        return FetchItems().ToArray();
    }
}
\$\endgroup\$
  • \$\begingroup\$ Thanks - "using" is an interesting idea for cleanly coding locks on higher level classes, where any extra overhead is unimportant. (I wouldn't use it for a "fundamental" class, such as CircularBuffer. Especially not CircularBuffer, whose purpose is to maintain a limited queue, WITHOUT any GC work.) \$\endgroup\$ – ToolmakerSteve Jan 30 '17 at 18:49
  • \$\begingroup\$ .. though to be clear, using the self-disposing ReaderLockSlim on Latest is fine, as that is a rare operation. It is Add operation that would best be done without the extra GC work of allocating/disposing a WriterLockSlim. And given that, for consistent coding style, one should stick to try/finally throughout this class. \$\endgroup\$ – ToolmakerSteve Jan 30 '17 at 18:57

protected by Jamal Jul 20 at 3:22

Thank you for your interest in this question. Because it has attracted low-quality or spam answers that had to be removed, posting an answer now requires 10 reputation on this site (the association bonus does not count).

Would you like to answer one of these unanswered questions instead?

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