Joseph Doherty
467 lines
20 KiB
C#
467 lines
20 KiB
C#
using System.Diagnostics;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Opc.Ua;
|
|
using Opc.Ua.Server;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
|
|
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
|
|
|
/// <summary>
|
|
/// Arch-review 03/S3 — the node-manager's HistoryRead arms must serve a multi-node batch with BOUNDED
|
|
/// per-node parallelism, a per-request deadline, and a process-wide concurrent-batch limiter, rather
|
|
/// than block-bridging the gateway per node sequentially on an SDK request thread. Boots the same real
|
|
/// <see cref="OtOpcUaSdkServer"/> harness as <see cref="NodeManagerHistoryReadTests"/> and drives the
|
|
/// public <c>HistoryRead</c> with a delaying fake <see cref="IHistorianDataSource"/> that records the
|
|
/// maximum observed concurrent reads (deterministic — no raw-timing dependence for the primary
|
|
/// assertions).
|
|
/// </summary>
|
|
public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
|
|
{
|
|
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
|
|
|
private readonly string _pkiRoot = Path.Combine(
|
|
Path.GetTempPath(),
|
|
$"otopcua-historyread-conc-{Guid.NewGuid():N}");
|
|
|
|
/// <summary>
|
|
/// RED repro (03/S3): a single Raw HistoryRead naming 8 historized nodes against a 150 ms-per-read
|
|
/// backend must serve at least two nodes concurrently. On the current sequential <c>foreach</c> the
|
|
/// max observed concurrency is pinned at 1, so this fails.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Raw_batch_is_served_with_bounded_parallelism()
|
|
{
|
|
var (host, server) = await BootAsync();
|
|
var nm = server.NodeManager!;
|
|
var fake = new ConcurrencyTrackingHistorianDataSource(TimeSpan.FromMilliseconds(150));
|
|
nm.HistorianDataSource = fake;
|
|
|
|
var nodeIds = MaterializeHistorizedNodes(nm, count: 8);
|
|
|
|
var details = new ReadRawModifiedDetails
|
|
{
|
|
StartTime = DateTime.UtcNow.AddHours(-1),
|
|
EndTime = DateTime.UtcNow,
|
|
NumValuesPerNode = 10,
|
|
IsReadModified = false,
|
|
};
|
|
|
|
var sw = Stopwatch.StartNew();
|
|
await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
|
|
.WaitAsync(TimeSpan.FromSeconds(15), Ct);
|
|
sw.Stop();
|
|
|
|
// Primary (deterministic): the batch was served with >1 node in flight at once.
|
|
fake.MaxObservedConcurrency.ShouldBeGreaterThanOrEqualTo(2);
|
|
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Parallelism is bounded ABOVE by <c>HistoryReadBatchParallelism</c>: with the property set to 3 and
|
|
/// 12 nodes in flight, no more than 3 reads are ever concurrent (03/S3).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Parallelism_is_bounded_above()
|
|
{
|
|
var (host, server) = await BootAsync();
|
|
var nm = server.NodeManager!;
|
|
nm.HistoryReadBatchParallelism = 3;
|
|
var fake = new ConcurrencyTrackingHistorianDataSource(TimeSpan.FromMilliseconds(100));
|
|
nm.HistorianDataSource = fake;
|
|
|
|
var nodeIds = MaterializeHistorizedNodes(nm, count: 12);
|
|
var details = new ReadRawModifiedDetails
|
|
{
|
|
StartTime = DateTime.UtcNow.AddHours(-1),
|
|
EndTime = DateTime.UtcNow,
|
|
NumValuesPerNode = 10,
|
|
IsReadModified = false,
|
|
};
|
|
|
|
await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
|
|
.WaitAsync(TimeSpan.FromSeconds(15), Ct);
|
|
|
|
fake.MaxObservedConcurrency.ShouldBeGreaterThanOrEqualTo(2);
|
|
fake.MaxObservedConcurrency.ShouldBeLessThanOrEqualTo(3);
|
|
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// One node's backend throw does not poison a parallel batch: that node surfaces
|
|
/// <c>BadHistoryOperationUnsupported</c> while the other seven return Good (per-node isolation holds
|
|
/// under parallelism — 03/S3).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task One_node_failure_does_not_poison_a_parallel_batch()
|
|
{
|
|
var (host, server) = await BootAsync();
|
|
var nm = server.NodeManager!;
|
|
|
|
var src = DateTime.UtcNow.AddSeconds(-5);
|
|
var srv = DateTime.UtcNow;
|
|
// Node index 3's tagname throws; all others return a single good sample.
|
|
var fake = new PerTagnameFake(tagname => tagname == "WW.Tag3"
|
|
? throw new InvalidOperationException("backend boom for Tag3")
|
|
: new HistorianRead(new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, src, srv) }, null));
|
|
nm.HistorianDataSource = fake;
|
|
|
|
var nodeIds = MaterializeHistorizedNodes(nm, count: 8);
|
|
var details = new ReadRawModifiedDetails
|
|
{
|
|
StartTime = DateTime.UtcNow.AddHours(-1),
|
|
EndTime = DateTime.UtcNow,
|
|
NumValuesPerNode = 10,
|
|
IsReadModified = false,
|
|
};
|
|
|
|
var (results, errors) = await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
|
|
.WaitAsync(TimeSpan.FromSeconds(15), Ct);
|
|
|
|
for (var i = 0; i < 8; i++)
|
|
{
|
|
if (i == 3)
|
|
{
|
|
StatusCode.IsBad(errors[i].StatusCode).ShouldBeTrue();
|
|
errors[i].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
|
}
|
|
else
|
|
{
|
|
errors[i].StatusCode.Code.ShouldBe(StatusCodes.Good);
|
|
results[i].StatusCode.Code.ShouldBe(StatusCodes.Good);
|
|
}
|
|
}
|
|
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// A per-request deadline bounds a hung backend: reads that only complete on token cancellation
|
|
/// surface <c>BadTimeout</c> for every node and the call returns within the deadline (plus slack) —
|
|
/// it does not hang an SDK request thread indefinitely (03/S3).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Deadline_bounds_a_hung_backend()
|
|
{
|
|
var (host, server) = await BootAsync();
|
|
var nm = server.NodeManager!;
|
|
nm.HistoryReadDeadline = TimeSpan.FromMilliseconds(500);
|
|
nm.HistorianDataSource = new HangingHistorianDataSource();
|
|
|
|
var nodeIds = MaterializeHistorizedNodes(nm, count: 4);
|
|
var details = new ReadRawModifiedDetails
|
|
{
|
|
StartTime = DateTime.UtcNow.AddHours(-1),
|
|
EndTime = DateTime.UtcNow,
|
|
NumValuesPerNode = 10,
|
|
IsReadModified = false,
|
|
};
|
|
|
|
var (_, errors) = await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
|
|
.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
|
|
|
for (var i = 0; i < 4; i++)
|
|
errors[i].StatusCode.Code.ShouldBe(StatusCodes.BadTimeout);
|
|
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// A saturated process-wide batch limiter degrades a second concurrent HistoryRead to
|
|
/// <c>BadTooManyOperations</c> (rather than exhausting the SDK request-thread pool); once the parked
|
|
/// batch completes the slot frees and a later read succeeds (03/S3).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task Saturated_limiter_rejects_with_BadTooManyOperations()
|
|
{
|
|
var (host, server) = await BootAsync();
|
|
var nm = server.NodeManager!;
|
|
nm.MaxConcurrentHistoryReadBatches = 1;
|
|
nm.HistoryReadDeadline = TimeSpan.FromMilliseconds(200);
|
|
var fake = new GatedRawFake();
|
|
nm.HistorianDataSource = fake;
|
|
|
|
var ids = MaterializeHistorizedNodes(nm, count: 2);
|
|
var details = new ReadRawModifiedDetails
|
|
{
|
|
StartTime = DateTime.UtcNow.AddHours(-1),
|
|
EndTime = DateTime.UtcNow,
|
|
NumValuesPerNode = 10,
|
|
IsReadModified = false,
|
|
};
|
|
|
|
// First batch parks in the gated fake, holding the single limiter permit.
|
|
var first = Task.Run(() => InvokeHistoryRead(server, nm, details, ids[0]), Ct);
|
|
|
|
var spin = Stopwatch.StartNew();
|
|
while (fake.Entered == 0 && spin.Elapsed < TimeSpan.FromSeconds(5))
|
|
await Task.Delay(10, Ct);
|
|
fake.Entered.ShouldBe(1);
|
|
|
|
// Second concurrent batch: the limiter is saturated → all handles BadTooManyOperations.
|
|
var (_, errors2) = await Task.Run(() => InvokeHistoryRead(server, nm, details, ids[1]), Ct)
|
|
.WaitAsync(TimeSpan.FromSeconds(5), Ct);
|
|
errors2[0].StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations);
|
|
fake.Entered.ShouldBe(1, "the rejected batch must never reach the backend");
|
|
|
|
// Release the parked batch; the slot frees and a later read succeeds.
|
|
fake.Release();
|
|
await first.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
|
|
|
var (_, errors3) = await Task.Run(() => InvokeHistoryRead(server, nm, details, ids[1]), Ct)
|
|
.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
|
errors3[0].StatusCode.Code.ShouldBe(StatusCodes.Good);
|
|
|
|
await host.DisposeAsync();
|
|
}
|
|
|
|
/// <summary>Materialises <paramref name="count"/> historized Float variables and returns their NodeIds.</summary>
|
|
private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count)
|
|
{
|
|
var ids = new NodeId[count];
|
|
for (var i = 0; i < count; i++)
|
|
{
|
|
var key = $"eq/tag{i}";
|
|
nm.EnsureVariable(key, parentFolderNodeId: null, displayName: $"Tag{i}", dataType: "Float",
|
|
writable: false, historianTagname: $"WW.Tag{i}");
|
|
ids[i] = nm.TryGetVariable(key)!.NodeId;
|
|
}
|
|
|
|
return ids;
|
|
}
|
|
|
|
/// <summary>Issues one HistoryRead batch for all supplied node ids in order.</summary>
|
|
private static (IList<SdkHistoryReadResult> Results, IList<ServiceResult> Errors) InvokeHistoryRead(
|
|
OtOpcUaSdkServer server, OtOpcUaNodeManager nm, HistoryReadDetails details, params NodeId[] nodeIds)
|
|
{
|
|
var context = new OperationContext(
|
|
new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
|
|
|
var nodesToRead = nodeIds.Select(id => new HistoryReadValueId { NodeId = id }).ToList();
|
|
var results = Enumerable.Repeat<SdkHistoryReadResult>(null!, nodeIds.Length).ToList();
|
|
var errors = Enumerable.Repeat<ServiceResult>(null!, nodeIds.Length).ToList();
|
|
|
|
nm.HistoryRead(
|
|
context,
|
|
details,
|
|
TimestampsToReturn.Both,
|
|
releaseContinuationPoints: false,
|
|
nodesToRead,
|
|
results,
|
|
errors);
|
|
|
|
return (results, errors);
|
|
}
|
|
|
|
/// <summary>
|
|
/// A delaying fake historian source that tracks the maximum number of reads in flight at once
|
|
/// (via interlocked live-counter) so a test can prove per-node parallelism deterministically.
|
|
/// </summary>
|
|
private sealed class ConcurrencyTrackingHistorianDataSource(TimeSpan delay) : IHistorianDataSource
|
|
{
|
|
private int _current;
|
|
private int _max;
|
|
|
|
/// <summary>The highest number of concurrently in-flight reads observed.</summary>
|
|
public int MaxObservedConcurrency => Volatile.Read(ref _max);
|
|
|
|
private async Task<HistorianRead> TrackAsync(CancellationToken ct)
|
|
{
|
|
var now = Interlocked.Increment(ref _current);
|
|
UpdateMax(now);
|
|
try
|
|
{
|
|
await Task.Delay(delay, ct).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
Interlocked.Decrement(ref _current);
|
|
}
|
|
|
|
return new HistorianRead(
|
|
new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow) }, null);
|
|
}
|
|
|
|
private void UpdateMax(int observed)
|
|
{
|
|
int cur;
|
|
while ((cur = Volatile.Read(ref _max)) < observed &&
|
|
Interlocked.CompareExchange(ref _max, observed, cur) != cur)
|
|
{
|
|
}
|
|
}
|
|
|
|
public Task<HistorianRead> ReadRawAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
|
CancellationToken cancellationToken) => TrackAsync(cancellationToken);
|
|
|
|
public Task<HistorianRead> ReadProcessedAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
|
HistoryAggregateType aggregate, CancellationToken cancellationToken) => TrackAsync(cancellationToken);
|
|
|
|
public Task<HistorianRead> ReadAtTimeAsync(
|
|
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
|
TrackAsync(cancellationToken);
|
|
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
|
CancellationToken cancellationToken) =>
|
|
Task.FromResult(new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null));
|
|
|
|
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
|
|
|
public void Dispose()
|
|
{
|
|
}
|
|
}
|
|
|
|
/// <summary>A fake whose reads complete ONLY on token cancellation — used to prove the per-request
|
|
/// deadline surfaces <c>BadTimeout</c> rather than hanging.</summary>
|
|
private sealed class HangingHistorianDataSource : IHistorianDataSource
|
|
{
|
|
private static async Task<HistorianRead> HangAsync(CancellationToken ct)
|
|
{
|
|
await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
|
|
return new HistorianRead(Array.Empty<DataValueSnapshot>(), null);
|
|
}
|
|
|
|
public Task<HistorianRead> ReadRawAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
|
CancellationToken cancellationToken) => HangAsync(cancellationToken);
|
|
|
|
public Task<HistorianRead> ReadProcessedAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
|
HistoryAggregateType aggregate, CancellationToken cancellationToken) => HangAsync(cancellationToken);
|
|
|
|
public Task<HistorianRead> ReadAtTimeAsync(
|
|
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
|
HangAsync(cancellationToken);
|
|
|
|
public async Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false);
|
|
return new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null);
|
|
}
|
|
|
|
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
|
|
|
public void Dispose() { }
|
|
}
|
|
|
|
/// <summary>A fake whose raw read parks (awaits a <see cref="TaskCompletionSource"/>) until the test
|
|
/// releases it — used to hold the process-wide batch limiter permit while a second batch is issued.</summary>
|
|
private sealed class GatedRawFake : IHistorianDataSource
|
|
{
|
|
private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
private int _entered;
|
|
|
|
/// <summary>Number of raw reads that reached the backend body.</summary>
|
|
public int Entered => Volatile.Read(ref _entered);
|
|
|
|
/// <summary>Releases the parked read so its awaiting batch can complete.</summary>
|
|
public void Release() => _gate.TrySetResult();
|
|
|
|
public async Task<HistorianRead> ReadRawAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
Interlocked.Increment(ref _entered);
|
|
// Deliberately ignore the per-request deadline token: the first batch must keep HOLDING the
|
|
// single limiter permit until the test releases it, so the second batch deterministically hits
|
|
// a saturated limiter (rather than the first batch's own deadline freeing the slot early).
|
|
await _gate.Task.ConfigureAwait(false);
|
|
return new HistorianRead(
|
|
new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow) }, null);
|
|
}
|
|
|
|
public Task<HistorianRead> ReadProcessedAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
|
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
|
|
NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);
|
|
|
|
public Task<HistorianRead> ReadAtTimeAsync(
|
|
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
|
NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
|
|
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
|
CancellationToken cancellationToken) =>
|
|
NullHistorianDataSource.Instance.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);
|
|
|
|
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
|
|
|
public void Dispose() { }
|
|
}
|
|
|
|
/// <summary>A per-tagname fake that delegates each raw read to a caller-supplied func (which may throw
|
|
/// to simulate a per-node backend failure). Processed/AtTime/Events delegate to the null source.</summary>
|
|
private sealed class PerTagnameFake(Func<string, HistorianRead> rawHandler) : IHistorianDataSource
|
|
{
|
|
public Task<HistorianRead> ReadRawAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
|
CancellationToken cancellationToken) => Task.FromResult(rawHandler(fullReference));
|
|
|
|
public Task<HistorianRead> ReadProcessedAsync(
|
|
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
|
HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
|
|
NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);
|
|
|
|
public Task<HistorianRead> ReadAtTimeAsync(
|
|
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
|
NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
|
|
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(
|
|
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
|
CancellationToken cancellationToken) =>
|
|
NullHistorianDataSource.Instance.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);
|
|
|
|
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
|
|
|
public void Dispose() { }
|
|
}
|
|
|
|
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
|
|
{
|
|
var host = new OpcUaApplicationHost(
|
|
new OpcUaApplicationHostOptions
|
|
{
|
|
ApplicationName = "OtOpcUa.HistoryReadConcTest",
|
|
ApplicationUri = $"urn:OtOpcUa.HistoryReadConcTest:{Guid.NewGuid():N}",
|
|
OpcUaPort = AllocateFreePort(),
|
|
PublicHostname = "localhost",
|
|
PkiStoreRoot = _pkiRoot,
|
|
},
|
|
NullLogger<OpcUaApplicationHost>.Instance);
|
|
|
|
var server = new OtOpcUaSdkServer();
|
|
await host.StartAsync(server, Ct);
|
|
return (host, server);
|
|
}
|
|
|
|
private static int AllocateFreePort()
|
|
{
|
|
using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
|
|
listener.Start();
|
|
var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
|
|
listener.Stop();
|
|
return port;
|
|
}
|
|
|
|
/// <summary>Cleans up the PKI root directory.</summary>
|
|
public void Dispose()
|
|
{
|
|
if (Directory.Exists(_pkiRoot))
|
|
{
|
|
try { Directory.Delete(_pkiRoot, recursive: true); }
|
|
catch { /* best-effort cleanup */ }
|
|
}
|
|
}
|
|
}
|