GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-28 22:55:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

Server: keep the publish timer on a fixed cadence and parallelize the sweep by work (#4829)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
romanett
2026-10-19 06:48:41 +02:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 014f99a53a
commit 37a2a831f3
4 changed files with 504 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Opc.Ua.Server/Subscription/Subscription.cs
+16 -5
View File
@@ -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
src/Opc.Ua.Server/Subscription/SubscriptionManager.cs
+131 -43
View File
@@ -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<ISubscriptionPublishPipeline> 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
}
}
/// <summary>
/// Runs a pass every <paramref name="periodMs"/> milliseconds until it returns
/// <c>false</c>.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
internal static async ValueTask RunAtFixedPeriodAsync(
TimeProvider timeProvider,
int periodMs,
Func<bool> 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;
}
}
}
/// <summary>
/// Checks the publish timer of every subscription once.
/// </summary>
/// <returns><c>false</c> when the manager shuts down.</returns>
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<ISubscriptionPublishPipeline> 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);
}
/// <summary>
/// Checks the publish timer of each session publish queue, on the calling thread
/// while the pass stays within <paramref name="sequentialBudget"/> and in parallel
/// for the queues that are left once it is used up.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
/// <typeparam name="T">The queue type.</typeparam>
/// <param name="queues">The queues to check.</param>
/// <param name="check">Checks the publish timer of one queue.</param>
/// <param name="timeProvider">The clock that measures the pass.</param>
/// <param name="sequentialBudget">The time, in <see cref="TimeProvider.GetTimestamp"/>
/// ticks, after which the remaining queues are checked in parallel.</param>
/// <returns><c>true</c> when part of the pass ran in parallel.</returns>
internal static bool CheckPublishTimers<T>(
T[] queues,
Action<T> 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;
}
/// <summary>
/// Captures the abandoned subscriptions processed by one publish timer pass.
/// </summary>
@@ -3435,11 +3523,10 @@ namespace Opc.Ua.Server
}
/// <summary>
/// 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.
/// </summary>
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;
tests/Opc.Ua.Server.Tests/Subscription/PublishTimerCadenceTests.cs
+272
View File
@@ -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
{
/// <summary>
/// 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.
/// </summary>
[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<long> 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<long> 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<long> 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<long> 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<OperationCanceledException>());
Assert.That(passes, Is.EqualTo(1));
}
[Test]
public void CheapPassChecksEveryQueueOnTheCallingThread()
{
var clock = new FakeTimeProvider();
int[] queues = CreateQueues(1000);
var checkedOn = new ConcurrentDictionary<int, int>();
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<int, int>();
// 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;
}
/// <summary>
/// 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
/// <paramref name="lateWakes"/> delays the end of the n-th wait beyond its deadline.
/// </summary>
private static async Task<List<long>> RunPassesAsync(
DelayRecordingTimeProvider clock,
long[] durations,
long[]? lateWakes = null)
{
var starts = new List<long>();
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<bool> 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;
}
/// <summary>
/// A fake clock that records the due time of every timer the loop creates.
/// </summary>
private sealed class DelayRecordingTimeProvider : TimeProvider
{
public FakeTimeProvider Fake { get; } = new FakeTimeProvider();
public List<long> 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;
}
}
}
}
tests/Opc.Ua.Server.Tests/SubscriptionStateMachineRegressionTests.cs
+85
View File
@@ -128,6 +128,91 @@ namespace Opc.Ua.Server.Tests
Assert.That(subscription.PublishTimerExpired(), Is.EqualTo(PublishingState.Expired));
}
/// <summary>
/// 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.
/// </summary>
[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));
}
/// <summary>
/// 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.
/// </summary>
[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));
}
/// <summary>
/// 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.
/// </summary>
[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));
}
/// <summary>
/// M1-1: a client that keeps answering keep-alives never lets the lifetime expire.
/// </summary>