Thread-safe fixed-size circular buffer with sequence ids

Question

I need a queue with these capabilities:

• fixed-size (i.e. circular buffer)
• queue items have ids (like a primary key), which are sequential
• thread-safe (used from multiple ASP.NET Core requests)

To avoid locking, I tried a ConcurrentQueue but found race conditions. So I'm trying a custom approach.

public interface IQueueItem
{
    long Id { get; set; }
}

public class CircularBuffer<T> : LinkedList<T> where T : class, IQueueItem
{
    public CircularBuffer(int capacity) => _capacity = capacity;
    private readonly int _capacity;

    private long _counter = 0;
    private readonly object _lock = new();

    public void Enqueue(T item)
    {
        lock (_lock) {         // works but feels "heavy"
            _counter++;
            item.Id = _counter;
            if (Count == _capacity) RemoveFirst();
            AddLast(item);
        }
    }
}

And to test:

public class Item : IQueueItem
{
    public long Id { get; set; }
    //...
}

public class Program
{
    public static void Main()
    {
        var q = new CircularBuffer<Item>(10);
        Parallel.For(0, 15, i => q.Enqueue(new Item()));
        Console.WriteLine(string.Join(", ", q.Select(x => x.Id)));
    }
}

Which gives correct output (is ordered even though enqueued by competing threads, and has fixed size with oldest items dequeued):

6, 7, 8, 9, 10, 11, 12, 13, 14, 15

In reality, I have web requests that read (i.e. enumerate) that queue.

The problem: if one thread is enumerating the queue while another thread is adding to it, I will have errors. (I could use a ToList() before the read, but for a large queue that will suck up all the server's memory as this could be done many times a second by multiple requests). How can I deal with that scenario? I used a linked list, but I'm flexible to use any structure.

(Also, that seems to be a really heavy lock section; is there a more performant way?)

UPDATE
As asked in comments below: I expect the queue to have from a few hundred to a few tens of thousand items, but the items themselves are small (just a few primitive data types). I expect an enqueue every second. Reads from web requests are less often, let's say a few times per minute (but can occur concurrently to the server writing to the queue).

Answer 1

Based on the metrics that you provided in the question, you have plenty of options. The anticipated usage of the CircularBuffer<T> is not really that heavy. Wrapping a lock-protected Queue<T> should work pretty well. The cost of copying the contents of the queue into an array on each enumeration (copying 10,000 elements a few times per minute) is unlikely to be noticeable. Modern machines can do such things in the blink of an eye. You'd have to enumerate the collection hundreds of times per second for this to start (slightly) becoming an issue.

In my original answer (revision 3) I had proposed using an ImmutableQueue<T> as underlying storage. After closer inspection I noticed that this class is not exactly pay-for-play. The first time it is enumerated it invokes the internal BackwardsReversed property (source code), which is quite costly. My performance tests confirmed that it's a worse solution than the simple lock-protected Queue<T> that is shown in lonix's answer, regarding both CPU-time and allocations.

Below is a lower-level implementation of a similar idea, that exploits the fact that we need only a subset of the functionality of the ImmutableQueue<T> class. The items are stored in a singly linked-list structure, that can be enumerated without additional costs:

public class ConcurrentCircularBuffer<T> : IEnumerable<T> where T : IQueueItem
{
    private readonly object _locker = new();
    private readonly int _capacity;
    private Node _head;
    private Node _tail;
    private int _count = 0;
    private long _lastId = 0;

    private class Node
    {
        public readonly T Item;
        public Node Next;
        public Node(T item) => Item = item;
    }

    public ConcurrentCircularBuffer(int capacity)
    {
        if (capacity < 1) throw new ArgumentOutOfRangeException(nameof(capacity));
        _capacity = capacity;
    }

    public int Count => Volatile.Read(ref _count);

    public void Enqueue(T item)
    {
        lock (_locker)
        {
            Node node = new(item);
            if (_head is null) _head = node;
            if (_tail is not null) _tail.Next = node;
            _tail = node;
            if (_count < _capacity) _count++; else _head = _head.Next;
            item.Id = ++_lastId;
        }
    }

    public IEnumerator<T> GetEnumerator()
    {
        Node node; int count;
        lock (_locker) { node = _head; count = _count; }
        for (int i = 0; i < count && node is not null; i++, node = node.Next)
            yield return node.Item;
    }

    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
}

The main advantage of this approach over the lock-protected Queue<T> is that it minimizes the contention. The work done while holding the lock is minuscule.

An alternative implementation of the ConcurrentCircularBuffer<T> class, based on two array buffers and having different pros and cons, can be found in the 5th revision of this answer.

Answer 2

Since ConcurrentQueue is out in this question, you can try fixed array.

IQueueItem[] items = new IQueueItem[SIZE];
long id = 0;

Enqueue is simple.

void Enqueue(IQueueItem item)
{
    long id2 = Interlocked.Increment(ref id);
    item.Id = id2 - 1;
    items[id2 % SIZE] = item;
}

To output the data, you just need copy the array to a new one, then sort it. (of course, it can be optimized here)

var arr = new IQueueItem[SIZE];
Array.Copy(items, arr, SIZE);
return arr.Where(a => a != null).OrderBy(a => a.Id);

There may be some gaps in the array because of the concurrent insertions, you can take a sequence till a gap is found.

var e = arr.Where(a => a != null).OrderBy(a => a.Id);
var firstId = e.First().Id;
return e.TakeWhile((a, index) => a.Id - index == firstId);

Answer 3

Here is another implementation, using a Queue<T> with locking.

public interface IQueueItem
{
    long Id { get; set; }
}

public class CircularBuffer<T> : IEnumerable<T> where T : class, IQueueItem
{
    private readonly int _capacity;
    private readonly Queue<T> _queue;
    private long _lastId = 0;
    private readonly object _lock = new();

    public CircularBuffer(int capacity) {
        _capacity = capacity;
        _queue = new Queue<T>(capacity);
    }

    public void Enqueue(T item)
    {
        lock (_lock) {
            if (_capacity < _queue.Count)
                _queue.Dequeue();
            item.Id = ++_lastId;
            _queue.Enqueue(item);
        }
    }

    public IEnumerator<T> GetEnumerator()
    {
        lock (_lock) {
            var copy = _queue.ToArray();
            return ((IEnumerable<T>)copy).GetEnumerator();
        }
    }

    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();

}

And to test:

public class Item : IQueueItem
{
    private long _id;

    public long Id
    {
        get => Volatile.Read(ref _id);
        set => Volatile.Write(ref _id, value);
    }
}

public class Program
{
    public static void Main()
    {
        var q = new CircularBuffer<Item>(10);
        Parallel.For(0, 15, i => q.Enqueue(new Item()));
        Console.WriteLine(string.Join(", ", q.Select(x => x.Id)));
    }
}

Result:

6, 7, 8, 9, 10, 11, 12, 13, 14, 15