diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs index 8e58c466..25fc806f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs @@ -57,6 +57,21 @@ public sealed class ContinuousHistorizationOptions /// Steady drain cadence in seconds (also the post-success reschedule). Defaults to 2. public double DrainIntervalSeconds { get; init; } = 2; + /// + /// Maximum historian tags a single drain pass writes concurrently. Defaults to 8; clamped to at + /// least 1 by the recorder (1 restores the fully sequential drain). + /// + /// One peeked batch is grouped per historian tag and each group is an independent gateway + /// round-trip, so a sequential pass costs tags × latency — with a wide batch that is the + /// dominant term in how fast the outbox can drain. The window bounds how many of those + /// round-trips are outstanding at once, so a large tag count cannot flood the gateway. Nothing + /// about the ack changes: the pass is still acked all-or-nothing after every group has + /// reported, and the Gitea #538 per-tag quarantine still keys on "another group wrote this + /// pass". + /// + /// + public int DrainConcurrency { get; init; } = 8; + /// /// Maximum un-acked outbox entries before the drop-oldest capacity policy kicks in; 0 /// means unbounded. @@ -102,6 +117,8 @@ public sealed class ContinuousHistorizationOptions warnings.Add("ContinuousHistorization:OutboxPath is empty while historization is enabled — the durable outbox has no directory to persist to; host startup will fail when the outbox is constructed. Set an absolute path on durable storage."); if (string.Equals(CommitMode, "Periodic", StringComparison.OrdinalIgnoreCase) && CommitIntervalMs <= 0) warnings.Add($"ContinuousHistorization:CommitIntervalMs is {CommitIntervalMs} — must be > 0 in Periodic commit mode; the periodic-commit loop cannot run."); + if (DrainConcurrency < 1) + warnings.Add($"ContinuousHistorization:DrainConcurrency is {DrainConcurrency} — must be >= 1; the recorder clamps it to 1 (a fully sequential drain)."); return warnings; } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs index 14fef1f2..da8b0fa4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs @@ -135,6 +135,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// writing successfully — before its queued entries are abandoned so the FIFO head can advance. private readonly int _maxWriteAttempts; + /// Max per-tag groups one drain pass writes concurrently. The groups in a pass are + /// independent gateway round-trips, so a sequential pass costs tags × latency; this bounds how + /// many are outstanding at once so a wide batch cannot flood the gateway. + private readonly int _drainConcurrency; + /// Consecutive failed write passes per historian tag, counted only across passes that /// proved the backend reachable. Reset on the tag's first success. Mutated exclusively inside /// , which keeps single-flight. @@ -174,6 +179,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// and for tests. /// Gitea #538: consecutive failed write passes one tag may /// accumulate, while other tags are succeeding, before its entries are abandoned. + /// Max historian tags one drain pass writes concurrently (must be + /// positive; 1 is the fully sequential drain). /// Props for the actor. public static Props Props( IActorRef dependencyMux, @@ -185,10 +192,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers TimeSpan? minBackoff = null, TimeSpan? maxBackoff = null, Func? captureGate = null, - int maxWriteAttempts = 5) => + int maxWriteAttempts = 5, + int drainConcurrency = 8) => Akka.Actor.Props.Create(() => new ContinuousHistorizationRecorder( dependencyMux, writer, outbox, historizedRefs, drainBatchSize, drainInterval, minBackoff, maxBackoff, - captureGate, maxWriteAttempts)); + captureGate, maxWriteAttempts, drainConcurrency)); /// Initializes a new instance of the class. /// The per-node dependency mux to register historized-ref interest with. @@ -204,6 +212,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers /// Gitea #538: consecutive failed write passes one tag may /// accumulate, while other tags are succeeding, before its entries are abandoned. Clamped to at /// least 1. + /// Max historian tags one drain pass writes concurrently. Clamped to + /// at least 1; 1 is the fully sequential drain. public ContinuousHistorizationRecorder( IActorRef dependencyMux, IHistorianValueWriter writer, @@ -214,7 +224,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers TimeSpan? minBackoff = null, TimeSpan? maxBackoff = null, Func? captureGate = null, - int maxWriteAttempts = 5) + int maxWriteAttempts = 5, + int drainConcurrency = 8) { _dependencyMux = dependencyMux ?? throw new ArgumentNullException(nameof(dependencyMux)); _writer = writer ?? throw new ArgumentNullException(nameof(writer)); @@ -238,6 +249,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers // gate), and a test that does not care about the role. _captureGate = captureGate ?? (static () => true); _maxWriteAttempts = maxWriteAttempts > 0 ? maxWriteAttempts : 1; + _drainConcurrency = drainConcurrency > 0 ? drainConcurrency : 1; ReceiveAsync(OnValueChangedAsync); ReceiveAsync(OnValueBatchAsync); @@ -537,35 +549,68 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers // id, so partial per-tag acking could drop un-written older entries; the ack stays // all-or-nothing — only when nothing is left blocking do we truncate the whole peeked // prefix (one commit). - var succeeded = 0; - var failedGroups = new List<(string Tag, int Count)>(); - foreach (IGrouping group in - batch.GroupBy(e => e.Tag, StringComparer.Ordinal)) - { - var values = group - .Select(e => new HistorizationValue(e.TimestampUtc, e.NumericValue, e.Quality)) - .ToList(); + // + // The groups are INDEPENDENT gateway round-trips against distinct historian tags, so they run + // concurrently under a bounded window rather than one after another: a sequential pass costs + // `tags × latency`, which is the dominant term in how fast a wide batch can drain. The window + // (_drainConcurrency) is what stops a large tag count from flooding the gateway; a window of 1 + // is exactly the old sequential loop. + var groups = batch + .GroupBy(e => e.Tag, StringComparer.Ordinal) + .Select(g => ( + Key: g.Key, + Values: (IReadOnlyList)g + .Select(e => new HistorizationValue(e.TimestampUtc, e.NumericValue, e.Quality)) + .ToList())) + .ToList(); - bool ok; + using var window = new SemaphoreSlim(_drainConcurrency); + var tasks = groups.Select(async g => + { + await window.WaitAsync(ct).ConfigureAwait(false); try { - ok = await _writer.WriteLiveValuesAsync(group.Key, values, ct).ConfigureAwait(false); - } - catch - { - // The writer is non-throwing by contract; guard defensively so one tag's fault - // cannot fail the whole drain task. - ok = false; - } + bool ok; + try + { + ok = await _writer.WriteLiveValuesAsync(g.Key, g.Values, ct).ConfigureAwait(false); + } + catch + { + // The writer is non-throwing by contract; guard defensively so one tag's fault + // cannot fail the whole drain task (and, now, cannot fault the WhenAll either). + ok = false; + } + return (Tag: g.Key, Count: g.Values.Count, Ok: ok); + } + finally + { + window.Release(); + } + }).ToList(); + + // Task.WhenAll waits for EVERY task to complete before it faults, so no lambda can still be + // touching `window` when the using-scope disposes it — including on the cancellation path, + // where the not-yet-entered tasks fault out of WaitAsync without reaching the finally. + (string Tag, int Count, bool Ok)[] outcomes = await Task.WhenAll(tasks).ConfigureAwait(false); + + // Fold the outcomes on the single logical drain flow — NEVER inside the parallel lambdas. + // _tagWriteFailures is actor state (kept single-flight by _draining); mutating it from N + // concurrent tasks would corrupt the Gitea #538 counters. Nothing below observes completion + // ORDER, so the fold is deterministic whatever order the writes finished in. + var succeeded = 0; + var failedGroups = new List<(string Tag, int Count)>(); + foreach ((string tag, int count, bool ok) in outcomes) + { if (ok) { succeeded++; - _tagWriteFailures.Remove(group.Key); + _tagWriteFailures.Remove(tag); } else { - failedGroups.Add((group.Key, values.Count)); + failedGroups.Add((tag, count)); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 98daf31a..a5601f48 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -477,7 +477,10 @@ public static class ServiceCollectionExtensions captureGate: resolver.GetService() is { } histRoleView ? () => histRoleView.ShouldRecordHistorizedValues : null, - maxWriteAttempts: continuousOptions.MaxWriteAttempts), + maxWriteAttempts: continuousOptions.MaxWriteAttempts, + // Bounded-parallel per-tag drain: the groups in one pass are independent + // gateway round-trips, so a sequential pass costs `tags × latency`. + drainConcurrency: continuousOptions.DrainConcurrency), ContinuousHistorizationRecorderActorName); registry.Register(continuousRecorder); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs index 78dc4e51..221f7bfa 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs @@ -607,6 +607,189 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit }, TimeSpan.FromSeconds(10)); } + // ---------------- Bounded-parallel drain (DrainConcurrency) ---------------- + + /// + /// A drain pass writes distinct historian tags CONCURRENTLY, bounded by DrainConcurrency. + /// + /// + /// The tags in one peeked batch are independent gateway round-trips, so a sequential foreach makes + /// a pass cost tags × latency. The load-bearing observation is a POSITIVE one — two writes + /// genuinely in flight at the same instant — which is why the writer double blocks every call on a + /// TCS and records the peak in-flight count. Measuring wall-clock time instead would be a flake + /// generator and would not distinguish "parallel" from "fast". + /// + /// The values arrive as ONE deliberately: + /// the per-value handler nudges a DrainTick after each append, so three separate messages could + /// start the drain on a batch of one and the pass would hold a single group — nothing to + /// parallelise. A batch message appends all three before any DrainTick is dequeued. + /// + /// + [Fact] + public async Task Drain_writes_distinct_tags_concurrently_within_the_window() + { + var mux = CreateTestProbe(); + var writer = new GatingValueWriter(); + var outbox = new InMemoryOutbox(); + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" }, + drainInterval: TimeSpan.FromMilliseconds(50), + drainConcurrency: 8)); + + rec.Tell(new VirtualTagActor.DependencyValueBatch(new[] + { + new VirtualTagActor.DependencyValueChanged("A", 1.0, DateTime.UtcNow), + new VirtualTagActor.DependencyValueChanged("B", 2.0, DateTime.UtcNow), + new VirtualTagActor.DependencyValueChanged("C", 3.0, DateTime.UtcNow), + })); + + await AwaitAssertAsync( + () => + { + Assert.True( + writer.MaxObservedConcurrency >= 2, + $"expected overlapping writes; peak in-flight was {writer.MaxObservedConcurrency}"); + return Task.CompletedTask; + }, + TimeSpan.FromSeconds(10)); + + writer.Release(); + + // …and the pass still acks all-or-nothing: the whole peeked prefix leaves the outbox. + await AwaitAssertAsync(async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(0, status.QueuedDepth); + Assert.Equal(0, status.AbandonedEntries); + }, TimeSpan.FromSeconds(10)); + } + + /// + /// Falsification control: DrainConcurrency = 1 is exactly the old sequential drain. + /// + /// + /// Same fixture as the test above, so the only difference is the window size. Without this an + /// implementation that ignored the option entirely (always unbounded) would pass the concurrency + /// test and nothing would pin the bound. + /// + [Fact] + public async Task Drain_concurrency_of_one_is_sequential() + { + var mux = CreateTestProbe(); + var writer = new GatingValueWriter(); + var outbox = new InMemoryOutbox(); + + IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props( + mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" }, + drainInterval: TimeSpan.FromMilliseconds(50), + drainConcurrency: 1)); + + rec.Tell(new VirtualTagActor.DependencyValueBatch(new[] + { + new VirtualTagActor.DependencyValueChanged("A", 1.0, DateTime.UtcNow), + new VirtualTagActor.DependencyValueChanged("B", 2.0, DateTime.UtcNow), + new VirtualTagActor.DependencyValueChanged("C", 3.0, DateTime.UtcNow), + })); + + // The first write is in flight and blocked… + await AwaitAssertAsync( + () => + { + Assert.True(writer.CallCount >= 1, "the drain must have started a write"); + return Task.CompletedTask; + }, + TimeSpan.FromSeconds(10)); + + // …and stays alone: with a window of 8 the other two groups would have entered by now. + await Task.Delay(TimeSpan.FromMilliseconds(300)); + Assert.Equal(1, writer.MaxObservedConcurrency); + Assert.Equal(1, writer.CallCount); + + writer.Release(); + + await AwaitAssertAsync(async () => + { + var status = await rec.Ask( + ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3)); + Assert.Equal(0, status.QueuedDepth); + }, TimeSpan.FromSeconds(10)); + Assert.Equal(1, writer.MaxObservedConcurrency); + } + + /// + /// double that BLOCKS every call until , + /// recording the peak number of calls in flight simultaneously. That peak is what distinguishes a + /// bounded-parallel drain from a sequential one; a writer that returns synchronously (like + /// ) can never observe overlap however parallel the caller is. + /// + private sealed class GatingValueWriter : IHistorianValueWriter + { + private readonly Lock _gate = new(); + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _inFlight; + private int _maxObserved; + private int _calls; + + /// Peak number of calls observed in flight at once. + public int MaxObservedConcurrency + { + get + { + lock (_gate) + { + return _maxObserved; + } + } + } + + /// Lifetime count of invocations. + public int CallCount + { + get + { + lock (_gate) + { + return _calls; + } + } + } + + /// Unblocks every pending and future call. + public void Release() => _release.TrySetResult(); + + public async Task WriteLiveValuesAsync( + string tag, IReadOnlyList values, CancellationToken ct) + { + lock (_gate) + { + _calls++; + _inFlight++; + if (_inFlight > _maxObserved) + { + _maxObserved = _inFlight; + } + } + + try + { + // A bounded wait rather than an unbounded one: a regression that never releases fails the + // assertion rather than hanging the whole test run. + await _release.Task.WaitAsync(TimeSpan.FromSeconds(30), ct).ConfigureAwait(false); + } + finally + { + lock (_gate) + { + _inFlight--; + } + } + + return true; + } + } + /// In-memory double: records every value written and /// returns as the ack. Thread-safe — the recorder drains off the actor thread. private sealed class FakeValueWriter : IHistorianValueWriter