A buffer that holds and provides the latest n items

Question

I have developed the following code to hold n latest items it has received, and when asked for provide these n latest items. The interface is:

public interface ICircularBuffer<T> {
        void Put(T item);  // put an item
        T[] Read(); // provides the last "n" requests
}

The idea is to build a buffer that will hold "n" latest items that have been put and return those items when asked for.

For example:

var buf = new ConcurrentBuffer<int>(3); // creates a buffer of size = 3
buf.Put(15);
buf.Put(10);
buf.Put(20);

int[] arr = buf.Read() // returns - 20,10,15

buf.Put(25);
int[] arr = buf.Read() // returns - 25,20,10

The implementation uses "locks" and array to represent the circular buffer:

public class ConcuurentCircularBuffer<T> : ICircularBuffer<T> {

        private T[] buffer;
        private int last = 0;
        private int sz;
        private object lockObject = new object();

        public ConcuurentCircularBuffer(int sz) {
            // array index starts at 1
            this.sz = sz;
            buffer = new T[sz + 1];
        }

        public void Put(T item) {
            lock (lockObject) {
                last++;
                last = last > sz ? 1 : last;
                buffer[last] = item;
            }
        }

        public T[] Read() {
            T[] arr = new T[sz];

            lock (lockObject) {
                int iterator = 0;
                for (int read = 0; read < sz; read++) {
                    int index = last - iterator;
                    index = index <= 0 ? (sz + index) : index;
                    if (buffer[index] != null) {
                        arr[iterator] = buffer[index];
                    } else {
                        break;
                    }
                    iterator++;
                }
            }
            return arr;
        }
    }

The unit tests I am using to check the performance are as follows:

[TestMethod()]
        public void TestParallelPut() {
            int sz = 10;
            ConcuurentCircularBuffer<int> buf = new ConcuurentCircularBuffer<int>(sz);

            Stopwatch watch = new Stopwatch();
            ManualResetEventSlim evt = new ManualResetEventSlim(false);

            int nTimes = 100000;

            Task[] tasks = new Task[nTimes];
            watch.Start();
            for (int i = 0; i < nTimes; i++) {
                tasks[i] = Task.Run(() => {
                    evt.Wait();
                    buf.Put(i);
                });
            }

            // signal all to proceed
            evt.Set();
            Task.WaitAll(tasks);

            watch.Stop();
            double timeMs = watch.Elapsed.TotalMilliseconds;
            TestContext.WriteLine("Buffer size: {0}", sz);
            TestContext.WriteLine("Average time per put: {0} ms, total time taken for {1} parallel requests: {2} ms",timeMs/nTimes,nTimes,timeMs);

    }

This test tries to simulate maximum contention, by trying to put all n items in parallel.

The following test, puts n items in parallel, and also continues to read in parallel:

[TestMethod()]
        public void TestParallelPutAndRead() {
            int sz = 10;
            ConcuurentCircularBuffer<int> buf = new ConcuurentCircularBuffer<int>(sz);

            Stopwatch watch = new Stopwatch();
            ManualResetEventSlim evt = new ManualResetEventSlim(false);
            ManualResetEventSlim completeEvt = new ManualResetEventSlim(false);
            int nTimes = 100000;

            Task[] tasks = new Task[nTimes];
            watch.Start();
            for (int i = 0; i < nTimes; i++) {
                tasks[i] = Task.Run(() => {
                    evt.Wait();
                    buf.Put(i);
                });
            }

            // start reading, until told to stop
            Task.Run(() => {
                for(;;) {
                    if (completeEvt.IsSet) {
                        break;
                    } else {
                        int[] arr = buf.Read();
                        Assert.IsTrue(arr.Length == sz);
                    }
                }
            });

            // signal all to proceed
            evt.Set();
            Task.WaitAll(tasks);
            completeEvt.Set();

            watch.Stop();
            double timeMs = watch.Elapsed.TotalMilliseconds;
            TestContext.WriteLine("Test ParallelPut and Read - Buffer size: {0}", sz);
            TestContext.WriteLine("Average time per put: {0} ms, total time taken for {1} parallel requests: {2} ms", timeMs / nTimes, nTimes, timeMs);

        }

The following test, start "nTasks" in parallel and each task puts nTimes/nTasks. Another tasks keeps reading it as well:

[TestMethod()]
        public void TestPutAndNParallelAndRead() {
            int sz = 10;
            ConcuurentCircularBuffer<int> buf = new ConcuurentCircularBuffer<int>(sz);

            Stopwatch watch = new Stopwatch();
            ManualResetEventSlim evt = new ManualResetEventSlim(false);
            ManualResetEventSlim completeEvt = new ManualResetEventSlim(false);
            int nTimes = 1000000;
            int nTasks = 100;

            Task[] tasks = new Task[nTasks];
            watch.Start();
            for (int i = 0; i < nTasks; i++) {
                tasks[i] = Task.Run(() => {
                    evt.Wait();
                    for (int n = 0; n < nTimes / nTasks; n++) {
                        buf.Put(i);
                    }
                });
            }

            // start reading, until told to stop
            Task.Run(() => {
                for (;;) {
                    if (completeEvt.IsSet) {
                        break;
                    } else {
                        int[] arr = buf.Read();
                        Assert.IsTrue(arr.Length == sz);
                    }
                }
            });

            // signal all to proceed
            evt.Set();
            Task.WaitAll(tasks);
            completeEvt.Set();

            watch.Stop();
            double timeMs = watch.Elapsed.TotalMilliseconds;
            TestContext.WriteLine("Test Put and {0} Parallel Put and Read - Buffer size: {1}", nTasks,sz);
            TestContext.WriteLine("Average time per put: {0} ms, total time taken for {1} requests: {2} ms", timeMs / nTimes, nTimes, timeMs);

        }

The tests results are follows on Windows 7 Enterprise with Intel Core i5-4310U CPU @ 2.00 GHz:

Test Name:  TestParallelPut 
TestContext Messages: 
Test ParallelPut - Buffer size: 10 Average time per put: 0.002393078 ms, total time taken for 100000 parallel requests: 239.3078 ms

Test Name:  TestParallelPutAndRead 
Test Outcome:   Passed 
TestContext Messages: Test ParallelPut and Read - Buffer size: 10 Average time per put: 0.002982335 ms, total time taken for 100000 parallel requests:
298.2335 ms


Test Name:  TestPutAndNParallelAndRead 
Test Outcome:   Passed 
TestContext Messages: Test Put and 100 Parallel Put and Read - Buffer size: 10 Average time per put: 0.0003211726 ms, total time taken for 1000000 requests: 321.1726 ms

Is there any alternative approach (with better performance) that can be followed to achieve the required functionality?

Please review the code and tests.

Answer 1

Naming

• There is one 'u' in ConcuurentCircularBuffer that wants to be a 'r' --> ConcurrentCircularBuffer
• There is no need to abbreviate name like size (sz --> size)
• I would call last to something more descriptive like lastIndex.

Code Style

• sz, buffer and lockObject should be read-only.
• There is no need to skip the first array element, just start with 0.
• iterator has always the same value as read. Therfore you can drop it.

Behavior

• The array returned by Read has always the same size as the buffer, independend of the actual number of items. To avoid that, consider to change the type of arr to List<T>.
• If one of the buffer's elements is null, you assume, that the end of the buffer is reached. Therefore: circularBuffer.Put(null) would clear the buffer. That is an unexpected behavior. Suggestion: Don't allow to put null values to the buffer (throw ArgumentNullException) and add a method Clear if needed.
• When using a primitive type like int for T, the Read method returns an array with the size of the buffer with the default value, even if no item was added. That is also a little bit strange... I suppose you need something like a count property that stores the number of added items.

Answer 2

In addition to @JanDotNet's answer:

You can use "less strict" lock in your code, specifically ReaderWriterLock or ReaderWriterLockSlim. As I see from your code, you can execute Read in parallel threads but cannot execute Put and Read in parallel. These classes are exactly for this case and will improve overall performance.

Answer 3

You can make your code much simpler if you use a LinkedList<T> instead of an array. It has also a much better performance when inserting and removing items in this case which are both O(1) operations:

public class ConcurrentCircularBuffer<T> : ICircularBuffer<T>
{
    private readonly LinkedList<T> _buffer;
    private int _maxItemCount;

    public ConcurrentCircularBuffer(int maxItemCount)
    {
        _maxItemCount = maxItemCount;
        _buffer = new LinkedList<T>();
    }

    public void Put(T item)
    {
        lock (_buffer)
        {
            _buffer.AddFirst(item);
            if (_buffer.Count > _maxItemCount)
            {
                _buffer.RemoveLast();
            }
        }
    }

    public IEnumerable<T> Read()
    {
        lock(_buffer) { return _buffer.ToArray(); }
    }
}

MSDN:

LinkedList.RemoveLast Method ()

LinkedList.AddFirst Method (T)

Answer 4

I just wanted to add, that for large buffers Buffer.BlockCopy (or Array.Copy or other similar methods) will give a way better performance, than element-by-element copying.

Also this:

// array index starts at 1
buffer = new T[sz + 1];

looks like an accident waiting to happen. The entire framework is built around zero-based indexes. I would not recommend having it any other way in your collections. Otherwise it is really easy to make a mistake: your program will have hundreds of arrays flying around, where most of them will be regular arrays, except for that one special array, where you have to skip first element and subtract 1 from Length before doing any operations (including LINQ queries!). At least make a custom array class, which would override Length and would throw an exception when you access 0 index. Or save yourself a trouble and just use regular array. And add -1 to index when reading form it.

Answer 5

Just scratching the surface of your code

1. Your interface ICircularBuffer<T> breaks SRP- Interface Segregration. This inteface is responsible for Putting and Reading items .

2. There is a lot of repetition in your test methods. One thing you have to avoid in pogramming is DRY . Instead of defining ConcuurentCircularBuffer<int>, Stopwatch, ManualResetEventSlim e.g

   [TestInitialize]
   public void Initialize() {
       int sz = 10;
       ConcuurentCircularBuffer<int> buf = new ConcuurentCircularBuffer<int>(sz);
       Stopwatch watch = new Stopwatch();;
       ManualResetEventSlim evt=new ManualResetEventSlim(false);;
       ManualResetEventSlim completeEvt=new ManualResetEventSlim(false);;      
   }
   

3. You could call a TestCleanup to set your objects to null e.g

   [TestCleanup()]
   public void Cleanup()
   { /* assign null to the objects */ }
   

4. Your naming convention for your test methods can be improved. Most unit testers use this convention-Naming Standard for Unit test by Roy Osherove

   [UnitOfWork_StateUnderTest_ExpectedBehavior]
   

5. A possible bug that could occur - If a negative number was passed as a parameter to the constructor e.g -1

   public ConcuurentCircularBuffer(int sz) {
       // array index starts at 1
       this.sz = sz;
       buffer = new T[sz + 1];
   }
   

   you want to check if a buffer length > 0 - then some computations should not be performed.

6. Before progressing into parallelPut and ParallelRead you should test one, two, three items are added or read to the buffer. In Nunit you can use the Assert class to check if the items returned are equal to what's stored in the buffer.