/* ======================================================================== * 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.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; using Moq; using NUnit.Framework; using Opc.Ua.Tests; // The obsolete RequestCompleted is the direct way to complete a request in these tests. #pragma warning disable CS0618 namespace Opc.Ua.Server.Tests { /// /// Verifies the lock-free request table of the and that /// drains, Cancel and Session close still see every request started before them while /// registrations and completions run concurrently. /// [TestFixture] [Category("RequestManager")] public class RequestTableTests { [TestCase(1, 16)] [TestCase(16, 16)] [TestCase(17, 32)] [TestCase(4000, 4096)] public void CapacityIsAPowerOfTwoOfAtLeastTheProbeLimit(int capacity, int expected) { Assert.That(new RequestTable(capacity).Capacity, Is.EqualTo(expected)); } [Test] public void AddIsIdempotentAndRemoveSucceedsOnce() { var table = new RequestTable(64); var entry = new TestEntry(7); var sameKey = new TestEntry(7); table.Add(entry); table.Add(entry); table.Add(sameKey); Assert.That(table.GetEntries(), Is.EquivalentTo(new[] { entry, sameKey }), "entries are tracked by reference"); Assert.That(table.Contains(entry), Is.True); Assert.That(table.Remove(entry), Is.True); Assert.That(table.Remove(entry), Is.False); Assert.That(table.Contains(entry), Is.False); Assert.That(table.Contains(sameKey), Is.True); Assert.That(table.Remove(new TestEntry(7)), Is.False, "an entry never added is not removed"); // an entry can be added again after it was removed. table.Add(entry); Assert.That(table.Contains(entry), Is.True); } /// /// Concurrent adds of the same entry place it once, so one remove takes it out and no /// ghost entry is left behind. /// [Test] public async Task ConcurrentAddsOfTheSameEntryPlaceItOnceAsync() { var table = new RequestTable(64); for (int attempt = 0; attempt < 2000; attempt++) { var entry = new TestEntry((uint)attempt); using var start = new Barrier(4); var adders = new Task[4]; for (int ii = 0; ii < adders.Length; ii++) { adders[ii] = Task.Run(() => { start.SignalAndWait(); table.Add(entry); }); } await Task.WhenAll(adders).WaitAsync(TimeSpan.FromSeconds(30)).ConfigureAwait(false); Assert.That(table.GetEntries(), Has.Count.EqualTo(1), $"attempt {attempt}"); Assert.That(table.Remove(entry), Is.True); Assert.That(table.GetEntries(), Is.Empty, $"attempt {attempt}: ghost entry left"); } } [Test] public void EntriesBeyondTheProbedSlotsOverflowAndAreStillTracked() { var table = new RequestTable(16); var entries = new List(); for (int ii = 0; ii < 40; ii++) { var entry = new TestEntry(5); entries.Add(entry); table.Add(entry); } Assert.That(table.OverflowCount, Is.EqualTo(40 - 16)); Assert.That(entries.TrueForAll(table.Contains), Is.True); Assert.That(table.GetEntries(), Is.EquivalentTo(entries)); Assert.That(entries.TrueForAll(table.Remove), Is.True); Assert.That(table.GetEntries(), Is.Empty); Assert.That(table.OverflowCount, Is.Zero); foreach (TestEntry entry in entries) { table.Add(entry); } Assert.That(table.Clear(), Is.EquivalentTo(entries)); Assert.That(table.GetEntries(), Is.Empty); Assert.That(entries.TrueForAll(entry => !table.Contains(entry)), Is.True); } /// /// Writers add and remove their own entries while another thread enumerates: every /// entry is removed exactly once and the table ends empty. /// [Test] public async Task ConcurrentAddRemoveAndEnumerationLeaveTheTableEmptyAsync() { var table = new RequestTable(256); using var stop = new CancellationTokenSource(); int failedRemovals = 0; int seenByScanner = 0; Task scanner = Task.Run(() => { while (!stop.IsCancellationRequested) { Interlocked.Add(ref seenByScanner, table.GetEntries().Count); } }); var writers = new Task[8]; for (int writer = 0; writer < writers.Length; writer++) { uint seed = (uint)writer * 1_000_000; writers[writer] = Task.Run(() => { var mine = new TestEntry[64]; for (int round = 0; round < 2000; round++) { for (int ii = 0; ii < mine.Length; ii++) { mine[ii] = new TestEntry(seed++); table.Add(mine[ii]); } foreach (TestEntry entry in mine) { if (!table.Remove(entry)) { Interlocked.Increment(ref failedRemovals); } } } }); } await Task.WhenAll(writers).WaitAsync(TimeSpan.FromMinutes(2)).ConfigureAwait(false); stop.Cancel(); await scanner.ConfigureAwait(false); Assert.That(failedRemovals, Is.Zero); Assert.That(table.GetEntries(), Is.Empty); Assert.That(table.OverflowCount, Is.Zero); TestContext.Out.WriteLine($"entries seen by the scanner: {seenByScanner}"); } private sealed class TestEntry : IRequestTableEntry { public TestEntry(uint key) { TableKey = key; } public uint TableKey { get; } public ref int TableSlot => ref m_tableSlot; private int m_tableSlot = RequestTable.kNoSlot; } /// /// A drain started while many requests are executing waits for each of /// them, however the completions interleave across threads, and completes exactly /// once after the last one. /// [Test] [Category("NodeManagerLifecycle")] public async Task DrainWaitsForEveryRequestStartedBeforeItWhileThreadsCompleteThemAsync() { using RequestManager manager = CreateManager(); const int count = 2000; var lifetimes = new RequestLifetime[count]; var contexts = new OperationContext[count]; for (int ii = 0; ii < count; ii++) { lifetimes[ii] = new RequestLifetime(); contexts[ii] = CreateContext((uint)ii, lifetimes[ii]); manager.RequestReceived(contexts[ii]); } Task drain = manager.WaitForCurrentRequestsAsync().AsTask(); Assert.That(drain.IsCompleted, Is.False); // Complete all but the last request from several threads, interleaved with // requests that start after the drain and must not extend it. int next = -1; var workers = new Task[8]; for (int worker = 0; worker < workers.Length; worker++) { workers[worker] = Task.Run(() => { int index; while ((index = Interlocked.Increment(ref next)) < count - 1) { using var lateLifetime = new RequestLifetime(); OperationContext late = CreateContext(uint.MaxValue, lateLifetime); manager.RequestReceived(late); manager.RequestCompleted(contexts[index]); manager.RequestCompleted(late); } }); } await Task.WhenAll(workers).ConfigureAwait(false); Assert.That(drain.IsCompleted, Is.False, "the drain completed before the last request it waits for"); manager.RequestCompleted(contexts[count - 1]); await drain.WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false); Assert.That(drain.Status, Is.EqualTo(TaskStatus.RanToCompletion)); Assert.That(lifetimes.All(lifetime => lifetime.StatusCode == StatusCodes.Good), Is.True, "completion marks each lifetime completed without cancelling it"); foreach (RequestLifetime lifetime in lifetimes) { lifetime.Dispose(); } } /// /// A Session close racing registrations cancels every request of the Session that /// was registered before it, exactly once, and none of another Session. /// [Test] public async Task SessionCloseCancelsEveryEarlierRequestExactlyOnceWhileOthersRegisterAsync() { using RequestManager manager = CreateManager(); var closing = new Mock(); closing.Setup(s => s.Id).Returns(new NodeId("Closing", 0)); var other = new Mock(); other.Setup(s => s.Id).Returns(new NodeId("Other", 0)); const int count = 1000; var before = new List<(OperationContext Context, RequestLifetime Lifetime)>(); for (int ii = 0; ii < count; ii++) { var lifetime = new RequestLifetime(); var context = new OperationContext( new RequestHeader { RequestHandle = (uint)ii }, null!, RequestType.Read, lifetime, closing.Object); manager.RequestReceived(context); before.Add((context, lifetime)); } using var stop = new CancellationTokenSource(); int otherCancelled = 0; Task churn = Task.Run(() => { while (!stop.IsCancellationRequested) { using var lifetime = new RequestLifetime(); var context = new OperationContext( new RequestHeader { RequestHandle = 1 }, null!, RequestType.Read, lifetime, other.Object); manager.RequestReceived(context); if (lifetime.CancellationToken.IsCancellationRequested) { Interlocked.Increment(ref otherCancelled); } manager.RequestCompleted(context); } }); uint aborted = manager.CancelSessionRequests(closing.Object.Id, 0, StatusCodes.BadSessionClosed); uint again = manager.CancelSessionRequests(closing.Object.Id, 0, StatusCodes.BadSessionClosed); stop.Cancel(); await churn.ConfigureAwait(false); Assert.Multiple(() => { Assert.That(aborted, Is.EqualTo(count)); Assert.That(again, Is.Zero, "a request is cancelled once"); Assert.That(before.All(entry => entry.Context.OperationStatus.Code == StatusCodes.BadSessionClosed), Is.True); Assert.That(otherCancelled, Is.Zero); }); foreach ((OperationContext context, RequestLifetime lifetime) in before) { manager.RequestCompleted(context); lifetime.Dispose(); } } /// /// A Cancel by request handle finds every matching request wherever it is in the table. /// [Test] public void CancelByHandleFindsEveryMatchingRequest() { using RequestManager manager = CreateManager(); var session = new Mock(); session.Setup(s => s.Id).Returns(new NodeId("CancelByHandle", 0)); var lifetimes = new List(); var matching = new List(); for (int ii = 0; ii < 256; ii++) { var lifetime = new RequestLifetime(); lifetimes.Add(lifetime); uint handle = ii % 2 == 0 ? 42u : (uint)ii; var context = new OperationContext( new RequestHeader { RequestHandle = handle }, null!, RequestType.Read, lifetime, session.Object); manager.RequestReceived(context); if (handle == 42) { matching.Add(context); } } manager.CancelRequests(session.Object.Id, 42, out uint cancelCount); Assert.That(cancelCount, Is.EqualTo((uint)matching.Count)); Assert.That(matching.All(context => context.OperationStatus.Code == StatusCodes.BadRequestCancelledByClient), Is.True); Assert.That(lifetimes.Count(lifetime => lifetime.CancellationToken.IsCancellationRequested), Is.EqualTo(matching.Count)); lifetimes.ForEach(lifetime => lifetime.Dispose()); } private static RequestManager CreateManager() { var server = new Mock(); server.Setup(s => s.Telemetry).Returns(NUnitTelemetryContext.Create()); return new RequestManager(server.Object); } private static OperationContext CreateContext(uint requestHandle, RequestLifetime lifetime) { return new OperationContext( new RequestHeader { RequestHandle = requestHandle }, null, RequestType.Read, lifetime); } } }