UA-.NETStandard/tests/Opc.Ua.Client.Tests/Utils/AsyncReaderWriterLockTests.cs

/* ========================================================================
 * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
 *
 * OPC Foundation MIT License 1.00
 *
 * Permission is hereby granted, free of charge, to any person
 * obtaining a copy of this software and associated documentation
 * files (the "Software"), to deal in the Software without
 * restriction, including without limitation the rights to use,
 * copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following
 * conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
 * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 * OTHER DEALINGS IN THE SOFTWARE.
 *
 * The complete license agreement can be found here:
 * http://opcfoundation.org/License/MIT/1.00/
 * ======================================================================*/

using System;
using System.Threading;
using System.Threading.Tasks;
using NUnit.Framework;

namespace Opc.Ua.Client.Tests.AsyncPrimitives
{
    /// <summary>
    /// Unit tests for the local hand-rolled
    /// <see cref="AsyncReaderWriterLock"/>.
    /// </summary>
    [TestFixture]
    [Category("Unit")]
    public sealed class AsyncReaderWriterLockTests
    {
        [TestCase(false)]
        [TestCase(true)]
        public async Task ReaderAvailabilityHookIsCapturedOnceAndNullIsAllowedAsync(bool hasHook)
        {
            using var rwLock = new AsyncReaderWriterLock();
            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
            var gate = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
            int calls = 0;
            if (hasHook)
            {
                rwLock.BeforeReaderLockAsync = ct =>
                {
                    Interlocked.Increment(ref calls);
                    return new ValueTask(gate.Task.WaitAsync(ct));
                };
            }
            Task<AsyncReaderWriterLock.Releaser> pending = rwLock.ReaderLockAsync(timeout.Token).AsTask();
            try
            {
                Assert.That(pending.IsCompleted, Is.EqualTo(!hasHook));
                rwLock.BeforeReaderLockAsync = _ => throw new InvalidOperationException("Replacement was invoked.");
            }
            finally
            {
                gate.TrySetResult(true);
            }
            using AsyncReaderWriterLock.Releaser reader = await pending.ConfigureAwait(false);
            Assert.That(calls, Is.EqualTo(hasHook ? 1 : 0));
        }

        /// <summary>
        /// L2-3: recovery must not wait for readers stuck on a dead channel
        /// until their request timeout. With a drain timeout the writer
        /// proceeds while stale readers are still active, and new readers
        /// stay excluded until the writer releases.
        /// </summary>
        [Test]
        public async Task WriterWithDrainTimeoutProceedsPastStuckReaderAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();
            AsyncReaderWriterLock.Releaser stuckReader =
                await rwLock.ReaderLockAsync().ConfigureAwait(false);

            Task<AsyncReaderWriterLock.Releaser> writerTask = rwLock
                .WriterLockAsync(TimeSpan.FromMilliseconds(100), TimeProvider.System, () => true)
                .AsTask();
            AsyncReaderWriterLock.Releaser writer =
                await writerTask.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);

            Task<AsyncReaderWriterLock.Releaser> newReader = rwLock.ReaderLockAsync().AsTask();
            await Task.Delay(50).ConfigureAwait(false);
            Assert.That(newReader.IsCompleted, Is.False,
                "New readers must wait while the writer holds the lock.");

            stuckReader.Dispose();
            writer.Dispose();
            using AsyncReaderWriterLock.Releaser reader =
                await newReader.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);

            // The lock is consistent afterwards: a plain writer drains normally.
            reader.Dispose();
            using AsyncReaderWriterLock.Releaser next = await rwLock.WriterLockAsync()
                .AsTask().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
        }

        /// <summary>
        /// Review G8a: while the readers can still complete normally (healthy
        /// channel), the recovery writer stays exclusive past the drain
        /// timeout and proceeds only once the readers leave.
        /// </summary>
        [Test]
        public async Task WriterWithDrainTimeoutStaysExclusiveWhileReadersMayCompleteAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();
            AsyncReaderWriterLock.Releaser reader =
                await rwLock.ReaderLockAsync().ConfigureAwait(false);
            int checks = 0;

            Task<AsyncReaderWriterLock.Releaser> writerTask = rwLock
                .WriterLockAsync(TimeSpan.FromMilliseconds(50), TimeProvider.System, () =>
                {
                    Interlocked.Increment(ref checks);
                    return false;
                })
                .AsTask();

            await Task.Delay(400).ConfigureAwait(false);
            Assert.That(writerTask.IsCompleted, Is.False,
                "The writer must keep waiting for readers on a healthy channel.");
            Assert.That(Volatile.Read(ref checks), Is.GreaterThan(1),
                "The channel health must be re-checked at every drain timeout.");

            reader.Dispose();
            using AsyncReaderWriterLock.Releaser writer =
                await writerTask.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
        }

        /// <summary>
        /// Review G8a: the channel may die while the writer waits for the
        /// readers; the next check then lets the writer proceed.
        /// </summary>
        [Test]
        public async Task WriterWithDrainTimeoutProceedsOnceChannelBecomesUnhealthyAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();
            AsyncReaderWriterLock.Releaser stuckReader =
                await rwLock.ReaderLockAsync().ConfigureAwait(false);
            bool unhealthy = false;

            Task<AsyncReaderWriterLock.Releaser> writerTask = rwLock
                .WriterLockAsync(TimeSpan.FromMilliseconds(50), TimeProvider.System, () => Volatile.Read(ref unhealthy))
                .AsTask();

            await Task.Delay(200).ConfigureAwait(false);
            Assert.That(writerTask.IsCompleted, Is.False);

            Volatile.Write(ref unhealthy, true);
            using (await writerTask.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false))
            {
                stuckReader.Dispose();
            }

            using AsyncReaderWriterLock.Releaser next = await rwLock.WriterLockAsync()
                .AsTask().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
        }

        [Test]
        public async Task ReadersDoNotMutuallyExcludeAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();

            // Acquire two readers concurrently — neither should
            // block the other.
            using AsyncReaderWriterLock.Releaser r1 =
                await rwLock.ReaderLockAsync().ConfigureAwait(false);
            using AsyncReaderWriterLock.Releaser r2 =
                await rwLock.ReaderLockAsync().ConfigureAwait(false);

            // No assertion needed — passing means the second
            // ReaderLockAsync did not block on the first.
            Assert.Pass();
        }

        [Test]
        public async Task WriterExcludesReadersAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();

            AsyncReaderWriterLock.Releaser writer =
                await rwLock.WriterLockAsync().ConfigureAwait(false);

            Task<AsyncReaderWriterLock.Releaser> readerTask =
                rwLock.ReaderLockAsync().AsTask();

            // Reader must not complete while writer is held.
            await Task.Delay(50).ConfigureAwait(false);
            Assert.That(readerTask.IsCompleted, Is.False,
                "Reader must wait while writer holds the lock.");

            writer.Dispose();

            using AsyncReaderWriterLock.Releaser reader =
                await readerTask.ConfigureAwait(false);
            // Reader proceeded after writer released.
        }

        [Test]
        public async Task WriterWaitsForReadersToDrainAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();

            AsyncReaderWriterLock.Releaser r1 =
                await rwLock.ReaderLockAsync().ConfigureAwait(false);
            AsyncReaderWriterLock.Releaser r2 =
                await rwLock.ReaderLockAsync().ConfigureAwait(false);

            Task<AsyncReaderWriterLock.Releaser> writerTask =
                rwLock.WriterLockAsync().AsTask();

            await Task.Delay(50).ConfigureAwait(false);
            Assert.That(writerTask.IsCompleted, Is.False,
                "Writer must not acquire while readers are active.");

            r1.Dispose();
            await Task.Delay(50).ConfigureAwait(false);
            Assert.That(writerTask.IsCompleted, Is.False,
                "Writer must wait for ALL readers to drain, not just " +
                "the first.");

            r2.Dispose();

            using AsyncReaderWriterLock.Releaser writer =
                await writerTask.ConfigureAwait(false);
            // Writer acquired after last reader drained.
        }

        [Test]
        [Repeat(50)]
        [CancelAfter(30_000)]
        public async Task LastReaderNextReaderRaceDoesNotAdmitWriterEarlyAsync(
            CancellationToken ct)
        {
            // Stress test the race the rubber-duck review identified:
            // a naive `Interlocked` reader counter without serialized
            // pairing of counter + drain signal could let a writer
            // observe drain==signaled while a fresh reader has just
            // entered. This test runs many tight reader cycles
            // alongside one writer that asserts m_activeReaders == 0
            // (via a probe) when it acquires.
            using var rwLock = new AsyncReaderWriterLock();
            int activeReaders = 0;
            int maxObservedReadersInsideWriter = 0;

            // Start a churn of readers in tight loops.
            using var churnCts = CancellationTokenSource
                .CreateLinkedTokenSource(ct);
            var churn = new Task[8];
            for (int i = 0; i < churn.Length; i++)
            {
                churn[i] = Task.Run(async () =>
                {
                    while (!churnCts.IsCancellationRequested)
                    {
                        using AsyncReaderWriterLock.Releaser r =
                            await rwLock.ReaderLockAsync(churnCts.Token).ConfigureAwait(false);
                        Interlocked.Increment(ref activeReaders);
                        await Task.Yield();
                        Interlocked.Decrement(ref activeReaders);
                    }
                }, churnCts.Token);
            }

            // Run several writer acquisitions interleaved with the churn.
            for (int round = 0; round < 20; round++)
            {
                using AsyncReaderWriterLock.Releaser w = await rwLock
                    .WriterLockAsync(ct).ConfigureAwait(false);
                int observed = Volatile.Read(ref activeReaders);
                if (observed > maxObservedReadersInsideWriter)
                {
                    maxObservedReadersInsideWriter = observed;
                }
                Assert.That(observed, Is.Zero,
                    $"Writer #{round} observed {observed} active " +
                    "readers while holding the writer lock — the " +
                    "reader-counter / drain-signal pairing is racy.");
                await Task.Yield();
            }

            await churnCts.CancelAsync().ConfigureAwait(false);
            try
            {
                await Task.WhenAll(churn).ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
                // expected on shutdown
            }

            Assert.That(maxObservedReadersInsideWriter, Is.Zero);
        }

        [Test]
        [CancelAfter(10_000)]
        public async Task WriterCancellationWhileDrainingReleasesSemaphoreAsync(
            CancellationToken testCt)
        {
            using var rwLock = new AsyncReaderWriterLock();

            // Hold a reader so the writer must wait for drain.
            AsyncReaderWriterLock.Releaser reader =
                await rwLock.ReaderLockAsync(testCt).ConfigureAwait(false);

            using var writerCts = new CancellationTokenSource();
            Task<AsyncReaderWriterLock.Releaser> writerTask =
                rwLock.WriterLockAsync(writerCts.Token).AsTask();

            // Ensure the writer has reached the drain wait.
            await Task.Delay(50, testCt).ConfigureAwait(false);
            Assert.That(writerTask.IsCompleted, Is.False);

            // Cancel the writer while draining.
            await writerCts.CancelAsync().ConfigureAwait(false);
            Assert.That(
                () => writerTask,
                Throws.InstanceOf<OperationCanceledException>());

            // The cancelled writer must have released the writer-entry
            // semaphore — the next writer should be able to acquire
            // immediately once the reader leaves.
            reader.Dispose();
            using AsyncReaderWriterLock.Releaser nextWriter =
                await rwLock.WriterLockAsync(testCt).ConfigureAwait(false);
            // If the cancelled writer leaked the semaphore, the test
            // would hang here (caught by [CancelAfter]).
        }

        [Test]
        [CancelAfter(10_000)]
        public async Task WriterCanReacquireAfterReleaseAsync(
            CancellationToken ct)
        {
            using var rwLock = new AsyncReaderWriterLock();

            using (AsyncReaderWriterLock.Releaser w1 =
                await rwLock.WriterLockAsync(ct).ConfigureAwait(false))
            {
                // hold and release
            }

            using AsyncReaderWriterLock.Releaser w2 =
                await rwLock.WriterLockAsync(ct).ConfigureAwait(false);

            // would deadlock if the previous writer leaked.
        }

        [Test]
        [CancelAfter(10_000)]
        public async Task WriterIsNotReentrantAndDeadlocksWithSelfAsync(CancellationToken ct)
        {
            // Sanity: reentrancy is intentionally NOT supported. A
            // writer that asks for the writer lock again on the same
            // logical flow must NOT silently succeed (which would
            // indicate accidental reentrancy). It deadlocks; we
            // detect by short timeout.
            using var rwLock = new AsyncReaderWriterLock();
            using AsyncReaderWriterLock.Releaser outer =
                await rwLock.WriterLockAsync(ct).ConfigureAwait(false);

            using var innerCts = CancellationTokenSource
                .CreateLinkedTokenSource(ct);
            innerCts.CancelAfter(TimeSpan.FromMilliseconds(300));

            Assert.That(
                async () => await rwLock
                    .WriterLockAsync(innerCts.Token)
                    .ConfigureAwait(false),
                Throws.InstanceOf<OperationCanceledException>());
        }

        [Test]
        [CancelAfter(10_000)]
        public async Task ManyParallelReadersAdmittedAsync(CancellationToken ct)
        {
            // Spin up 32 readers in parallel; all should hold the lock
            // simultaneously without blocking each other.
            using var rwLock = new AsyncReaderWriterLock();
            var startGate = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            int holdingCount = 0;
            int peakHolding = 0;
            const int kReaders = 32;
            var releaseGate = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);

            var readers = new Task[kReaders];
            for (int i = 0; i < kReaders; i++)
            {
                readers[i] = Task.Run(async () =>
                {
                    using AsyncReaderWriterLock.Releaser r =
                    await rwLock.ReaderLockAsync(ct).ConfigureAwait(false);
                    int now = Interlocked.Increment(ref holdingCount);
                    int peak;
                    do
                    {
                        peak = Volatile.Read(ref peakHolding);
                        if (now <= peak)
                        {
                            break;
                        }
                    } while (Interlocked.CompareExchange(ref peakHolding, now, peak) != peak);
                    await releaseGate.Task.ConfigureAwait(false);
                    Interlocked.Decrement(ref holdingCount);
                }, ct);
            }

            // Give all readers time to acquire.
            await Task.Delay(200, ct).ConfigureAwait(false);
            releaseGate.SetResult(true);
            await Task.WhenAll(readers).ConfigureAwait(false);

            Assert.That(peakHolding, Is.EqualTo(kReaders),
                "All 32 readers should have held the lock simultaneously.");
        }

        [Test]
        public async Task TryEnterReaderFailsWhileAWriterHoldsTheLockAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();
            using (await rwLock.WriterLockAsync().ConfigureAwait(false))
            {
                Assert.That(rwLock.TryEnterReader(), Is.False);
            }
            Assert.That(rwLock.TryEnterReader(), Is.True);
            rwLock.ExitReader();
        }

        [Test]
        public async Task ReaderFromTryEnterIsHeldUntilTheOperationCompletesAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();
            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
            var operation = new TaskCompletionSource<int>();
            Assert.That(rwLock.TryEnterReader(), Is.True);
            ValueTask<int> call = rwLock.ExitReaderOnCompletion(new ValueTask<int>(operation.Task));

            Task<AsyncReaderWriterLock.Releaser> writer = rwLock.WriterLockAsync(timeout.Token).AsTask();
            await Task.Delay(50, timeout.Token).ConfigureAwait(false);
            Assert.That(writer.IsCompleted, Is.False, "the writer waits for the reader to drain");

            operation.SetResult(42);
            using (await writer.WaitAsync(timeout.Token).ConfigureAwait(false))
            {
                Assert.That(await call.ConfigureAwait(false), Is.EqualTo(42));
            }
        }

        [Test]
        public async Task AwaiterResumesAfterTheReaderFromTryEnterIsReleasedAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();
            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
            var operation = new TaskCompletionSource<int>();
            Assert.That(rwLock.TryEnterReader(), Is.True);
            ValueTask<int> call = rwLock.ExitReaderOnCompletion(new ValueTask<int>(operation.Task));

            // A writer requested in the awaiter's continuation finds no reader left,
            // as after a using block around the awaited operation.
            bool writerAcquiredAtOnce = false;
            Task resumed = Task.Run(async () =>
            {
                await call.ConfigureAwait(false);
                ValueTask<AsyncReaderWriterLock.Releaser> writer = rwLock.WriterLockAsync(timeout.Token);
                writerAcquiredAtOnce = writer.IsCompleted;
                (await writer.ConfigureAwait(false)).Dispose();
            });
            await Task.Delay(50, timeout.Token).ConfigureAwait(false);
            operation.SetResult(42);

            await resumed.WaitAsync(timeout.Token).ConfigureAwait(false);
            Assert.That(writerAcquiredAtOnce, Is.True);
        }

        [Test]
        public async Task ReaderFromTryEnterIsReleasedWhenTheOperationFailsAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();
            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
            var operation = new TaskCompletionSource<int>();
            Assert.That(rwLock.TryEnterReader(), Is.True);
            ValueTask<int> call = rwLock.ExitReaderOnCompletion(new ValueTask<int>(operation.Task));
            operation.SetException(new InvalidOperationException("failed"));

            Assert.That(async () => await call.ConfigureAwait(false), Throws.InvalidOperationException);
            using AsyncReaderWriterLock.Releaser writer =
                await rwLock.WriterLockAsync(timeout.Token).ConfigureAwait(false);
        }

        [Test]
        public async Task ReaderFromTryEnterIsReleasedAtOnceForACompletedOperationAsync()
        {
            using var rwLock = new AsyncReaderWriterLock();
            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
            Assert.That(rwLock.TryEnterReader(), Is.True);
            ValueTask<int> call = rwLock.ExitReaderOnCompletion(new ValueTask<int>(7));
            Assert.That(call.IsCompleted, Is.True);
            Assert.That(await call.ConfigureAwait(false), Is.EqualTo(7));
            using AsyncReaderWriterLock.Releaser writer =
                await rwLock.WriterLockAsync(timeout.Token).ConfigureAwait(false);
        }
    }
}