Asynchronous Circular Buffer in C#

Question

Preamble

I want to be able to test methods which communicate over a simple exclusively asynchronous two-way stream-like interface which may underneath use any of a number of communication methods (e.g. sockets, named pipe. In order to test these, I need a memory-backed implementation, where I can read and write stuff leisurely. In order to implement this, I need a one-way memory-backed channel.

The main purpose is to facilitate throwing chunks of bytes around, but I decided it would be fun and possibly useful to make it generic. Correctness is very important, since I'll be using this type to test other types. The motivation for the implementation is that (for a sufficiently large buffer) everything should be nice and fast, with lots of synchronous writes, but that it will cope with the reader falling behind without trying to guzzle memory.

I can't help but think this nightmarish mess might be over complicated, so I thought I'd ask here. In my defence, it does pass all its tests, and most of the code is just devoted to throwing exceptions (my favourite kind of code).

Specification

A class implementing a circular buffer which provides asynchronous read and write capabilities for any generic type such that:

• Can read or write an arbitrarily large amount of data
• Reads and Writes never fail (excepting manual cancellation or disposal)
• A Read or Write only finishes once all data has been read or written
  • Reads will wait for data if it is yet to be written
  • Writes will wait for space in the buffer to be freed by reads
• Uses an essentially constant (predictable) amount of memory
• The instance may be disposed
• Reads may be cancelled: doing so must dispose the instance
• Reads and Writes must not block on a disposed instance
• Must support clearing read data from the buffer (e.g. so that references are not held)

Goals:

• Simple API
• Well tested and Documented
• Allow arbitrary backing memory
• Attempt to detect 'concurrent' Read/Write

Non-goals:

• Thread safe simultaneous reads and writes (this doesn't make sense at this layer)
  • Should try to throw if this is attempted
• Ludicrous performance
  • Speed is a concern, because it will be hammered by a bunch of tests (probably run in parallel...), but so long as it can pass 1GB around in under a minute on my old i7 I'm sure I'll cope
  • Not eating all the memory on my machine is more important
  • Any optimisation should be in favour of the non-waiting case

Notes:

• ValueTask is used to be consistent with parts of an external API. I don't think I gain anything by using ValueTask here, but I'm not too worried by performance, and some informal benchmarking suggests it doesn't make much difference. I'd be interested on comments, but I probably won't be changing this.

• The EagerSignallingThreshold is used to avoid the situation where, say, a Read wants 1000 bytes, but there are only 900 available: if the amount available is large relative to the total buffer size, then we signal the reader that it might as well get started, which hopefully maximises CPU utilisation, and means that the space freed by the reader will be available sooner to be written. Informal testing suggests this doesn't make much difference, but the capability is there if I find a situation which might be useful in the future.

• Three different concurrency methods are used for three different things:

  • A lock (on AvailabilityLock) is used to ensure correctness of ReadAvailable, ReadRequest, WriteAvailable, and WriteRequest: these are the fields used to communicate between the Read and Write mechanisms.
  • Semaphores to allow readers to wait for something to be written, and writers to wait for space to become available.
  • CAS to perform mutual exclusion in DisposeInternal, ReadAsync, and WriteAsync.

• The Disposed event is provided so that, for example, a wrapper which manages simultaneous reads and writes has a dedicated way of detecting a failure. It also means there is a direct mechanism to allow releasing of unmanaged backing memory.

• I have tried to break the ReadAsync and WriteAsync methods down into meaningful parts. This has the downside of making everything a little more confusing, but the upside of enforcing structure, and does mean the code is slightly more flexible. It is still the case, for example, that CompleteRead must be called after PerformRead, and that you must release the Writers is CompleteRead tells you to (because it has already started that process), but atleast I can't get excited and shuffle the code between them, and if I feel like snapping PerformRead in two (so that someone else can do the 'overrun' check), that will be easier.

• I have no intention of doing anything along the lines of realising that I have a writer sitting around waiting to write, and reader waiting to read lots, and passing data directly from the writer to the reader. That could be a lot of fun as well, but it is not the idea here (I don't want writers blocking longer than they have to within the memory constraints, and I don't want the additional complexity of detecting and performing this efficiently here).

Description of operation

The Read and Write mechanism are almost completely symmetric, and have 'symmetric' properties to go with them. (I have considered coercing them into a single set of methods, but that would probably only create more confusion.) Reading has two added complexities, so I will describe only the Read mechanism here:

A call to ReadAsync starts by ensuring we haven't been disposed, and that a read is not already in progress (both are cause for an exception), setting ReadState = ResourceInUse in the process.

If this does not fail, it determines the requested number of values. It proceeds in a loop attempting to read as much memory as it can at that particular moment. It does so by requesting as much as could possibly copy (i.e. the size of the backing memory), which involves the call to RequestRead and associated awaiting of the ReadAvailableSemaphore.

If there are any values available (ReadAvailable > 0), RequestRead returns true, and required is clamped to this value, before performing the read itself and remembering how much we have read so far before we start the loop again. We make no attempt to wait for 'more' values to become available (e.g. with a timeout) if there are any available in that moment.

If there were no values available, then RequestRead returns false, and we await the ReadAvailableSemaphore, which will be signaled by the write mechanism when there are values available again. Reads can be cancelled (writes cannot): if the read is cancelled, we dispose ourselves and throw. Once the semaphore is released, we again decide how many values are available, and proceed to perform the actual read on the next slice of the buffer.

PerformRead starts by again checking we have not been disposed, caches a copy of the MemoryBuffer, and performs some internal checks to ensure we are not trying to do something stupid.

Next, it decides whether the read needs to be cut into two because it extends beyond the buffer, and copies lots of memory about. At this point, if ClearOnRead is set, it will fill the buffer with default values. Finally it updates ReadPosition.

CompleteRead then locks AvailabilityLock. Within the lock it updates ReadAvailable and WriteAvailable, and decides whether it should signal a waiting, clearing WriteRequest if it decides that it will, and returning true (otherwise false).

The loop in ReadAsync concludes by signalling the reader if CompleteRead instructed it to, and tallying how much is left to be read to the target buffer.

When the whole buffer has been filled, the ReadState is reset so that ReadAsync can be called again.

Code

Example Usage

using System;
using System.Threading.Tasks;
using Comm;

namespace CommExamples
{
    static class Program
    {
        static async Task Main(string[] args)
        {
            using (CommMemory<byte> cm = new CommMemory<byte>(capacity: 1024, clearOnRead: false))
            {
                var write = WriteString(cm, "Hello, miserable world.");
                var read = ReadString(cm);

                Console.WriteLine(await read);
                await write;
            }

            Console.ReadKey(true);
        }

        /// <summary>
        /// Writes a length-prefix (Pascal) string to the given <see cref="CommMemory{byte}"/>.
        /// </summary>
        /// <param name="cm">The <see cref="CommMemory{byte}"/> to which to write.</param>
        /// <param name="str">The <see cref="string"/> to write to the <see cref="CommMemory{byte}"/></param>
        private static async ValueTask WriteString(CommMemory<byte> cm, string str)
        {
            // NOTE: this is example code: use a reusable string writer for performance
            var stringBuffer = System.Text.UTF8Encoding.UTF8.GetBytes(str);
            var intBuffer = System.BitConverter.GetBytes(stringBuffer.Length);

            await cm.WriteAsync(intBuffer);
            await cm.WriteAsync(stringBuffer);
        }

        /// <summary>
        /// Reads a length-prefixed (Pascal) string from the given <see cref="CommMemory{byte}"/>.
        /// </summary>
        /// <param name="cm">The <see cref="CommMemory{byte}"/> to which to write.</param>
        /// <returns>The string read from the given <see cref="CommMemory{byte}"/>.</returns>
        private static async ValueTask<string> ReadString(CommMemory<byte> cm)
        {
            // NOTE: this is example code: use a reusable string reader for performance
            var intBuffer = new byte[4];
            await cm.ReadAsync(intBuffer);

            var length = System.BitConverter.ToInt32(intBuffer);
            var stringBuffer = new byte[length];
            await cm.ReadAsync(stringBuffer);

            return System.Text.UTF8Encoding.UTF8.GetString(stringBuffer);
        }
    }
}

CommMemory.cs (Implementation)

using System;
using System.Threading;
using System.Threading.Tasks;

namespace Comm
{
    /// <summary>
    /// Exception indicating that the <see cref="CommMemory{TValue}"/> instance consulted was Disposed and can no longer be used.
    /// </summary>
    public class CommMemoryDisposedException : Exception
    {
        public CommMemoryDisposedException(string message) : base(message)
        {
            // nix
        }

        public CommMemoryDisposedException(string message, Exception innerException) : base(message, innerException)
        {
            // nix
        }
    }

    /// <summary>
    /// Callback for when a <see cref="CommMemory{TValue}"/> instance is disposed.
    /// </summary>
    /// <param name="disposedInstance">The <see cref="CommMemory{TValue}"/> instance disposed</param>
    /// <typeparam name="TValue">The type of the values handled by the <see cref="CommMemory{TValue}"/>.</typeparam>
    public delegate void CommMemoryDisposed<TValue>(CommMemory<TValue> disposedInstance);

    /// <summary>
    /// Provides Asyncronous Read/Write into a cyclic memory buffer.
    /// <see cref="ReadAsync(ArraySegment{TValue}, CancellationToken)"/> and <see cref="WriteAsync(ArraySegment{TValue})"/> Tasks only complete when all data has been read,
    /// or the <see cref="CommMemory{TValue}"/> is Disposed by successfully cancelling a call to <see cref="ReadAsync(ArraySegment{TValue}, CancellationToken)"/> or calling <see cref="Dispose"/>.
    /// Thread-safe for single simulataneous Read and Write.
    /// Not thread-safe for multiple simulataneous Reads or Writes.
    /// </summary>
    /// <typeparam name="TValue">The type of Data handled by the <see cref="CommMemory{TValue}"/>.</typeparam>
    public class CommMemory<TValue> : IDisposable
    {
        /// <summary>
        /// Checks the capacity is valid, and returns it.
        /// Throws if the capacity is not valid.
        /// </summary>
        private static int CheckCapacity(int capacity)
        {
            if (capacity <= 0)
                throw new ArgumentOutOfRangeException("Capacity must be non-negative (greater than 0)");

            return capacity;
        }

        /// <summary>
        /// Creates a <see cref="CommMemory{TValue}"/> instance backed by an Array with the given capacity.
        /// </summary>
        /// <param name="capacity">The capcity of the <see cref="CommMemory{TValue}"/> instance; the length of the Array created.</param>
        public CommMemory(int capacity, bool clearOnRead) : this(new Memory<TValue>(new TValue[CheckCapacity(capacity)]), clearOnRead)
        {
            // nix
        }

        public CommMemory(Memory<TValue> buffer, bool clearOnRead)
        {
            if (buffer.Length == 0)
                throw new ArgumentException("Cannot use a buffer of length 0 for CommMemory");

            MemoryBuffer = buffer;
            ClearOnRead = clearOnRead;

            Capacity = MemoryBuffer.Length;

            ReadPosition = 0;
            ReadAvailable = 0;

            WritePosition = 0;
            WriteAvailable = Capacity;

            EagerSignallingThreshold = Capacity / 2;
        }

        /// <summary>
        /// The threshold, beyond which to signal early even if the requested number of values is not available.
        /// Must be no more than Capacity.
        /// </summary>
        private readonly int EagerSignallingThreshold;

        /// <summary>
        /// The index from which to read.
        /// May only be modifier by a reader.
        /// </summary>
        private int ReadPosition;

        /// <summary>
        /// The number of values available to be read.
        /// May only be increased by a Writer, and while AvailabilityLock is held.
        /// May only be decreased by Reader, and while AvailabilityLock is held.
        /// </summary>
        private int ReadAvailable;

        /// <summary>
        /// The index to which to write.
        /// May only be modified by a writer.
        /// </summary>
        private int WritePosition;

        /// <summary>
        /// The number of values free to be written.
        /// May only be increased by Reader, and while AvailabilityLock is held.
        /// May only be decreased by a Writer, and while AvailabilityLock is held.
        /// </summary>
        private int WriteAvailable;

        /// <summary>
        /// The Memory used as the cyclic buffer.
        /// </summary>
        private Memory<TValue> MemoryBuffer;

        /// <summary>
        /// The total capacity of this CommMemory object when it is alive.
        /// </summary>
        public readonly int Capacity;

        /// <summary>
        /// Availablility lock to be held whenever Request or Available values are modified.
        /// Any time a decision is made based on Available, Request must be set within the same lock.
        /// </summary>
        private readonly object AvailabilityLock = new object();

        /// <summary>
        /// The number of values waiting to be read.
        /// Positive only if a reader is going to wait ReadAvailableSemaphore
        /// (must set ReadRequest before waiting on ReadAvailableSemaphore).
        /// Can only be set positive by a reader, and while AvailabilityLock is held.
        /// Can only be set to 0 by a writer, and while AvailabilityLock is held.
        /// </summary>
        private int ReadRequest = 0;

        /// <summary>
        /// The number of values waiting to be written.
        /// Positive only if a writer is going to wait WriteAvailableSemaphore
        /// (must set WriteRequest before waiting on WriteAvailableSemaphore)
        /// Can only be set positive by a writer, and while AvailabilityLock is held.
        /// Can only be set to 0 by a reader, and while AvailabilityLock is held.
        /// </summary>
        private int WriteRequest = 0;

        /// <summary>
        /// Signals when a Reader should try reading again because ReadAvailable has increased.
        /// </summary>
        private readonly SemaphoreSlim ReadAvailableSemaphore = new SemaphoreSlim(0);

        /// <summary>
        /// Signals when a Writer should try writing again because WriterAvailable has increased.
        /// </summary>
        private readonly SemaphoreSlim WriteAvailableSemaphore = new SemaphoreSlim(0);

        /// <summary>
        /// Indicidates whether this <see cref="CommMemory{TValue}"/> instance has become invalid.
        /// </summary>
        public bool IsDisposed => State == DisposedState || MemoryBuffer.Length == 0;

        /// <summary>
        /// Indicates whether this <see cref="CommMemory{TValue}"/> clears values that have been read.
        /// </summary>
        public bool ClearOnRead { get; }

        /// <summary>
        /// Throws if the CommMemory instance is dead.
        /// </summary>
        private void EnsureAlive()
        {
            if (IsDisposed)
                throw new CommMemoryDisposedException("The CommMemory instance is disposed.");
        }

        /// <summary>
        /// Returns a useable copy of the MemoryBuffer.
        /// Throws if the instance is dead.
        /// </summary>
        private Memory<TValue> EnsureAliveAcquire()
        {
            var memory = MemoryBuffer;

            // must check our own copy of memory
            if (State == DisposedState || memory.Length == 0)
                throw new CommMemoryDisposedException("The CommMemory instance is disposed.");

            return memory;
        }

        /// <summary>
        /// Called exactly once when the <see cref="CommMemory{TValue}"/> instance is disposed.
        /// Will only be called once.
        /// </summary>
        public event CommMemoryDisposed<TValue> Disposed;

        /// <summary>
        /// Indicates the <see cref="CommMemory{TValue}"/> is alive.
        /// </summary>
        private const int AliveState = 0;

        /// <summary>
        /// Indicates the <see cref="CommMemory{TValue}"/> is disposed,
        /// or in the process of being Disposed.
        /// </summary>
        private const int DisposedState = 1;

        /// <summary>
        /// The state of the <see cref="CommMemory{TValue}"/> instance.
        /// Provides a Mutex for <see cref="DisposeInternal"/> ensuring it is only run once.
        /// </summary>
        private int State = AliveState;

        /// <summary>
        /// Indicates that a given resource is free.
        /// </summary>
        private const int ResourceFree = 0;

        /// <summary>
        /// Indicates that a given resource is in use.
        /// </summary>
        private const int ResourceInUse = 1;

        /// <summary>
        /// Mutex for <see cref="ReadAsync(Memory{TValue}, CancellationToken)"/>.
        /// Has value <see cref="ResourceFree"/> when no read is pending.
        /// Takes value <see cref="ResourceInUse"/> when a read is pending.
        /// </summary>
        private int ReadState = ResourceFree;

        /// <summary>
        /// Mutex for <see cref="WriteAsync(ReadOnlyMemory{TValue})"/>.
        /// Has value <see cref="ResourceFree"/> when no write is pending.
        /// Takes value <see cref="ResourceInUse"/> when a write is pending.
        /// </summary>
        private int WriteState = ResourceFree;

        /// <summary>
        /// Kills the instance.
        /// 'Clears' the MemoryBuffer.
        /// Releasing (and disposes) all semaphors.
        /// Invokes <see cref="Disposed"/>.
        /// </summary>
        private void DisposeInternal()
        {
            // only dispose once; tolerate multiple calls
            if (System.Threading.Interlocked.CompareExchange(ref State, DisposedState, AliveState) != AliveState)
                return;

            MemoryBuffer = Memory<TValue>.Empty;

            ReadAvailableSemaphore.Release();
            ReadAvailableSemaphore.Dispose();
            WriteAvailableSemaphore.Release();
            WriteAvailableSemaphore.Dispose();

            Disposed?.Invoke(this);
        }

        /// <summary>
        /// Asynchronously writes all the values from the given source <see cref="ReadOnlyMemory{T}"/>.
        /// Only one write may be in progress at a time.
        /// </summary>
        /// <param name="buffer">The buffer containing the values to be written.</param>
        public async ValueTask WriteAsync(ReadOnlyMemory<TValue> source)
        {
            EnsureAlive();

            if (System.Threading.Interlocked.CompareExchange(ref WriteState, ResourceInUse, ResourceFree) != ResourceFree)
            {
                // double-check we are still alive: if we have been disposed, then we are probably just an unreleased Mutex
                EnsureAlive();
                throw new InvalidOperationException("A Write operation is still pending");
            }

            int requested = source.Length;
            int writtenSoFar = 0;

            while (requested > 0)
            {
                // only try to write as much as could fit in memory at a time
                int required = Math.Min(requested, Capacity);

                // check if values are available
                if (!RequestWrite(required))
                {
                    // wait for requested values to become available
                    await WriteAvailableSemaphore.WaitAsync();
                    EnsureAlive();
                }

                // take as much as we can
                required = Math.Min(required, WriteAvailable);

                // moves data around
                PerformWrite(source.Slice(writtenSoFar, required));

                // communicate state to Readers
                if (CompleteWrite(required))
                {
                    ReadAvailableSemaphore.Release();
                }

                writtenSoFar += required;
                requested -= required;
            }

            // we only clear WriteState on a successful write
            WriteState = ResourceFree;
        }

        /// <summary>
        /// Returns true if there are values available for writing;
        /// otherwise, sets <see cref="WriteRequest"/> and returns false.
        /// You must await <see cref="WriteAvailableSemaphore"/> if this method returns false.
        /// </summary>
        private bool RequestWrite(int requested)
        {
            // preliminary check to avoid unnecessary locking
            if (WriteAvailable == 0)
            {
                // check we still need to wait, and set WriteRequested if need to wait
                lock (AvailabilityLock)
                {
                    if (WriteAvailable == 0)
                    {
                        WriteRequest = requested;
                        return false;
                    }
                }
            }

            return true;
        }

        /// <summary>
        /// Writes the given source <see cref="ReadOnlyMemory{T}"/> to the MemoryBuffer at the WritePosition, updating the WritePosition.
        /// </summary>
        private void PerformWrite(ReadOnlyMemory<TValue> source)
        {
            // acquire a copy of MemoryBuffer so that it can't change beneath us
            var memory = EnsureAliveAcquire();

            int required = source.Length;

            // check we have enough space to actually perform the write (should be verified by callers)
            if (required > WriteAvailable)
            {
                DisposeInternal();
                throw new Exception("Internal error in CommMemory: attempted to perform a write with insufficient space available.");
            }

            if (required == 0)
            {
                DisposeInternal();
                throw new Exception("Internal error in CommMemory: attempted to write into a buffer of length 0.");
            }

            // determine whether we need to cut the write in two
            if (WritePosition + required > Capacity)
            {
                int headLength = Capacity - WritePosition; // length of the first write
                int tailLength = required - headLength; // length of the second write

                var s0 = source.Slice(0, headLength);
                var m0 = memory.Slice(WritePosition, headLength);
                s0.CopyTo(m0);

                var s1 = source.Slice(headLength, tailLength);
                var m1 = memory.Slice(0, tailLength);
                s1.CopyTo(m1);

                WritePosition = tailLength;
            }
            else
            {
                Memory<TValue> m = memory.Slice(WritePosition, required);
                source.CopyTo(m);

                WritePosition += required;

                if (WritePosition == Capacity)
                    WritePosition = 0;
            }
        }

        /// <summary>
        /// Finishes the write by transfering the given space from WriteAvailable to ReadAvailable.
        /// Returns true if there is a waiting reader that should be signalled; clears ReadRequest accordingly.
        /// </summary>
        /// <param name="consumed">The number of values consumed</param>
        private bool CompleteWrite(int consumed)
        {
            lock (AvailabilityLock)
            {
                // update availables
                WriteAvailable -= consumed;
                ReadAvailable += consumed;

                // check if a reader is waiting
                if (ReadRequest > 0)
                {
                    // signal reader if there are enough values availale to be read, and clear the ReadRequest
                    if (ReadAvailable >= ReadRequest || ReadAvailable >= EagerSignallingThreshold)
                    {
                        ReadRequest = 0;
                        return true;
                    }
                }
            }

            return false;
        }

        /// <summary>
        /// Asynchronously reads values into the given destination <see cref="Memory{T}"/>.
        /// Cancelling a read will cause the CommMemory instance to be Disposed, and any pending write to fail.
        /// Only one read may be in progress at a time.
        /// </summary>
        /// <param name="ct">A cancellation token.</param>
        public async ValueTask ReadAsync(Memory<TValue> destination, CancellationToken ct = default)
        {
            EnsureAlive();

            if (System.Threading.Interlocked.CompareExchange(ref ReadState, ResourceInUse, ResourceFree) != ResourceFree)
            {
                // double-check we are still alive: if we have been disposed, then we are probably just an unreleased Mutex
                EnsureAlive();
                throw new InvalidOperationException("A Read operation is still pending");
            }

            int requested = destination.Length;
            int readSoFar = 0;

            while (requested > 0)
            {
                // only try to write as much as could fit in memory at a time
                int required = Math.Min(requested, Capacity);

                if (!RequestRead(required))
                {
                    try
                    {
                        // wait for requested values to become available
                        await ReadAvailableSemaphore.WaitAsync(ct);
                    }
                    catch (OperationCanceledException ocex)
                    {
                        // task cancelled: invalidate everything
                        DisposeInternal();
                        throw new CommMemoryDisposedException("Invalided by Read cancellation.", ocex);
                    }
                }

                // take as much as we can
                required = Math.Min(required, ReadAvailable);

                PerformRead(destination.Slice(readSoFar, required));

                if (CompleteRead(required))
                {
                    WriteAvailableSemaphore.Release();
                }

                readSoFar += required;
                requested -= required;
            }

            // Only clear ReadState on a successful read
            ReadState = ResourceFree;
        }

        /// <summary>
        /// Returns true if there are values available for reading;
        /// otherwise, sets <see cref="ReadRequest"/> and returns false.
        /// You must await <see cref="ReadAvailableSemaphore"/> if this method returns false.
        /// </summary>
        private bool RequestRead(int requested)
        {
            // preliminary check to avoid unnecessary locking
            if (ReadAvailable == 0)
            {
                // check we still need to wait, and set ReadRequested if need to wait
                lock (AvailabilityLock)
                {
                    if (ReadAvailable == 0)
                    {
                        ReadRequest = requested;
                        return false;
                    }
                }
            }

            return true;
        }

        /// <summary>
        /// Reads to the given destination <see cref="Memory{T}"/> from the MemoryBuffer at the ReadPosition, updating the ReadPosition.
        /// </summary>
        private void PerformRead(Memory<TValue> destination)
        {
            // acquire a copy of MemoryBuffer so that it can't change beneath us
            var memory = EnsureAliveAcquire();

            int required = destination.Length;

            // check we have enough space to actually perform the read
            if (required > ReadAvailable)
            {
                DisposeInternal();
                throw new Exception("Internal error in CommMemory: attempted to perform a read with insufficient space available.");
            }

            if (required == 0)
            {
                DisposeInternal();
                throw new Exception("Internal error in CommMemory: attempted to read into a buffer of length 0.");
            }

            // determine whether we need to cut the read in two
            if (ReadPosition + required > Capacity)
            {
                int headLength = Capacity - ReadPosition; // length of the first read
                int tailLength = required - headLength; // length of the second read

                var d0 = destination.Slice(0, headLength);
                var m0 = memory.Slice(ReadPosition, headLength);
                m0.CopyTo(d0);

                var d1 = destination.Slice(headLength, tailLength);
                var m1 = memory.Slice(0, tailLength);
                m1.CopyTo(d1);

                if (ClearOnRead)
                {
                    m0.Span.Fill(default(TValue));
                    m1.Span.Fill(default(TValue));
                }

                ReadPosition = tailLength;
            }
            else
            {
                var m = memory.Slice(ReadPosition, required);
                m.CopyTo(destination);

                if (ClearOnRead)
                {
                    m.Span.Fill(default(TValue));
                }

                ReadPosition += required;

                if (ReadPosition == Capacity)
                    ReadPosition = 0;
            }
        }

        /// <summary>
        /// Finishes the read by transfering the given space from ReadAvailable to WriteAvailable.
        /// Returns true if there is a waiting writer that must be signalled; clears WriteRequest accordingly.
        /// </summary>
        /// <param name="consumed">The number of values consumed</param>
        private bool CompleteRead(int consumed)
        {
            lock (AvailabilityLock)
            {
                // update availables
                ReadAvailable -= consumed;
                WriteAvailable += consumed;

                // check if a write is waiting
                if (WriteRequest > 0)
                {
                    // signal write if there is enough space, and clear the WriteRequest
                    if (WriteAvailable >= WriteRequest || WriteAvailable >= EagerSignallingThreshold)
                    {
                        WriteRequest = 0;
                        return true;
                    }
                }
            }

            return false;
        }

        /// <summary>
        /// Disposes the <see cref="CommMemory{TValue}"/> instances.
        /// Terminating any pending Read or Write.
        /// </summary>
        public void Dispose()
        {
            DisposeInternal();
        }
    }
}

CommMemoryTests.cs (Tests)

using Microsoft.VisualStudio.TestTools.UnitTesting;
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Comm.Tests
{
    /// <summary>
    /// Wraps a <see cref="CommMemory{byte}"/> for Read/Write testing.
    /// </summary>
    public struct ReadWriteCommMemoryByteWrapper : IAsyncByteReader, IAsyncByteWriter
    {
        public ReadWriteCommMemoryByteWrapper(CommMemory<byte> commMemory)
        {
            CommMemory = commMemory ?? throw new ArgumentNullException(nameof(commMemory));
        }

        public CommMemory<byte> CommMemory { get; }

        public ValueTask ReadAsync(ArraySegment<byte> buffer, CancellationToken ct)
        {
            return CommMemory.ReadAsync(buffer, ct);
        }

        public ValueTask WriteAsync(ArraySegment<byte> buffer)
        {
            return CommMemory.WriteAsync(buffer);
        }
    }

    [TestClass]
    public class CommMemoryTests
    {
        /// <summary>
        /// Check that repeated calls to <see cref="CommMemory{TValue}.Dispose"/> are tolerated,
        /// and that <see cref="CommMemory{TValue}.IsDisposed"/> is only called once.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(ArgumentOutOfRangeException))]
        public void TolerateRepeatedDisposalTest()
        {
            int count = 0;
            using CommMemory<byte> cm = new CommMemory<byte>(0, false);
            cm.Disposed += _cm => count++;

            for (int i = 0; i < 10; i++)
                cm.Dispose();

            Assert.AreEqual(1, count, "Disposal should only run once");
        }

        /// <summary>
        /// Check that a capcity of zero is reject with an <see cref="ArgumentOutOfRangeException"/>.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(ArgumentOutOfRangeException))]
        public void InvalidBufferSize0Test()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(0, false);
        }

        /// <summary>
        /// Check that a negative capcity is reject with an <see cref="ArgumentOutOfRangeException"/>.
        /// </summary>
        /// <summary>
        [TestMethod]
        [ExpectedException(typeof(ArgumentOutOfRangeException))]
        public void InvalidBufferSizeNegativeTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(-10, false);
        }

        /// <summary>
        /// Check that an empty buffer is rejected with an <see cref="ArgumentException"/>.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(ArgumentException))]
        public void InvalidBufferEmptyTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(new byte[0], false);
        }

        /// <summary>
        /// Check that a null buffer is rejected with an <see cref="ArgumentException"/>.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(ArgumentException))]
        public void InvalidBufferNullTest()
        {
            // NOTE: we would prefer this failed statically, but it performs an implict cast from byte[]
            using CommMemory<byte> cm = new CommMemory<byte>(null, false);
        }

        /// <summary>
        /// Check that the backing buffer is cleared on read is <see cref="CommMemory{TValue}.ClearOnRead"/> is true.
        /// </summary>
        [TestMethod]
        public async Task ClearOnReadTest()
        {
            // tests the ClearOnTrue property when set to true
            // no attempt is made to test when ClearOnTrue is false
            string str = "TestString";

            int backingArraySize = 10;
            int dataSize = 6;

            var backingArray = new string[backingArraySize];
            using CommMemory<string> cm = new CommMemory<string>(backingArray, true);

            var data = new string[dataSize];
            data.AsSpan().Fill(str);

            await cm.WriteAsync(data);

            // data should appear in the array, otherwise our test is wrong
            Assert.AreEqual(dataSize, backingArray.Where(s => s == str).Count(), $"Backing array should contain {dataSize} copies of the string: test is broken");
            Assert.AreEqual(backingArraySize - dataSize, backingArray.Where(s => s == null).Count(), $"Backing array should contain {backingArraySize - dataSize} nulls: test is broken");

            await cm.ReadAsync(data);

            // data should not appear in the array, otherwise the implementation is wrong
            Assert.AreEqual(0, backingArray.Where(s => s == str).Count(), $"Backing array should contain 0 copies of the string");
            Assert.AreEqual(backingArraySize, backingArray.Where(s => s == null).Count(), $"Backing array should contain {backingArraySize} nulls");
        }
    }

    [TestClass]
    public class CommMemoryByteTestsNoClearOnRead : CommMemoryByteTests
    {
        protected override bool ClearOnRead => false;
    }

    [TestClass]
    public class CommMemoryByteTestsClearOnRead : CommMemoryByteTests
    {
        protected override bool ClearOnRead => true;
    }

    public abstract class CommMemoryByteTests
    {
        protected abstract bool ClearOnRead { get; }

        [TestMethod]
        public async Task SimpleLargeBufferTest()
        {
            await SimpleTest(10000, ClearOnRead);
        }

        [TestMethod]
        public async Task SimpleSmallBufferTest()
        {
            await SimpleTest(10, ClearOnRead);
        }

        [TestMethod]
        public async Task SimpleTinyBufferTest()
        {
            await SimpleTest(1, ClearOnRead);
        }

        /// <summary>
        /// Check that a <see cref="CommMemoryDisposedException"/> is thrown when a Read is cancelled.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(CommMemoryDisposedException))]
        public async Task CancelledReadExceptionTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);

            var cts = new System.Threading.CancellationTokenSource(100);
            await cm.ReadAsync(new byte[10], cts.Token);
        }

        /// <summary>
        /// Check that <see cref="CommMemory{TValue}.IsDisposed"/> is set when a Read is cancelled.
        /// </summary>
        [TestMethod]
        public async Task CancelledReadStateTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);

            try
            {
                var cts = new System.Threading.CancellationTokenSource(500);
                await cm.ReadAsync(new byte[10], cts.Token);

                Assert.Fail("ReadAsync should have failed");
            }
            catch
            {
            }

            Assert.IsTrue(cm.IsDisposed, "CommMemory should be Disposed");
        }

        /// <summary>
        /// Check that a waiting Write throws a <see cref="CommMemoryDisposedException"/> if a waiting pending read is cancelled.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(CommMemoryDisposedException))]
        public async Task CancelledAwaitingWriteTest()
        {
            // set up a writer which will block
            // then kill a reader while the writer is still blocked
            // then await the writer, so we find out how (if) it died

            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);

            await cm.WriteAsync(new byte[5]);
            var blockedWriter = cm.WriteAsync(new byte[20]);

            try
            {
                var cts = new System.Threading.CancellationTokenSource();
                cts.Cancel();
                await cm.ReadAsync(new byte[15], cts.Token);

                Assert.Fail("ReadAsync should have failed");
            }
            catch
            {
            }

            // this should throw CommMemoryDisposedException, NOT InvalidOperationException
            await blockedWriter;
        }

        /// <summary>
        /// Check that attempting to Read from a Disposed <see cref="CommMemory{TValue}"/> results in a <see cref="CommMemoryDisposedException"/>.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(CommMemoryDisposedException))]
        public async Task DeadReadTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);

            try
            {
                var cts = new System.Threading.CancellationTokenSource();
                cts.Cancel();
                await cm.ReadAsync(new byte[15], cts.Token);

                Assert.Fail("ReadAsync should have failed");
            }
            catch
            {
            }

            // this should throw
            await cm.ReadAsync(new byte[5]);
        }

        /// <summary>
        /// Check that attempting to Write to a Disposed <see cref="CommMemory{TValue}"/> results in a <see cref="CommMemoryDisposedException"/>.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(CommMemoryDisposedException))]
        public async Task DeadWriteTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);

            try
            {
                var cts = new System.Threading.CancellationTokenSource();
                cts.Cancel();
                await cm.ReadAsync(new byte[15], cts.Token);

                Assert.Fail("ReadAsync should have failed");
            }
            catch
            {
            }

            // this should throw
            await cm.WriteAsync(new byte[5]);
        }

        /// <summary>
        /// Check that <see cref="CommMemory{TValue}.Disposed"/> is called when we cancel a read.
        /// </summary>
        [TestMethod]
        public async Task DisposedCallbackOnReadCancellationTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);
            bool disposed = false;
            cm.Disposed += _cm => disposed = true;

            try
            {
                var cts = new System.Threading.CancellationTokenSource();
                cts.Cancel();
                await cm.ReadAsync(new byte[15], cts.Token);

                Assert.Fail("ReadAsync should have failed");
            }
            catch
            {
            }

            Assert.IsTrue(disposed);
        }

        /// <summary>
        /// Check that <see cref="CommMemory{TValue}.Disposed"/> is called when we call <see cref="CommMemory{TValue}.Dispose"/>.
        /// </summary>
        [TestMethod]
        public void DisposedCallbackOnDisposeTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);
            bool disposed = false;
            cm.Disposed += _cm => disposed = true;
            cm.Dispose();
            Assert.IsTrue(disposed);
        }

        /// <summary>
        /// Check that <see cref="CommMemory{TValue}.Disposed"/> is called with the correct instance.
        /// </summary>
        [TestMethod]
        public void DisposedCallbackIsntanceOnDisposeTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);
            CommMemory<byte> disposed = null;
            cm.Disposed += _cm => disposed = _cm;
            cm.Dispose();

            Assert.AreEqual(cm, disposed);
        }

        /// <summary>
        /// Check that a waiting Read throws a <see cref="CommMemoryDisposedException"/> if the <see cref="CommMemory{TValue}"/> is manually Disposed.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(CommMemoryDisposedException))]
        public async Task ReadReleasedOnDispose()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);
            var read = cm.ReadAsync(new byte[15]);
            cm.Dispose();
            await read;
        }

        /// <summary>
        /// Check that a waiting Write throws a <see cref="CommMemoryDisposedException"/> if the <see cref="CommMemory{TValue}"/> is manually Disposed.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(CommMemoryDisposedException))]
        public async Task WriteReleasedOnDispose()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);
            var read = cm.WriteAsync(new byte[15]);
            cm.Dispose();
            await read;
        }

        /// <summary>
        /// Check that we get an <see cref="InvalidOperationException"/> if we try to Read multiple times at once.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(InvalidOperationException))]
        public async Task DetectConcurrentReadTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(10, ClearOnRead);
            var buffer = new byte[10];
            var r0 = cm.ReadAsync(buffer);
            var r1 = cm.ReadAsync(buffer);

            try
            {
                await r1;
            }
            catch
            {
                throw;
            }
            finally
            {
                try
                {
                    cm.Dispose();
                    await r0;
                }
                catch
                {
                }
            }
        }

        /// <summary>
        /// Check that we get an <see cref="InvalidOperationException"/> if we try to Write multiple times at once.
        /// </summary>
        [TestMethod]
        [ExpectedException(typeof(InvalidOperationException))]
        public async Task DetectConcurrentWriteTest()
        {
            using CommMemory<byte> cm = new CommMemory<byte>(5, ClearOnRead);
            var buffer = new byte[10]; // data larger than buffer so that first write waits
            var w0 = cm.WriteAsync(buffer);
            var w1 = cm.WriteAsync(buffer);

            try
            {
                await w1;
            }
            catch
            {
                throw;
            }
            finally
            {
                try
                {
                    cm.Dispose();
                    await w0;
                }
                catch
                {
                }
            }
        }

        /// <summary>
        /// Performs basic tests concerning data-integrety and blocking<see cref="CommMemory{TValue}"/> state.
        /// </summary>
        /// <param name="capacity">The capacity of the <see cref="CommMemory{TValue}"/> to use.</param>
        /// <param name="clearOnRead">The value to provide for <see cref="CommMemory{TValue}.ClearOnRead"/>.</param>
        private static async Task SimpleTest(int capacity, bool clearOnRead)
        {
            using var cm = new CommMemory<byte>(capacity, clearOnRead);

            Assert.AreEqual(capacity, cm.Capacity);
            await SimpleTest(cm);
            Assert.AreEqual(capacity, cm.Capacity);
        }

        private static async Task SimpleTest(CommMemory<byte> cm)
        {
            // check basic read/write with signal works
            var sstr = "Hello, miserable world.";
            string rstr;

            bool disposeEventCalled = false;
            cm.Disposed += _cm => disposeEventCalled = true;

            var sbytes = System.Text.ASCIIEncoding.ASCII.GetBytes(sstr);
            var rbytes = new byte[sbytes.Length];

            // writer first
            var strWriter = cm.WriteAsync(sbytes);
            await Task.Delay(100); // give it a moment to sink in
            await cm.ReadAsync(rbytes);
            await strWriter;
            rstr = System.Text.ASCIIEncoding.ASCII.GetString(rbytes);

            Assert.AreEqual(sstr, rstr, "Data integrity not preserved");
            Assert.IsFalse(cm.IsDisposed, "CommMemory should not be dead");
            Assert.IsFalse(disposeEventCalled, "CommMemory.Disposed should not have been called");

            // reader first
            var strReader = cm.ReadAsync(rbytes);
            await Task.Delay(100); // give it a moment to sink in
            await cm.WriteAsync(sbytes);
            await strReader;
            rstr = System.Text.ASCIIEncoding.ASCII.GetString(rbytes);

            Assert.AreEqual(sstr, rstr, "Data integrity not preserved");
            Assert.IsFalse(cm.IsDisposed, "CommMemory should not be dead");
            Assert.IsFalse(disposeEventCalled, "CommMemory.Disposed should not have been called");
        }

        /// <summary>
        /// Checks that Reading and Writing many times with different buffer sizes produces consistent results.
        /// Non-deterministic: specificity is ~255/256
        /// </summary>
        [TestMethod]
        public async Task StressTest()
        {
            int bufferSize = 1000;
            int amount = 10000000; // 10million -> ~2seconds per test
            int maxChunk = 10000;

            Assert.IsTrue(maxChunk > bufferSize, "Test should include the situation where the maxChunk is greater than the buffer size: test is broken");

            CommMemory<byte> cm = new CommMemory<byte>(bufferSize, ClearOnRead);
            var wrapper = new ReadWriteCommMemoryByteWrapper(cm);

            await Utilities.StressReadWriteTest(wrapper, wrapper, amount, maxChunk);
        }
    }
}

Utilities.cs (Test utilities)

using Microsoft.VisualStudio.TestTools.UnitTesting;
using System;
using System.Threading;
using System.Threading.Tasks;

namespace Comm.Tests
{
    /// <summary>
    /// Test interface
    /// </summary>
    public interface IAsyncByteWriter
    {
        ValueTask WriteAsync(ArraySegment<byte> buffer);
    }

    /// <summary>
    /// Test interface
    /// </summary>
    public interface IAsyncByteReader
    {
        ValueTask ReadAsync(ArraySegment<byte> buffer, CancellationToken ct);
    }

    /// <summary>
    /// Utilty methods for testing.
    /// </summary>
    public static class Utilities
    {
        /// <summary>
        /// Writes random bytes to the writer, and reads them from the reader, checking that the xor of all bytes written and read is the same.
        /// </summary>
        /// <typeparam name="TReader">The type of <see cref="IAsyncByteReader" /> from which to read.</typeparam>
        /// <typeparam name="TWriter">The type of <see cref="IAsyncByteWriter" /> to which to write.</typeparam>
        /// <param name="reader">The <typeparamref name="TReader"/> from which to read.</param>
        /// <param name="writer">The <typeparamref name="TWriter"/> to which to write.</param>
        /// <param name="amount">The total number of bytes to be written.</param>
        /// <param name="maxChunk">The maximum number of bytes to be sent or requested at a time.</param>
        public static async Task StressReadWriteTest<TReader, TWriter>(TReader reader, TWriter writer, int amount, int maxChunk)
            where TReader : IAsyncByteReader where TWriter : IAsyncByteWriter
        {
            // tests read and write, where they try to write random sized blocks, some of which won't fit in in buffers
            var read = ReadRandom(reader, amount, maxChunk);
            var write = WriteRandom(writer, amount, maxChunk);

            var wx = await write;
            var rx = await read;
            Assert.AreEqual(wx, rx);
        }

        /// <summary>
        /// Writes <paramref name="amount"/> bytes to the writer in random quantities in the range [1, <paramref name="maxChunk"/>].
        /// Returns the xor of all bytes written.
        /// </summary>
        /// <typeparam name="TWriter">The type of <see cref="IAsyncByteWriter" /> to which to write.</typeparam>
        /// <param name="writer">The <typeparamref name="TWriter"/> to which to write.</param>
        /// <param name="amount">The total number of bytes to be written.</param>
        /// <param name="maxChunk">The maximum number of bytes to send per call to <see cref="IAsyncByteWriter.WriteAsync(ArraySegment{byte}, CancellationToken)"/>.</param>
        public static async Task<byte> WriteRandom<TWriter>(TWriter writer, int amount, int maxChunk)
            where TWriter : IAsyncByteWriter
        {
            int count = 0;

            byte xor = 0;
            byte[] buffer = new byte[maxChunk];

            Random rnd = new Random();
            while (count < amount)
            {
                int chunk = rnd.Next(buffer.Length - 1) + 1;
                chunk = Math.Min(chunk, amount - count);

                var data = new ArraySegment<byte>(buffer, 0, chunk);

                rnd.NextBytes(data);
                var write = writer.WriteAsync(data);

                for (int i = 0; i < chunk; i++)
                    xor ^= data[i];

                await write;
                count += chunk;
            }

            return xor;
        }

        /// <summary>
        /// Reads <paramref name="amount"/> bytes to the writer in random quantities in the range [1, <paramref name="maxChunk"/>].
        /// Returns the xor of all bytes read.
        /// </summary>
        /// <typeparam name="TReader">The type of <see cref="IAsyncByteReader" /> from which to read.</typeparam>
        /// <param name="reader">The <typeparamref name="TReader"/> from which to read.</param>
        /// <param name="amount">The total number of bytes to be written.</param>
        /// <param name="maxChunk">The maximum number of bytes requested per call to <see cref="IAsyncByteReader.ReadAsync(ArraySegment{byte}, CancellationToken)"/>.</param>
        public static async Task<byte> ReadRandom<TReader>(TReader reader, int amount, int maxChunk)
            where TReader : IAsyncByteReader
        {
            int count = 0;

            byte xor = 0;
            byte[] buffer = new byte[maxChunk];

            Random rnd = new Random();
            while (count < amount)
            {
                int chunk = rnd.Next(buffer.Length - 1) + 1;
                chunk = Math.Min(chunk, amount - count);

                var data = new ArraySegment<byte>(buffer, 0, chunk);

                await reader.ReadAsync(data, CancellationToken.None);

                for (int i = 0; i < chunk; i++)
                    xor ^= data[i];

                count += chunk;
            }

            return xor;
        }
    }
}

Questions and Concerns

• I'm not happy with DisposeInternal: I feel like releasing and disposing the semaphores is redundant, and liable to go wrong. I also feel I can't guarantee that Disposed is invoked, and that is less than ideal.

• Things like <see cref="CommMemory{byte}"/> don't work: it still calls it CommMemory<TValue> in intelli-sense: should I give up on the pretence, or is there some way to make it say CommMemory<byte>?

• Is there a better (BCL) Exception than InvalidOperationException for the concurrent-access detection?

• Is there an existing BCL class that does this (or can be wrapped up to do this) already which I have completely overlooked?

Answer 1

A little late answer. I have focused on CommMemory and tried to refactor it as I see fit. You have discussed most of the issues with dfhwze, so I have not much to add other than what I have commented inline in the code below. You may agree or not in my changes and/or use it as you want. Any way it was fun to review.

I have removed most of your own comments in order to clear it up a little bit (while reviewing - I don't mean your comments are misplaced) and my comments all start with "HH:".

  // HH: I definitely don't like the concept of self-disposing, so the below omits that possibility
  //     as well as the Disposed event. Instead is introduced a reinforced cancellation pattern, that allows for
  //     cancellation via a constructor supplied cancellation token, and the read and write processes react to the state of that.

  public sealed class CommMemory<TValue> : IDisposable
  {
    // HH: Added a CancellationToken as mandatory argument
    public CommMemory(int capacity, bool clearOnRead, CancellationToken cancellationToken)
      : this(new Memory<TValue>(new TValue[CheckCapacity(capacity)]), clearOnRead, cancellationToken)
    {
      // nix
    }

    // HH: Added a CancellationToken as mandatory argument
    public CommMemory(Memory<TValue> buffer, bool clearOnRead, CancellationToken cancellationToken)
    {
      if (buffer.Length == 0)
        throw new ArgumentException("Cannot use a buffer of length 0 for CommMemory");

      memoryBuffer = buffer;
      ClearOnRead = clearOnRead;
      this.cancellationToken = cancellationToken;
      memPosition = 0;
      memUsed = 0;
    }

    // HH: All these fields:...
    //private int ReadPosition;
    //private int ReadAvailable;
    //private int WritePosition;
    //private int WriteAvailable;
    // HH: ... can be replaced with these:
    private int memPosition;
    private int memUsed; // HH: See further below.

    private readonly Memory<TValue> memoryBuffer;
    private readonly CancellationToken cancellationToken; // HH: Added this to the constructor in order to cancel from both read and write clients
    public int Capacity => memoryBuffer.Length; // HH: Capacity is a property of memoryBuffer

    private readonly object availabilityLock = new object();
    private readonly object disposeLock = new object();
    private readonly object guardLock = new object();

    // HH: changed name from ReadAvailableSemaphore
    private readonly SemaphoreSlim readGuard = new SemaphoreSlim(0);
    // HH: Guard against concurrent read
    private readonly SemaphoreSlim readEntryGuard = new SemaphoreSlim(1);
    // HH: changed named from WriteAvailableSemaphore
    private readonly SemaphoreSlim writeGuard = new SemaphoreSlim(0);
    // HH: Guard against concurrent write
    private readonly SemaphoreSlim writeEntryGuard = new SemaphoreSlim(1);

    private volatile bool isDisposed = false;
    public bool IsDisposed => isDisposed;
    public bool ClearOnRead { get; }

    private static int CheckCapacity(int capacity)
    {
      if (capacity <= 0)
        throw new ArgumentOutOfRangeException("Capacity must be non-negative (greater than 0)");

      return capacity;
    }

    public async ValueTask WriteAsync(ReadOnlyMemory<TValue> source)
    {
      if (source.Length == 0) return;

      // HH: IMO all the EnsureAlive() is false security, because everywhere in between these calls in both read and write
      //     anything can happen. Therefore two initial checks of disposed state here:
      CheckDisposed();

      try
      {
        // HH: Guard against concurrent entrence to write
        await writeEntryGuard.WaitAsync(cancellationToken);

        CheckDisposed();

        int written = 0;

        while (written < source.Length)
        {
          cancellationToken.ThrowIfCancellationRequested();
          int currentWritten = PerformWrite(ref source, written);
          written += currentWritten;
          // HH: Here I let it be up to the read thread to determine if there is anything to read so far. If not read will wait
          CompleteWrite(currentWritten);
          TryReleaseGuard(readGuard);

          if (currentWritten == 0)
          {
            await writeGuard.WaitAsync(cancellationToken);
          }
        }
      }
      finally
      {
        writeEntryGuard.Release();
      }
    }

    private int PerformWrite(ref ReadOnlyMemory<TValue> source, int sourcePosition)
    {
      (int writePosition, int length) = GetWriteInfo();
      if (length == 0) return length;

      length = Math.Min(length, source.Length - sourcePosition);

      // HH: Moved all manipulation of source and memoryBuffer to here
      //     So no magic happens in the call to this method...
      ReadOnlyMemory<TValue> slice = source.Slice(sourcePosition, length);

      // HH: I don't think the below will ever happen (in the original), because you guard carefully against that in WriteAsync()
      //     And that holds for PerformRead() as well. That eliminates - as I see it - the possibility/need to self-dispose
      //if (required > WriteAvailable)
      //{
      //  DisposeInternal();
      //  throw new Exception("Internal error in CommMemory: attempted to perform a write with insufficient space available.");
      //}

      // determine whether we need to cut the write in two
      if (writePosition + length > Capacity)
      {
        int headLength = Capacity - writePosition; // length of the first write
        int tailLength = length - headLength; // length of the second write

        var s0 = slice.Slice(0, headLength);
        var m0 = memoryBuffer.Slice(writePosition, headLength);
        s0.CopyTo(m0);

        var s1 = slice.Slice(headLength, tailLength);
        var m1 = memoryBuffer.Slice(0, tailLength);
        s1.CopyTo(m1);
      }
      else
      {
        Memory<TValue> m = memoryBuffer.Slice(writePosition, length);
        slice.CopyTo(m);
      }

      return length;
    }

    public async ValueTask ReadAsync(Memory<TValue> destination)
    {
      if (destination.Length == 0) return;

      // HH: Equivalent considerations as for WriteAsync()

      CheckDisposed();

      try
      {
        await readEntryGuard.WaitAsync(cancellationToken);

        CheckDisposed();

        int read = 0;

        while (read < destination.Length)
        {
          cancellationToken.ThrowIfCancellationRequested();
          int currentRead = 0;
          currentRead = PerformRead(ref destination, read);
          read += currentRead;
          CompleteRead(currentRead);
          TryReleaseGuard(writeGuard);

          if (currentRead == 0)
          {
            await readGuard.WaitAsync(cancellationToken);
          }
        }
      }
      finally
      {
        readEntryGuard.Release();
      }
    }

    private int PerformRead(ref Memory<TValue> destination, int destinationStart)
    {
      // HH: The same considerations as for PerformWrite()

      (int readPosition, int length) = GetReadInfo();

      if (length == 0) return 0;

      length = Math.Min(length, destination.Length - destinationStart);

      Memory<TValue> slice = destination.Slice(destinationStart, length);

      // determine whether we need to cut the read in two
      if (readPosition + length > Capacity)
      {
        int headLength = Capacity - readPosition; // length of the first read
        int tailLength = length - headLength; // length of the second read

        var d0 = slice.Slice(0, headLength);
        var m0 = memoryBuffer.Slice(readPosition, headLength);
        m0.CopyTo(d0);

        var d1 = slice.Slice(headLength, tailLength);
        var m1 = memoryBuffer.Slice(0, tailLength);
        m1.CopyTo(d1);

        if (ClearOnRead)
        {
          m0.Span.Fill(default(TValue));
          m1.Span.Fill(default(TValue));
        }
      }
      else
      {
        var m = memoryBuffer.Slice(readPosition, length);
        m.CopyTo(slice);

        if (ClearOnRead)
        {
          m.Span.Fill(default(TValue));
        }
      }

      return length;
    }

    private (int position, int length) GetWriteInfo()
    {
      lock (availabilityLock)
      {
        // HH: The available memory for writing is all that are not waiting to be read.
        int position = (memPosition + memUsed) % Capacity;
        return (position, Capacity - memUsed);
      }
    }

    private (int position, int length) GetReadInfo()
    {
      lock (availabilityLock)
      {
        // HH: Avialable memory for reading is represented by the values of memPosition and memUsed
        return (memPosition, memUsed);
      }
    }

    private void CompleteWrite(int consumed)
    {
      if (consumed <= 0) return;
      lock (availabilityLock)
      {
        // HH: Just add the consumed memory to the already used
        memUsed += consumed;
      }
    }

    private void CompleteRead(int consumed)
    {
      if (consumed <= 0) return;

      lock (availabilityLock)
      {
        // HH: Offest memPosition by consumed and align to Capacity
        memPosition = (memPosition + consumed) % Capacity;
        // HH: reduce memUsed by the memory read 
        memUsed -= consumed;
      }
    }

    private void TryReleaseGuard(SemaphoreSlim guard)
    {
      lock (guardLock)
      {// HH: I'm not quite sure this is necessary, but I try to prevent the guard to allow more than one entrence at a time
        if (guard.CurrentCount == 0)
        {
          guard.Release();
        }
      }
    }

    public void Dispose()
    {
      Dispose(true);
    }

    // HH: renamed DisposeInternal() and implemented the normal Dispose pattern
    private void Dispose(bool isDisposing)
    {
      if (isDisposing)
      {
        lock (disposeLock)
        {
          if (isDisposed) return;
          isDisposed = true;

          memoryBuffer.Span.Clear();

          readEntryGuard.Release();
          readEntryGuard.Dispose();
          readGuard.Release();
          readGuard.Dispose();

          writeEntryGuard.Release();
          writeEntryGuard.Dispose();
          writeGuard.Release();
          writeGuard.Dispose();
        }
      }
    }

    private void CheckDisposed()
    {
      if (isDisposed)
      {
        // HH: changed from CommMemoryDisposedException
        throw new ObjectDisposedException(nameof(CommMemory<TValue>));
      }
    }
  }

Answer 2

Quick Review

An API like this, dealing with thread-sensitive operations, requires time and effort to test and review rigorously. When I will find this time, I will do a thorough review. But here are some things I notice right off the bat.

• CommMemoryDisposedException should inherit from ObjectDisposedException. This way, consumers can handle your exception as a common exception when an object is disposed.
• Private utility methods like CheckCapacity, EnsureAlive and EnsureAliveAcquire could/should be adorned with [DebuggerStepThrough] or any of its sibling attributes to enhance debugging interop.
• Local variables ReadAvailable and WriteAvailable may optimize performance, but they are not required.
• Local variables ReadRequest, WriteRequest, ReadState and WriteState and the logic build around them should be replaced with ReaderWriterLock. Although, reading your goals, you don't want to allow concurrent access. So only implement this if your goals would allow for it.
• Using Semaphore for signaling threads would not be my preferred option. I would opt to use ManualResetEventSlim.
• Variable IsDisposed can return true even if its state might indicate otherwise. Is this a hack? Perhaps you're also including IsDisposing in here.
• DisposeInternal should block new readers and writers while allowing existing operations to continue.
• DisposeInternal does not clean up event Disposed. This is a memory leak waiting to happen.
• Think about which thread should invoke the event Disposed. I would execute this event asynchronously not to influence the current thread that is disposing the instance.
• An object that is able to Dispose itself is an anti-pattern in my opinion. A mediator or owner object should handle its lifetime.
• Methods RequestWrite and RequestRead perform double-checked locking. This might seem a good idea not to eagerly acquire a lock. But it can still lead to race conditions. Avoid double-checking and just acquire the lock. There are several articles online that address these issues.
• You have written an elaborate question with a good spec and lots of unit tests. Your code also reads very well.

Your goals are clear about not allowing concurrent read/write access. However, if you do want to allow this, then..

• you are basically implementing a blocking queue. C# comes with BlockingCollection and ConcurrentQueue. Unfortunately, both don't exactly apply to your case. Java does have a class that does what you want, even using a circular buffer: ArrayBlockingQueue. Perhaps you could find some inspiration in its implementation. It uses Java's built in class Condition which unfortunately does not exist in C#. You have to work around that by using ManualResetEventSlim.