A lock-free C# implementation of single-reader multi-writer circular buffer.
// -------------------------------------------------------------------
// CircularBuffer
// https://github.com/open-telemetry/opentelemetry-dotnet/blob/master/src/OpenTelemetry/Internal/CircularBuffer.cs
// -------------------------------------------------------------------
// <copyright file="CircularBuffer.cs" company="OpenTelemetry Authors">
// Copyright The OpenTelemetry Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// </copyright>
using System;
using System.Runtime.CompilerServices;
using System.Threading;
namespace OpenTelemetry.Internal
{
/// <summary>
/// Lock-free implementation of single-reader multi-writer circular buffer.
/// </summary>
/// <typeparam name="T">The type of the underlying value.</typeparam>
internal class CircularBuffer<T>
where T : class
{
private readonly T[] trait;
private long head;
private long tail;
/// <summary>
/// Initializes a new instance of the <see cref="CircularBuffer{T}"/> class.
/// </summary>
/// <param name="capacity">The capacity of the circular buffer, must be a positive integer.</param>
public CircularBuffer(int capacity)
{
if (capacity <= 0)
{
throw new ArgumentOutOfRangeException(nameof(capacity), capacity, "capacity should be greater than zero.");
}
this.Capacity = capacity;
this.trait = new T[capacity];
}
/// <summary>
/// Gets the capacity of the <see cref="CircularBuffer{T}"/>.
/// </summary>
public int Capacity { get; }
/// <summary>
/// Gets the number of items contained in the <see cref="CircularBuffer{T}"/>.
/// </summary>
public int Count
{
get
{
var tailSnapshot = this.tail;
return (int)(this.head - tailSnapshot);
}
}
/// <summary>
/// Gets the number of items added to the <see cref="CircularBuffer{T}"/>.
/// </summary>
public long AddedCount => this.head;
/// <summary>
/// Gets the number of items removed from the <see cref="CircularBuffer{T}"/>.
/// </summary>
public long RemovedCount => this.tail;
/// <summary>
/// Adds the specified item to the buffer.
/// </summary>
/// <param name="value">The value to add.</param>
/// <returns>
/// Returns <c>true</c> if the item was added to the buffer successfully;
/// <c>false</c> if the buffer is full.
/// </returns>
public bool Add(T value)
{
if (value == null)
{
throw new ArgumentNullException(nameof(value));
}
while (true)
{
var tailSnapshot = this.tail;
var headSnapshot = this.head;
if (headSnapshot - tailSnapshot >= this.Capacity)
{
return false; // buffer is full
}
var head = Interlocked.CompareExchange(ref this.head, headSnapshot + 1, headSnapshot);
if (head != headSnapshot)
{
continue;
}
var index = (int)(head % this.Capacity);
this.trait[index] = value;
return true;
}
}
/// <summary>
/// Attempts to add the specified item to the buffer.
/// </summary>
/// <param name="value">The value to add.</param>
/// <param name="maxSpinCount">The maximum allowed spin count, when set to a negative number or zero, will spin indefinitely.</param>
/// <returns>
/// Returns <c>true</c> if the item was added to the buffer successfully;
/// <c>false</c> if the buffer is full or the spin count exceeded <paramref name="maxSpinCount"/>.
/// </returns>
public bool TryAdd(T value, int maxSpinCount)
{
if (maxSpinCount <= 0)
{
return this.Add(value);
}
if (value == null)
{
throw new ArgumentNullException(nameof(value));
}
var spinCountDown = maxSpinCount;
while (true)
{
var tailSnapshot = this.tail;
var headSnapshot = this.head;
if (headSnapshot - tailSnapshot >= this.Capacity)
{
return false; // buffer is full
}
var head = Interlocked.CompareExchange(ref this.head, headSnapshot + 1, headSnapshot);
if (head != headSnapshot)
{
if (spinCountDown-- == 0)
{
return false; // exceeded maximum spin count
}
continue;
}
var index = (int)(head % this.Capacity);
this.trait[index] = value;
return true;
}
}
/// <summary>
/// Reads an item from the <see cref="CircularBuffer{T}"/>.
/// </summary>
/// <remarks>
/// This function is not reentrant-safe, only one reader is allowed at any given time.
/// Warning: There is no bounds check in this method. Do not call unless you have verified Count > 0.
/// </remarks>
/// <returns>Item read.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public T Read()
{
var index = (int)(this.tail % this.Capacity);
while (true)
{
var value = this.trait[index];
if (value == null)
{
// If we got here it means a writer isn't done.
continue;
}
this.trait[index] = null;
this.tail++;
return value;
}
}
}
}
// -------------------------------------------------------------------
// Example Usage
// https://github.com/open-telemetry/opentelemetry-dotnet/blob/master/src/OpenTelemetry/Batch.cs
// -------------------------------------------------------------------
// <copyright file="Batch.cs" company="OpenTelemetry Authors">
// Copyright The OpenTelemetry Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// </copyright>
using System;
using System.Collections;
using System.Collections.Generic;
using System.Diagnostics;
using OpenTelemetry.Internal;
namespace OpenTelemetry
{
/// <summary>
/// Stores a batch of completed <typeparamref name="T"/> objects to be exported.
/// </summary>
/// <typeparam name="T">The type of object in the <see cref="Batch{T}"/>.</typeparam>
public readonly struct Batch<T>
where T : class
{
private readonly T item;
private readonly CircularBuffer<T> circularBuffer;
private readonly int maxSize;
internal Batch(T item)
{
this.item = item ?? throw new ArgumentNullException(nameof(item));
this.circularBuffer = null;
this.maxSize = 1;
}
internal Batch(CircularBuffer<T> circularBuffer, int maxSize)
{
Debug.Assert(maxSize > 0, $"{nameof(maxSize)} should be a positive number.");
this.item = null;
this.circularBuffer = circularBuffer ?? throw new ArgumentNullException(nameof(circularBuffer));
this.maxSize = maxSize;
}
/// <summary>
/// Returns an enumerator that iterates through the <see cref="Batch{T}"/>.
/// </summary>
/// <returns><see cref="Enumerator"/>.</returns>
public Enumerator GetEnumerator()
{
return this.circularBuffer != null
? new Enumerator(this.circularBuffer, this.maxSize)
: new Enumerator(this.item);
}
/// <summary>
/// Enumerates the elements of a <see cref="Batch{T}"/>.
/// </summary>
public struct Enumerator : IEnumerator<T>
{
private readonly CircularBuffer<T> circularBuffer;
private int count;
internal Enumerator(T item)
{
this.Current = item;
this.circularBuffer = null;
this.count = -1;
}
internal Enumerator(CircularBuffer<T> circularBuffer, int maxSize)
{
this.Current = null;
this.circularBuffer = circularBuffer;
this.count = Math.Min(maxSize, circularBuffer.Count);
}
/// <inheritdoc/>
public T Current { get; private set; }
/// <inheritdoc/>
object IEnumerator.Current => this.Current;
/// <inheritdoc/>
public void Dispose()
{
}
/// <inheritdoc/>
public bool MoveNext()
{
var circularBuffer = this.circularBuffer;
if (circularBuffer == null)
{
if (this.count >= 0)
{
this.Current = null;
return false;
}
this.count++;
return true;
}
if (this.count > 0)
{
this.Current = circularBuffer.Read();
this.count--;
return true;
}
this.Current = null;
return false;
}
/// <inheritdoc/>
public void Reset()
=> throw new NotSupportedException();
}
}
}
// -------------------------------------------------------------------
// Unit Tests
// https://github.com/open-telemetry/opentelemetry-dotnet/blob/1f04397c26d64de7d10289baf21084486f3fcd23/test/OpenTelemetry.Tests/Internal/CircularBufferTest.cs
// -------------------------------------------------------------------
// <copyright file="CircularBufferTest.cs" company="OpenTelemetry Authors">
// Copyright The OpenTelemetry Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// </copyright>
using System;
using OpenTelemetry.Internal;
using Xunit;
namespace OpenTelemetry.Tests.Internal
{
public class CircularBufferTest
{
[Fact]
public void CheckInvalidArgument()
{
Assert.Throws<ArgumentOutOfRangeException>(() => new CircularBuffer<string>(0));
}
[Fact]
public void CheckCapacity()
{
int capacity = 1;
var circularBuffer = new CircularBuffer<string>(capacity);
Assert.Equal(capacity, circularBuffer.Capacity);
}
[Fact]
public void CheckNullValueWhenAdding()
{
int capacity = 1;
var circularBuffer = new CircularBuffer<string>(capacity);
Assert.Throws<ArgumentNullException>(() => circularBuffer.Add(null));
}
[Fact]
public void CheckValueWhenAdding()
{
int capacity = 1;
var circularBuffer = new CircularBuffer<string>(capacity);
var result = circularBuffer.Add("a");
Assert.True(result);
Assert.Equal(1, circularBuffer.AddedCount);
Assert.Equal(1, circularBuffer.Count);
}
[Fact]
public void CheckBufferFull()
{
int capacity = 1;
var circularBuffer = new CircularBuffer<string>(capacity);
var result = circularBuffer.Add("a");
Assert.True(result);
Assert.Equal(1, circularBuffer.AddedCount);
Assert.Equal(1, circularBuffer.Count);
result = circularBuffer.Add("b");
Assert.False(result);
Assert.Equal(1, circularBuffer.AddedCount);
Assert.Equal(1, circularBuffer.Count);
}
[Fact]
public void CheckRead()
{
string value = "a";
int capacity = 1;
var circularBuffer = new CircularBuffer<string>(capacity);
var result = circularBuffer.Add(value);
Assert.True(result);
Assert.Equal(1, circularBuffer.AddedCount);
Assert.Equal(1, circularBuffer.Count);
string read = circularBuffer.Read();
Assert.Equal(value, read);
Assert.Equal(1, circularBuffer.AddedCount);
Assert.Equal(1, circularBuffer.RemovedCount);
Assert.Equal(0, circularBuffer.Count);
}
[Fact]
public void CheckAddedCountAndCount()
{
int capacity = 2;
var circularBuffer = new CircularBuffer<string>(capacity);
var result = circularBuffer.Add("a");
Assert.True(result);
Assert.Equal(1, circularBuffer.AddedCount);
Assert.Equal(1, circularBuffer.Count);
result = circularBuffer.Add("a");
Assert.True(result);
Assert.Equal(2, circularBuffer.AddedCount);
Assert.Equal(2, circularBuffer.Count);
_ = circularBuffer.Read();
Assert.Equal(2, circularBuffer.AddedCount);
Assert.Equal(1, circularBuffer.RemovedCount);
Assert.Equal(1, circularBuffer.Count);
}
}
}