diff --git a/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json b/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json index a538f66f..4eb04faf 100644 --- a/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json +++ b/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json @@ -11,7 +11,7 @@ { "id": "R2-08-T8", "subject": "ServerHistorianOptions: HistoryReadBatchParallelism/MaxConcurrentHistoryReadBatches/HistoryReadDeadline + Validate warnings + node-manager properties + hosted-service wiring", "status": "completed", "blockedBy": [] }, { "id": "R2-08-T9", "subject": "Async-ify ServeNode + Processed/AtTime arms (ct threading, OCE=>BadTimeout, behavior-neutral)", "status": "completed", "blockedBy": ["R2-08-T7", "R2-08-T8"] }, { "id": "R2-08-T10", "subject": "Async-ify ServeRawPaged (page read + tie-cluster over-fetch) + extract ServeEventsAsync", "status": "completed", "blockedBy": ["R2-08-T9"] }, - { "id": "R2-08-T11", "subject": "RunBounded fan-out in all four HistoryRead arms, single bridge per arm (turns T7 green) + upper-bound + isolation tests", "status": "pending", "blockedBy": ["R2-08-T9", "R2-08-T10"] }, + { "id": "R2-08-T11", "subject": "RunBounded fan-out in all four HistoryRead arms, single bridge per arm (turns T7 green) + upper-bound + isolation tests", "status": "completed", "blockedBy": ["R2-08-T9", "R2-08-T10"] }, { "id": "R2-08-T12", "subject": "Per-request HistoryRead deadline CTS — hung backend surfaces BadTimeout per node", "status": "pending", "blockedBy": ["R2-08-T11"] }, { "id": "R2-08-T13", "subject": "Process-wide concurrent-batch limiter — saturation degrades to BadTooManyOperations (lazy gate, bounded wait)", "status": "pending", "blockedBy": ["R2-08-T12"] }, { "id": "R2-08-T14", "subject": "Docs (Historian.md + CLAUDE.md ServerHistorian table + security.md) + full touched-suite and whole-solution regression sweep + STATUS.md entry", "status": "pending", "blockedBy": ["R2-08-T5", "R2-08-T6", "R2-08-T13"] } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index e6e8b0d4..7539af9c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1816,6 +1816,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 IDictionary cache) { var session = context.OperationContext?.Session; + var work = new List>(nodesToProcess.Count); foreach (var handle in nodesToProcess) { if (details.IsReadModified) @@ -1825,12 +1826,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 continue; } - // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice). - ServeRawPagedAsync( + work.Add(() => ServeRawPagedAsync( handle, session, nodesToRead, results, errors, details.StartTime, details.EndTime, details.NumValuesPerNode, - CancellationToken.None).GetAwaiter().GetResult(); + CancellationToken.None)); } + + RunBounded(work, HistoryReadBatchParallelism); } /// @@ -1846,6 +1848,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { // OPC UA ProcessingInterval is a Duration in milliseconds — convert once per batch. var interval = TimeSpan.FromMilliseconds(details.ProcessingInterval); + var work = new List>(nodesToProcess.Count); foreach (var handle in nodesToProcess) { // AggregateType is a per-node parallel collection (same length as nodesToRead, enforced by @@ -1863,15 +1866,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // and the single-shot backend returns every bucket in one read, so there is no "full page ⇒ // maybe more" signal to page on. Returning the complete aggregate result with a null CP is // spec-conformant (OPC UA Part 11 lets a server return all available data in one response). - // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice). - ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadProcessedAsync( + var agg = aggregate.Value; + work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadProcessedAsync( tagname, details.StartTime, details.EndTime, interval, - aggregate.Value, - token), CancellationToken.None).GetAwaiter().GetResult(); + agg, + token), CancellationToken.None)); } + + RunBounded(work, HistoryReadBatchParallelism); } /// @@ -1887,14 +1892,16 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { // Snapshot the requested timestamps once — the same list is read for every node. var timestamps = details.ReqTimes?.ToList() ?? new List(); + var work = new List>(nodesToProcess.Count); foreach (var handle in nodesToProcess) { - // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice). - ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync( + work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync( tagname, timestamps, - token), CancellationToken.None).GetAwaiter().GetResult(); + token), CancellationToken.None)); } + + RunBounded(work, HistoryReadBatchParallelism); } /// @@ -1911,6 +1918,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // Snapshot the select clauses once — the same filter projects every node's events. var selectClauses = details.Filter?.SelectClauses ?? new SimpleAttributeOperandCollection(); + var work = new List>(nodesToProcess.Count); foreach (var handle in nodesToProcess) { var idString = handle.NodeId.Identifier?.ToString(); @@ -1924,10 +1932,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 continue; } - // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice). - ServeEventsAsync(handle, sourceName, details, selectClauses, results, errors, CancellationToken.None) - .GetAwaiter().GetResult(); + var sn = sourceName; + work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, results, errors, CancellationToken.None)); } + + RunBounded(work, HistoryReadBatchParallelism); } /// @@ -2099,6 +2108,52 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// The service-level errors list to fill at handle.Index. /// Invokes the resolved data-source read with the resolved tagname; only called /// once the tagname is confirmed present. + /// + /// Run a batch of per-node HistoryRead work items with BOUNDED parallelism (arch-review 03/S3), + /// block-bridged ONCE at the arm boundary. Contract: every work item must be fully + /// self-contained — it fills its own results/errors slot and NEVER throws (all per-node handlers + /// catch every exception, including cancellation, and surface a Bad status for that node only). That + /// invariant is what makes the single GetAwaiter().GetResult() safe: + /// can never fault, so it cannot throw out of the synchronous SDK override. A degree ≤ 1 (or a + /// single item) runs sequentially — no semaphore, no extra tasks. + /// + /// The per-node work items (index-disjoint writes into the shared results/errors). + /// Max concurrent work items (). + private static void RunBounded(IReadOnlyList> work, int degree) + { + if (work.Count == 0) return; + if (degree <= 1 || work.Count == 1) + { + foreach (var item in work) + item().GetAwaiter().GetResult(); + return; + } + + using var gate = new SemaphoreSlim(degree); + var tasks = new Task[work.Count]; + for (var i = 0; i < work.Count; i++) + tasks[i] = RunGatedAsync(gate, work[i]); + + // Safe: every body is self-contained + never throws, so WhenAll never faults (see contract above). + Task.WhenAll(tasks).GetAwaiter().GetResult(); + } + + /// Awaits one work item under a semaphore permit, releasing it when the item completes. + /// The concurrency-limiting semaphore. + /// The per-node work item. + private static async Task RunGatedAsync(SemaphoreSlim gate, Func work) + { + await gate.WaitAsync().ConfigureAwait(false); + try + { + await work().ConfigureAwait(false); + } + finally + { + gate.Release(); + } + } + private async Task ServeNodeAsync( NodeHandle handle, IList results, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs index 37599d8a..78bccd9a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs @@ -61,6 +61,85 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable await host.DisposeAsync(); } + /// + /// Parallelism is bounded ABOVE by HistoryReadBatchParallelism: with the property set to 3 and + /// 12 nodes in flight, no more than 3 reads are ever concurrent (03/S3). + /// + [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(); + } + + /// + /// One node's backend throw does not poison a parallel batch: that node surfaces + /// BadHistoryOperationUnsupported while the other seven return Good (per-node isolation holds + /// under parallelism — 03/S3). + /// + [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(); + } + /// Materialises historized Float variables and returns their NodeIds. private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count) { @@ -161,6 +240,33 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable } } + /// 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. + private sealed class PerTagnameFake(Func rawHandler) : IHistorianDataSource + { + public Task ReadRawAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) => Task.FromResult(rawHandler(fullReference)); + + public Task ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => + NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken); + + public Task 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(