/* ========================================================================
* 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);
}
}
}