From f4907d098c4a52913fd560dbc45418741363d35c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 17:23:46 -0400 Subject: [PATCH] perf(historian): scan the historization outbox outside its state lock (lmxopcua#746) FasterLogHistorizationOutbox.PeekBatchAsync held the state lock across the FasterLog scan. Once a backlog's head ages out of FasterLog's in-memory window (any multi-minute historian outage), GetNext reads pages from disk synchronously, and the recorder's appends (which index under the same lock) blocked on an Akka dispatcher thread once per drain pass, exactly during outage recovery. The peek now snapshots the head under the lock, scans unlocked, and re-takes the lock only when it met undecodable records, to reconcile them with the index in O(records scanned) and drop one still at the head (re-checked, since an ack or drop-oldest may have moved it). A test seam pauses the scan and proves an append completes meanwhile; wrapping the scan in the lock again fails it. Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22 in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV --- .../Recorder/FasterLogHistorizationOutbox.cs | 79 +++++++++++++------ .../FasterLogHistorizationOutboxTests.cs | 27 +++++++ 2 files changed, 81 insertions(+), 25 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs index d3b1bdbcd..bc97933b5 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs @@ -206,47 +206,73 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox // here (truncate + commit), because nothing else ever would: a FIFO ack only passes records // older than the acked entry, so a corrupt record with no readable entry behind it would // otherwise pin the head (and the outbox count) forever. + // + // The scan runs OUTSIDE _state: once a backlog's head has aged out of FasterLog's in-memory window (any multi-minute + // historian outage) GetNext reads pages from the device synchronously, and holding the lock across that blocked the + // recorder's appends (index bookkeeping takes _state) on an Akka dispatcher thread, once per drain pass, exactly + // during outage recovery. The lock is taken only to snapshot the head and, when the scan met undecodable records, to + // reconcile them with the index (O(records scanned), never O(backlog)). public async ValueTask> PeekBatchAsync(int max, CancellationToken ct) { ArgumentOutOfRangeException.ThrowIfNegativeOrZero(max); - var batch = new List(Math.Min(max, 64)); - long? droppedHeadTo = null; + long from, tail; lock (_state) { - // _live and the scan are both in log order, so a forward cursor pairs each scanned record - // with its index node (by start address) in O(n) overall. - LinkedListNode? cursor = _live.First; - using FasterLogScanIterator iter = _log.Scan(_nextScanAddress, _log.TailAddress, recover: false); + from = _nextScanAddress; + tail = _log.TailAddress; + } + + var batch = new List(Math.Min(max, 64)); + List<(long Address, long Next, bool LeadsBatch)>? unreadable = null; + using (FasterLogScanIterator iter = _log.Scan(from, tail, recover: false)) + { while (batch.Count < max && iter.GetNext(out byte[] bytes, out int len, out long address, out long next)) { - while (cursor is not null && cursor.Value.Start < address) - cursor = cursor.Next; - LinkedListNode? node = cursor is not null && cursor.Value.Start == address ? cursor : null; - + ScanStepForTests?.Invoke(); if (HistorizationOutboxEntrySerializer.TryDeserialize(bytes.AsSpan(0, len), out HistorizationOutboxEntry? entry)) { batch.Add(entry); continue; } + (unreadable ??= []).Add((address, next, batch.Count == 0)); + } + } - // Count once: a node flagged at recovery (or by an earlier peek) is not re-counted. - // An unindexed record cannot be flagged, so it would count per peek — but only this - // process's own (readable) appends are ever unindexed, transiently. - if (node is null || !node.Value.Unreadable) + long? droppedHeadTo = null; + if (unreadable is not null) + { + lock (_state) + { + // _live and the scan are both in log order, so a forward cursor pairs each unreadable record with its + // index node (by start address). + LinkedListNode? cursor = _live.First; + foreach ((long address, long next, bool leadsBatch) in unreadable) { - _unreadableCount++; - if (node is not null) - node.Value = node.Value with { Unreadable = true }; - } + while (cursor is not null && cursor.Value.Start < address) + cursor = cursor.Next; + LinkedListNode? node = cursor is not null && cursor.Value.Start == address ? cursor : null; - if (batch.Count == 0 && node is not null && ReferenceEquals(node, _live.First)) - { - cursor = node.Next; - _index.Remove(node.Value.Id); - _live.RemoveFirst(); - _nextScanAddress = next; - droppedHeadTo = next; + // Count once: a node flagged at recovery (or by an earlier peek) is not re-counted. An unindexed record + // cannot be flagged, so it would count per peek — but only this process's own (readable) appends are + // ever unindexed, transiently. + if (node is null || !node.Value.Unreadable) + { + _unreadableCount++; + if (node is not null) + node.Value = node.Value with { Unreadable = true }; + } + + // Drop an unreadable record still at the HEAD, ahead of every readable entry (nothing else ever would). + // Re-checked under the lock: an ack or a drop-oldest may have moved the head while we scanned. + if (leadsBatch && node is not null && ReferenceEquals(node, _live.First)) + { + cursor = node.Next; + _index.Remove(node.Value.Id); + _live.RemoveFirst(); + _nextScanAddress = next; + droppedHeadTo = next; + } } } } @@ -260,6 +286,9 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox return batch; } + /// Test seam: invoked once per record the peek scans, OUTSIDE the state lock. + internal Action? ScanStepForTests { get; set; } + /// public async ValueTask RemoveAsync(Guid id, CancellationToken ct) { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/FasterLogHistorizationOutboxTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/FasterLogHistorizationOutboxTests.cs index 03c5a9c5f..e6826aa2a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/FasterLogHistorizationOutboxTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/FasterLogHistorizationOutboxTests.cs @@ -27,6 +27,33 @@ public sealed class FasterLogHistorizationOutboxTests : IDisposable private static HistorizationOutboxEntry E(string tag, double v) => new(Guid.NewGuid(), tag, HistorianValue.FromDouble(v), 0u, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc)); + [Fact] + public async Task A_peek_scan_does_not_hold_the_state_lock_so_appends_proceed_meanwhile() + { + // The scan may read pages from disk (a backlog aged out of the in-memory window); holding the state lock across it + // blocked the recorder's appends on an Akka dispatcher thread. + var dir = NewTempDir(); + using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry); + for (var i = 0; i < 5; i++) await o.AppendAsync(E("A", i), TestContext.Current.CancellationToken); + + bool? appendCompletedDuringScan = null; + o.ScanStepForTests = () => + { + if (appendCompletedDuringScan is not null) return; + var append = Task.Run(async () => await o.AppendAsync(E("B", 99), CancellationToken.None)); + appendCompletedDuringScan = append.Wait(TimeSpan.FromSeconds(5)); + }; + + var batch = await o.PeekBatchAsync(10, TestContext.Current.CancellationToken); + + Assert.True(appendCompletedDuringScan, "an append must not wait for a peek's scan"); + Assert.Equal(5, batch.Count(e => e.Tag == "A")); + Assert.Equal(6, await o.CountAsync(TestContext.Current.CancellationToken)); + // The concurrent append is FIFO-tracked: the next peek returns it after the five. + o.ScanStepForTests = null; + Assert.Equal("B", (await o.PeekBatchAsync(10, TestContext.Current.CancellationToken))[^1].Tag); + } + [Fact] public async Task Append_then_peek_returns_fifo() {