From 37a2a831f3d84552d299709100ecc85f3a009018 Mon Sep 17 00:00:00 2001 From: romanett Date: Mon, 19 Oct 2026 06:48:41 +0200 Subject: [PATCH] Server: keep the publish timer on a fixed cadence and parallelize the sweep by work (#4829) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The publish timer loop ran a pass over all session publish queues and then waited a full PublishingResolution, so the cycle period was pass time + resolution and a slow pass delayed every following cycle. Whether the pass ran in parallel depended only on the session count (256 or more). Fixed cadence: the loop now waits only for what is left of the period (RunAtFixedPeriodAsync on the server TimeProvider). A pass that overruns the period is followed by the next pass at once and the cadence restarts from there; missed periods are not caught up with extra passes. Shutdown, cancellation and abandoned subscription handling are unchanged. Parallelism by work: a pass checks the queues on the sweep thread until it has used a quarter of the publishing resolution (25 ms at 100 ms), then checks the remaining queues with Parallel.For (CheckPublishTimers). The session count no longer decides. Why the budget: - The pass cost is real per-item work. In the LoadTest subscribe shape (50 sessions x 15 subscriptions x 600 items, all 600 written every 150 ms, publish 100 ms) every pass promotes up to 450k items; the sequential pass takes 85-125 ms (~0.28 us per promoted item: ready list, IsReadyToReport, LinkedList move, trigger check, ready signal release). Ready-list promotion already skips items without pending values and subscriptions whose timer is not due, so there is no idle work left to skip; only parallelism makes the pass fit the period. - Fixed cadence alone lifts delivery from 64 % to 91 % (pass ~85 ms in a 100 ms period, no headroom); with the parallel remainder 97-98 %. - Parallel sweeps cost CPU when the pass fits anyway: at 10 sessions the sequential pass takes ~15 ms; a 2 ms budget made it parallel and raised server CPU per notification from 5.0 to 8.9 us. With the 25 ms budget it stays sequential. - Idle servers stay sequential: 1000 sessions x 1 slow subscription pass in ~2 ms sequential (was always parallel above 256 sessions); 2 sessions in 0.05 ms. Diagnostic probes (one build with temporary switches, one run each, 50x15x600 shape): | loop / sweep | delivered | us/notif | pass avg | cycle | |-------------------------|-----------|----------|----------|--------| | old loop, sequential | 63.9 % | 5.1 | ~125 ms | ~225 ms| | fixed, sequential | 91.2 % | 5.1 | ~85 ms | ~110 ms| | old loop, all parallel | 96.9 % | 3.9 | ~24 ms | ~132 ms| | fixed, all parallel | 97.0 % | 4.1 | ~21 ms | 100 ms | | fixed, 25 ms budget | 97.8 % | 4.4 | ~44 ms | 100 ms | Paired A/B against master (perf/ServerLoadHarness, Server GC, encrypted, alternating rounds, medians; runs repeated when processes outside the harness used more than 1.5 cores): | shape (sessions x subs x items) | n | delivered | us/notif | server cores | |-----------------------------------|---|--------------|-------------|--------------| | 50 x 15 x 600, pub 100, write 150 | 3 | 67.7 -> 97.8 | 5.5 -> 4.4 | 11.3 -> 13.3 | | 10 x 15 x 600, pub 100, write 150 | 3 | 100.4 -> 99.9| 5.5 -> 5.1 | 3.32 -> 3.11 | | 1000 x 1 x 1, pub 1000, write 500 | 7 | 50 -> 50 (*) | 216 -> 201 | 0.21 -> 0.20 | | 2 x 1 x 100, pub 100, write 150 | 3 | 100 -> 100 | 48.5 -> 46.1| 0.06 -> 0.06 | (*) two writes per publishing interval with queue size 1, by design. Side effect at 50 sessions: server allocation per notification rises from ~54 to ~96 B. An allocation trace attributes it to MonitoredItemNotification: the parallel remainder makes all subscriptions publish within a few ms, the in-flight burst exceeds the notification pool bound (kMaxPooledNotifications = 65536) and the pool misses; the slow sequential sweep spread gets and returns. Gen0 GCs in the 15 s window 3-5 -> 8, pause 20-50 -> ~60 ms; CPU per notification still drops. Addressed by the stacked #4830, which allocates these notifications instead of pooling them (tracking issue #4819). ### Follow-up commits - **Late wake / exact deadline** (review): a wait that ends a whole period late restarts the cadence from the late pass (no catch-up pass), and a sweep exactly at a subscription's deadline counts as due. - **Phase aliasing with equal periods** (f33893b7c): with the sweep at the publishing interval, a subscription's deadline keeps one phase relative to the sweep for the whole session. If it falls inside the ~16 ms timer jitter, a late sweep followed by an on-time one dropped a publishing cycle. The 2-session shape (write = publish = 100 ms, queue size 1) was bimodal: 6 of 24 runs at 56-85 %. The deadline now advances by one interval per expiry and skips missed expiries only when a whole interval behind after that. Result: 24 of 24 runs at 97-101 % (master: steady 88-98 %); write 150 ms gave 98.5-100 %, and the 50 x 15 x 600 shape stayed at 99.3-100.7 %. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../Subscription/Subscription.cs | 21 +- .../Subscription/SubscriptionManager.cs | 174 ++++++++--- .../Subscription/PublishTimerCadenceTests.cs | 272 ++++++++++++++++++ ...SubscriptionStateMachineRegressionTests.cs | 85 ++++++ 4 files changed, 504 insertions(+), 48 deletions(-) create mode 100644 tests/Opc.Ua.Server.Tests/Subscription/PublishTimerCadenceTests.cs diff --git a/src/Opc.Ua.Server/Subscription/Subscription.cs b/src/Opc.Ua.Server/Subscription/Subscription.cs index 4fbb768d04..4ada9c59e7 100644 --- a/src/Opc.Ua.Server/Subscription/Subscription.cs +++ b/src/Opc.Ua.Server/Subscription/Subscription.cs @@ -831,8 +831,9 @@ namespace Opc.Ua.Server long currentTime = m_timeProvider.GetTimestampMilliseconds(); - // check if publish interval has elapsed. - if (m_publishTimerExpiry >= currentTime) + // check if publish interval has elapsed; a sweep that runs exactly at the + // deadline counts as elapsed. + if (m_publishTimerExpiry > currentTime) { // check if waiting for publish. if (m_waitingForPublish) @@ -843,10 +844,20 @@ namespace Opc.Ua.Server return PublishingState.Idle; } - // set next expiry time. - while (m_publishTimerExpiry < currentTime) + // set next expiry time. The publish timer sweep runs on its own cadence, and + // when it runs at the publishing interval a sweep that wakes a little late + // can find the subscription a whole interval behind. Advancing by one + // interval keeps that expiry due for the next sweep instead of dropping a + // publishing cycle; only a subscription that is still behind by a whole + // interval after that (a stalled sweep) skips the missed expiries. + long publishingInterval = (long)m_publishingInterval; + m_publishTimerExpiry += publishingInterval; + if (m_publishTimerExpiry <= currentTime - publishingInterval) { - m_publishTimerExpiry += (long)m_publishingInterval; + while (m_publishTimerExpiry <= currentTime) + { + m_publishTimerExpiry += publishingInterval; + } } // OPC 10000-4 §5.14.1.2 row 27: the LifetimeCounter is decremented when the diff --git a/src/Opc.Ua.Server/Subscription/SubscriptionManager.cs b/src/Opc.Ua.Server/Subscription/SubscriptionManager.cs index 8f67cb0b63..9278603152 100644 --- a/src/Opc.Ua.Server/Subscription/SubscriptionManager.cs +++ b/src/Opc.Ua.Server/Subscription/SubscriptionManager.cs @@ -73,6 +73,8 @@ namespace Opc.Ua.Server m_minPublishingInterval = configuration.ServerConfiguration!.MinPublishingInterval; m_maxPublishingInterval = configuration.ServerConfiguration.MaxPublishingInterval; m_publishingResolution = configuration.ServerConfiguration.PublishingResolution; + m_sequentialPublishBudget = Math.Max(1, m_publishingResolution) * m_timeProvider.TimestampFrequency / + (1000L * kSequentialPublishBudgetDivisor); m_maxSubscriptionLifetime = (uint)configuration.ServerConfiguration .MaxSubscriptionLifetime; m_maxDurableSubscriptionLifetimeInHours = (uint) @@ -2997,46 +2999,13 @@ namespace Opc.Ua.Server { m_logger.SubscriptionPublishTaskTaskIdX8Started(Task.CurrentId); - int timeToWait = sleepCycle; + await RunAtFixedPeriodAsync( + m_timeProvider, + sleepCycle, + PublishTimerPass, + cancellationToken).ConfigureAwait(false); - while (true) - { - // ConcurrentDictionary enumeration is thread-safe and provides a stable - // snapshot for the current pass without taking the manager semaphore. - SessionPublishQueue[] queues = [.. m_publishQueues.Values]; - IReadOnlyList abandonedSubscriptions = - CaptureAbandonedPublishTimerSnapshot(); - - // check the publish timer for each subscription. Each queue is - // independent (its own lock and subscription state), so at high - // session counts the O(N) sweep is parallelized across cores to - // keep a single publishing cycle within its resolution budget. - if (queues.Length >= kParallelPublishThreshold) - { - Parallel.For( - 0, - queues.Length, - s_parallelPublishOptions, - ii => queues[ii].PublishTimerExpired()); - } - else - { - for (int ii = 0; ii < queues.Length; ii++) - { - queues[ii].PublishTimerExpired(); - } - } - - ProcessAbandonedPublishTimers(abandonedSubscriptions); - - if (m_shutdownEvent.WaitOne(0)) - { - m_logger.SubscriptionPublishTaskTaskIdX8ExitedNormally(Task.CurrentId); - break; - } - - await m_timeProvider.Delay(TimeSpan.FromMilliseconds(timeToWait), cancellationToken).ConfigureAwait(false); - } + m_logger.SubscriptionPublishTaskTaskIdX8ExitedNormally(Task.CurrentId); } catch (ObjectDisposedException) { @@ -3052,6 +3021,125 @@ namespace Opc.Ua.Server } } + /// + /// Runs a pass every milliseconds until it returns + /// false. + /// + /// + /// The passes start on a fixed cadence: the wait after a pass is what is left of the + /// period, so the time a pass takes does not push the following passes back. A pass + /// that overruns the period is followed by the next pass at once, and the cadence + /// restarts from that pass; the missed periods are not caught up with extra passes. + /// A wait that ends a full period or more after its deadline (a GC pause, a starved + /// thread pool) restarts the cadence from the late pass in the same way. + /// + internal static async ValueTask RunAtFixedPeriodAsync( + TimeProvider timeProvider, + int periodMs, + Func pass, + CancellationToken cancellationToken) + { + long period = Math.Max(1, periodMs); + long nextPass = timeProvider.GetTimestampMilliseconds(); + + while (true) + { + long start = timeProvider.GetTimestampMilliseconds(); + if (start - nextPass >= period) + { + nextPass = start; + } + + if (!pass()) + { + return; + } + + nextPass += period; + long now = timeProvider.GetTimestampMilliseconds(); + if (nextPass > now) + { + await timeProvider.Delay(TimeSpan.FromMilliseconds(nextPass - now), cancellationToken) + .ConfigureAwait(false); + } + else + { + cancellationToken.ThrowIfCancellationRequested(); + nextPass = now; + } + } + } + + /// + /// Checks the publish timer of every subscription once. + /// + /// false when the manager shuts down. + private bool PublishTimerPass() + { + // ConcurrentDictionary enumeration is thread-safe and provides a stable + // snapshot for the current pass without taking the manager semaphore. + SessionPublishQueue[] queues = [.. m_publishQueues.Values]; + IReadOnlyList abandonedSubscriptions = + CaptureAbandonedPublishTimerSnapshot(); + + // check the publish timer for each subscription. + CheckPublishTimers( + queues, + static queue => queue.PublishTimerExpired(), + m_timeProvider, + m_sequentialPublishBudget); + + ProcessAbandonedPublishTimers(abandonedSubscriptions); + + return !m_shutdownEvent.WaitOne(0); + } + + /// + /// Checks the publish timer of each session publish queue, on the calling thread + /// while the pass stays within and in parallel + /// for the queues that are left once it is used up. + /// + /// + /// The cost of a pass follows the monitored items that have data to report, not the + /// number of sessions. A pass with little to do completes on the calling thread + /// without waking pool threads, which costs more CPU than it saves when the pass + /// fits easily into the publishing resolution. A pass that takes a large part of + /// the resolution is completed across cores, so that the publish cycle keeps up. + /// Each queue has its own lock and subscription state, so the queues are + /// independent. + /// + /// The queue type. + /// The queues to check. + /// Checks the publish timer of one queue. + /// The clock that measures the pass. + /// The time, in + /// ticks, after which the remaining queues are checked in parallel. + /// true when part of the pass ran in parallel. + internal static bool CheckPublishTimers( + T[] queues, + Action check, + TimeProvider timeProvider, + long sequentialBudget) + { + long start = timeProvider.GetTimestamp(); + int next = 0; + while (next < queues.Length) + { + check(queues[next++]); + if (next < queues.Length && + timeProvider.GetTimestamp() - start > sequentialBudget) + { + Parallel.For( + next, + queues.Length, + s_parallelPublishOptions, + ii => check(queues[ii])); + return true; + } + } + return false; + } + /// /// Captures the abandoned subscriptions processed by one publish timer pass. /// @@ -3435,11 +3523,10 @@ namespace Opc.Ua.Server } /// - /// Above this many session publish queues the per-cycle sweep is - /// parallelized across cores so one publishing cycle keeps up with - /// thousands of subscriptions instead of serializing on a single thread. + /// The part of the publishing resolution a publish timer pass may take on the + /// sweep thread before the remaining queues are checked in parallel. /// - private const int kParallelPublishThreshold = 256; + private const int kSequentialPublishBudgetDivisor = 4; private static readonly ParallelOptions s_parallelPublishOptions = new() { @@ -3454,6 +3541,7 @@ namespace Opc.Ua.Server private readonly double m_minPublishingInterval; private readonly double m_maxPublishingInterval; private readonly int m_publishingResolution; + private readonly long m_sequentialPublishBudget; private readonly uint m_maxSubscriptionLifetime; private readonly uint m_maxDurableSubscriptionLifetimeInHours; private readonly uint m_minSubscriptionLifetime; diff --git a/tests/Opc.Ua.Server.Tests/Subscription/PublishTimerCadenceTests.cs b/tests/Opc.Ua.Server.Tests/Subscription/PublishTimerCadenceTests.cs new file mode 100644 index 0000000000..4d35a063a0 --- /dev/null +++ b/tests/Opc.Ua.Server.Tests/Subscription/PublishTimerCadenceTests.cs @@ -0,0 +1,272 @@ +/* ======================================================================== + * 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.Concurrent; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Time.Testing; +using NUnit.Framework; + +namespace Opc.Ua.Server.Tests +{ + /// + /// The publish timer loop starts its passes on a fixed cadence: the time a pass takes + /// is not added to the period, and a pass that overruns the period is followed by one + /// pass at once instead of a burst of catch-up passes. A pass runs on the sweep thread + /// until it uses up its budget and checks the remaining queues in parallel. + /// + [TestFixture] + [Category("Subscription")] + [Parallelizable] + public sealed class PublishTimerCadenceTests + { + private const int kPeriod = 100; + private static readonly TimeSpan s_timeout = TimeSpan.FromSeconds(10); + + [Test] + public async Task PassDurationDoesNotDelayTheNextPassAsync() + { + var clock = new DelayRecordingTimeProvider(); + List starts = await RunPassesAsync(clock, [40, 40, 40, 40, 40]).ConfigureAwait(false); + + Assert.That(starts, Is.EqualTo(new long[] { 0, 100, 200, 300, 400 })); + Assert.That(clock.Delays, Is.EqualTo(new long[] { 60, 60, 60, 60 })); + } + + [Test] + public async Task OverrunStartsOneImmediatePassWithoutCatchUpBurstAsync() + { + var clock = new DelayRecordingTimeProvider(); + + // the second pass takes 2.5 periods; the periods at 200 and 300 are missed. + List starts = await RunPassesAsync(clock, [10, 250, 10, 10, 10]).ConfigureAwait(false); + + // no wait follows the overrun, and the cadence restarts at 350. + Assert.That(starts, Is.EqualTo(new long[] { 0, 100, 350, 450, 550 })); + Assert.That(clock.Delays, Is.EqualTo(new long[] { 90, 90, 90 })); + } + + [Test] + public async Task LateWakeRestartsTheCadenceWithoutCatchUpPassAsync() + { + var clock = new DelayRecordingTimeProvider(); + + // the first wait ends 250 ms after its deadline (a GC pause or a starved pool). + List starts = await RunPassesAsync(clock, [10, 10, 10, 10], lateWakes: [250]).ConfigureAwait(false); + + // the late pass at 350 restarts the cadence; no extra pass follows it at once. + Assert.That(starts, Is.EqualTo(new long[] { 0, 350, 450, 550 })); + Assert.That(clock.Delays, Is.EqualTo(new long[] { 90, 90, 90 })); + } + + [Test] + public async Task SmallWakeJitterKeepsTheCadenceAsync() + { + var clock = new DelayRecordingTimeProvider(); + + // a wait that ends 15 ms late does not move the following deadlines. + List starts = await RunPassesAsync(clock, [10, 10, 10], lateWakes: [15]).ConfigureAwait(false); + + Assert.That(starts, Is.EqualTo(new long[] { 0, 115, 200 })); + Assert.That(clock.Delays, Is.EqualTo(new long[] { 90, 75 })); + } + + [Test] + public async Task CancellationStopsTheLoopWhileWaitingAsync() + { + var clock = new DelayRecordingTimeProvider(); + using var cts = new CancellationTokenSource(); + int passes = 0; + + Task loop = SubscriptionManager.RunAtFixedPeriodAsync( + clock, + kPeriod, + () => + { + passes++; + return true; + }, + cts.Token).AsTask(); + + Assert.That(await clock.TimerCreated.WaitAsync(s_timeout).ConfigureAwait(false), Is.True); + cts.Cancel(); + + Assert.That( + async () => await loop.ConfigureAwait(false), + Throws.InstanceOf()); + Assert.That(passes, Is.EqualTo(1)); + } + + [Test] + public void CheapPassChecksEveryQueueOnTheCallingThread() + { + var clock = new FakeTimeProvider(); + int[] queues = CreateQueues(1000); + var checkedOn = new ConcurrentDictionary(); + + bool parallel = SubscriptionManager.CheckPublishTimers( + queues, + queue => checkedOn.TryAdd(queue, Environment.CurrentManagedThreadId), + clock, + BudgetTicks(clock, 25)); + + Assert.That(parallel, Is.False); + Assert.That(checkedOn.Keys, Is.EquivalentTo(queues)); + Assert.That( + checkedOn.Values, + Is.All.EqualTo(Environment.CurrentManagedThreadId)); + } + + [Test] + public void PassOverBudgetChecksTheRemainingQueuesOnceInParallel() + { + var clock = new FakeTimeProvider(); + int[] queues = CreateQueues(200); + var checks = new ConcurrentDictionary(); + + // the first three queues take 10 ms each, so the budget is used up after the third. + bool parallel = SubscriptionManager.CheckPublishTimers( + queues, + queue => + { + checks.AddOrUpdate(queue, 1, (_, count) => count + 1); + if (queue < 3) + { + clock.Advance(TimeSpan.FromMilliseconds(10)); + } + }, + clock, + BudgetTicks(clock, 25)); + + Assert.That(parallel, Is.True); + Assert.That(checks.Keys, Is.EquivalentTo(queues)); + Assert.That(checks.Values, Is.All.EqualTo(1)); + } + + private static int[] CreateQueues(int count) + { + int[] queues = new int[count]; + for (int ii = 0; ii < count; ii++) + { + queues[ii] = ii; + } + return queues; + } + + private static long BudgetTicks(TimeProvider clock, int milliseconds) + { + return milliseconds * clock.TimestampFrequency / 1000; + } + + /// + /// Runs one pass per entry, each advancing the fake clock by its duration, and + /// returns the clock reading at the start of every pass. Entry n of + /// delays the end of the n-th wait beyond its deadline. + /// + private static async Task> RunPassesAsync( + DelayRecordingTimeProvider clock, + long[] durations, + long[]? lateWakes = null) + { + var starts = new List(); + long origin = clock.GetTimestampMilliseconds(); + int pass = 0; + + Task loop = SubscriptionManager.RunAtFixedPeriodAsync( + clock, + kPeriod, + () => + { + starts.Add(clock.GetTimestampMilliseconds() - origin); + clock.Fake.Advance(TimeSpan.FromMilliseconds(durations[pass])); + return ++pass < durations.Length; + }, + CancellationToken.None).AsTask(); + + // fire each wait the loop starts until the last pass returned false. + while (true) + { + Task timer = clock.TimerCreated.WaitAsync(s_timeout); + if (await Task.WhenAny(loop, timer).ConfigureAwait(false) == loop) + { + break; + } + Assert.That(await timer.ConfigureAwait(false), Is.True, "the loop did not wait"); + int wait = clock.Delays.Count - 1; + long late = lateWakes != null && wait < lateWakes.Length ? lateWakes[wait] : 0; + clock.Fake.Advance(TimeSpan.FromMilliseconds(clock.Delays[wait] + late)); + } + + await loop.ConfigureAwait(false); + Assert.That(pass, Is.EqualTo(durations.Length)); + return starts; + } + + /// + /// A fake clock that records the due time of every timer the loop creates. + /// + private sealed class DelayRecordingTimeProvider : TimeProvider + { + public FakeTimeProvider Fake { get; } = new FakeTimeProvider(); + + public List Delays { get; } = []; + + public SemaphoreSlim TimerCreated { get; } = new SemaphoreSlim(0); + + public override long TimestampFrequency => Fake.TimestampFrequency; + + public override long GetTimestamp() + { + return Fake.GetTimestamp(); + } + + public override DateTimeOffset GetUtcNow() + { + return Fake.GetUtcNow(); + } + + public override ITimer CreateTimer( + TimerCallback callback, + object? state, + TimeSpan dueTime, + TimeSpan period) + { + ITimer timer = Fake.CreateTimer(callback, state, dueTime, period); + lock (Delays) + { + Delays.Add((long)dueTime.TotalMilliseconds); + } + TimerCreated.Release(); + return timer; + } + } + } +} diff --git a/tests/Opc.Ua.Server.Tests/SubscriptionStateMachineRegressionTests.cs b/tests/Opc.Ua.Server.Tests/SubscriptionStateMachineRegressionTests.cs index 3910c85b2b..90ff700eb5 100644 --- a/tests/Opc.Ua.Server.Tests/SubscriptionStateMachineRegressionTests.cs +++ b/tests/Opc.Ua.Server.Tests/SubscriptionStateMachineRegressionTests.cs @@ -128,6 +128,91 @@ namespace Opc.Ua.Server.Tests Assert.That(subscription.PublishTimerExpired(), Is.EqualTo(PublishingState.Expired)); } + /// + /// A publish timer pass that runs exactly at the subscription's deadline processes + /// the expiry. The fixed-cadence sweep often lands on the deadline; treating it as not + /// yet due skipped every other cycle when the sweep alternated between 1 ms late and + /// on time. + /// + [Test] + public void SweepExactlyAtTheDeadlineProcessesTheExpiry() + { + var clock = new FakeTimeProvider(); + using Subscription subscription = CreateSubscription( + clock, + maxLifetimeCount: 1000, + maxKeepAliveCount: 300); + + // sweeps at 101, 200, 301, 400, ... ms after the creation; the deadlines are at 100, + // 200, 300, 400, ... + long[] steps = [101, 99, 101, 99, 101, 99, 101, 99]; + foreach (long step in steps) + { + clock.Advance(TimeSpan.FromMilliseconds(step)); + subscription.PublishTimerExpired(); + } + + // no Publish request is queued, so every processed expiry counts the lifetime up. + Assert.That(subscription.Diagnostics.CurrentLifetimeCount, Is.EqualTo((uint)steps.Length)); + } + + /// + /// A sweep that runs at the publishing interval and wakes a little early or late + /// around the deadline does not drop publishing cycles. The deadline moves by one + /// interval per expiry, so a sweep that finds the subscription a whole interval + /// behind because of timer jitter keeps the next expiry due for the next sweep. + /// + [Test] + public void SweepJitterAroundTheDeadlineDoesNotDropCycles() + { + var clock = new FakeTimeProvider(); + using Subscription subscription = CreateSubscription( + clock, + maxLifetimeCount: 1000, + maxKeepAliveCount: 300); + + // deadlines at 100, 200, 300, ...; sweeps at 95, 205, 295, 405, 495, 605, 695, 805. + long[] steps = [95, 110, 90, 110, 90, 110, 90, 110]; + foreach (long step in steps) + { + clock.Advance(TimeSpan.FromMilliseconds(step)); + subscription.PublishTimerExpired(); + } + + // the sweep at 95 is early; every later sweep processes one expiry. + Assert.That(subscription.Diagnostics.CurrentLifetimeCount, Is.EqualTo(7u)); + } + + /// + /// A subscription that fell behind by more than a publishing interval (a stalled + /// sweep) skips the missed expiries instead of catching up with a burst. + /// + [Test] + public void StalledSweepSkipsMissedExpiries() + { + var clock = new FakeTimeProvider(); + using Subscription subscription = CreateSubscription( + clock, + maxLifetimeCount: 1000, + maxKeepAliveCount: 300); + + clock.Advance(TimeSpan.FromMilliseconds(101)); + subscription.PublishTimerExpired(); + + // the sweep stalls for ten intervals; the late sweep processes one expiry. + clock.Advance(TimeSpan.FromMilliseconds(1000)); + subscription.PublishTimerExpired(); + Assert.That(subscription.Diagnostics.CurrentLifetimeCount, Is.EqualTo(2u)); + + // the next expiry is the first deadline after the late sweep (1200 ms). + clock.Advance(TimeSpan.FromMilliseconds(50)); + subscription.PublishTimerExpired(); + Assert.That(subscription.Diagnostics.CurrentLifetimeCount, Is.EqualTo(2u)); + clock.Advance(TimeSpan.FromMilliseconds(50)); + subscription.PublishTimerExpired(); + Assert.That(subscription.Diagnostics.CurrentLifetimeCount, Is.EqualTo(3u)); + } + /// /// M1-1: a client that keeps answering keep-alives never lets the lifetime expire. ///