Repository navigation
Keep the timer queue ordered when a timer is changed or disposed #85
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from 3 commits
7bdd184
e55e44a
922ac86
a73fbd8
9435f66
051b2a0
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,7 +20,11 @@ public sealed class SimulatedTimeProvider : TimeProvider | |
| private long _schedulerTicks; | ||
|
|
||
| private long _nextId; | ||
| private readonly PriorityQueue<ScheduledTimer, ScheduledTimer> _queue; | ||
|
|
||
| // Each entry carries the timer's schedule version and an immutable (due, id) priority. Change and Dispose | ||
| // bump the version and leave the old entry in place; Advance discards entries whose version is stale. An | ||
| // entry's priority never changes while it is queued, so the heap order always holds. | ||
| private readonly PriorityQueue<(ScheduledTimer Timer, long Version), (long DueAtTicks, long Id)> _queue; | ||
|
|
||
| /// <summary> | ||
| /// Gets lightweight counters that can be used to observe scheduling behavior during simulation. | ||
|
|
@@ -38,7 +42,7 @@ public SimulatedTimeProvider(DateTimeOffset? startTime = null, TimeZoneInfo? loc | |
| _localTimeZone = localTimeZone ?? TimeZoneInfo.Utc; | ||
|
|
||
| _schedulerTicks = 0; | ||
| _queue = new PriorityQueue<ScheduledTimer, ScheduledTimer>(new ScheduledTimerComparer()); | ||
| _queue = new PriorityQueue<(ScheduledTimer Timer, long Version), (long DueAtTicks, long Id)>(); | ||
| } | ||
|
|
||
| /// <summary> | ||
|
|
@@ -135,47 +139,38 @@ public void Advance(TimeSpan by) | |
|
|
||
| Statistics.RecordAdvance(by); | ||
|
|
||
| List<(TimerCallback Callback, object? State)>? due = null; | ||
| List<(ScheduledTimer Timer, long Version)>? due = null; | ||
|
|
||
| lock (_gate) | ||
| { | ||
| _utcNow = _utcNow.Add(by); | ||
| _schedulerTicks += by.Ticks; | ||
|
|
||
| while (_queue.TryPeek(out var timer, out _)) | ||
| while (_queue.TryPeek(out var entry, out var priority)) | ||
| { | ||
| if (timer.IsDisposed) | ||
| if (entry.Timer.IsDisposed || entry.Version != entry.Timer.Version) | ||
| { | ||
| // Disposed, or rescheduled since this entry was queued: a newer entry stands for it. | ||
| _queue.Dequeue(); | ||
| continue; | ||
| } | ||
|
|
||
| if (timer.DueAtTicks > _schedulerTicks) | ||
| if (priority.DueAtTicks > _schedulerTicks) | ||
| { | ||
| break; | ||
| } | ||
|
|
||
| _queue.Dequeue(); | ||
|
|
||
| if (timer.IsDisposed) | ||
| { | ||
| continue; | ||
| } | ||
|
|
||
| due ??= []; | ||
| due.Add((timer.Callback, timer.State)); | ||
| due.Add(entry); | ||
|
|
||
| // Periodic timers: coalesce on jump; schedule next occurrence from "now". | ||
| if (timer.PeriodTicks > 0) | ||
| if (entry.Timer.PeriodTicks > 0) | ||
| { | ||
| timer.DueAtTicks = _schedulerTicks + timer.PeriodTicks; | ||
| _queue.Enqueue(timer, timer); | ||
| entry.Timer.DueAtTicks = _schedulerTicks + entry.Timer.PeriodTicks; | ||
| Enqueue(entry.Timer); | ||
| Statistics.RecordPeriodicReschedule(_queue.Count); | ||
| } | ||
| else | ||
| { | ||
| timer.MarkDisposed(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -184,11 +179,25 @@ public void Advance(TimeSpan by) | |
| return; | ||
| } | ||
|
|
||
| // Fire callbacks outside lock. | ||
| foreach (var (callback, state) in due) | ||
| // Fire callbacks outside lock. A callback may dispose or reschedule a timer that this advance has already | ||
| // found due; such a timer does not fire for this occurrence. | ||
| foreach (var (timer, version) in due) | ||
| { | ||
| lock (_gate) | ||
| { | ||
| if (timer.IsDisposed || timer.Version != version) | ||
| { | ||
| continue; | ||
| } | ||
|
|
||
| if (timer.PeriodTicks == 0) | ||
| { | ||
| timer.MarkDisposed(); | ||
| } | ||
| } | ||
|
|
||
| Statistics.RecordCallbackFired(); | ||
| callback(state); | ||
| timer.Callback(timer.State); | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
When a callback throws, this Useful? React with 👍 / 👎. |
||
| } | ||
| } | ||
|
|
||
|
|
@@ -221,36 +230,15 @@ public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSp | |
| var periodTicks = period == Timeout.InfiniteTimeSpan ? 0 : period.Ticks; | ||
|
|
||
| var timer = new ScheduledTimer(this, id, callback, state, dueTicks, periodTicks); | ||
| _queue.Enqueue(timer, timer); | ||
| Enqueue(timer); | ||
| Statistics.RecordTimerCreated(_queue.Count); | ||
| return timer; | ||
| } | ||
| } | ||
|
|
||
| private void Reschedule(ScheduledTimer timer) | ||
| { | ||
| if (timer.IsDisposed) | ||
| { | ||
| return; | ||
| } | ||
|
|
||
| _queue.Enqueue(timer, timer); | ||
| Statistics.RecordQueueEnqueue(_queue.Count); | ||
| } | ||
|
|
||
| private sealed class ScheduledTimerComparer : IComparer<ScheduledTimer> | ||
| { | ||
| public int Compare(ScheduledTimer? x, ScheduledTimer? y) | ||
| { | ||
| if (ReferenceEquals(x, y)) return 0; | ||
| if (x is null) return -1; | ||
| if (y is null) return 1; | ||
|
|
||
| var cmp = x.DueAtTicks.CompareTo(y.DueAtTicks); | ||
| if (cmp != 0) return cmp; | ||
| return x.Id.CompareTo(y.Id); | ||
| } | ||
| } | ||
| // Queues the timer's current schedule. Callers hold _gate. | ||
| private void Enqueue(ScheduledTimer timer) => | ||
| _queue.Enqueue((timer, timer.Version), (timer.DueAtTicks, timer.Id)); | ||
|
|
||
| private sealed class ScheduledTimer : ITimer | ||
| { | ||
|
|
@@ -295,6 +283,12 @@ public ScheduledTimer( | |
| /// </summary> | ||
| public long PeriodTicks { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// Gets the version of the timer's schedule, which <see cref="Change"/> increments. A queued entry with an | ||
| /// older version is stale. Read and written under the owner's lock. | ||
| /// </summary> | ||
| public long Version { get; private set; } | ||
|
|
||
| /// <summary> | ||
| /// Gets a value indicating whether this timer has been disposed. | ||
| /// </summary> | ||
|
|
@@ -332,8 +326,11 @@ public bool Change(TimeSpan dueTime, TimeSpan period) | |
| DueAtTicks = dueTicks; | ||
| PeriodTicks = period == Timeout.InfiniteTimeSpan ? 0 : period.Ticks; | ||
|
|
||
| // Never reorder a queued entry: supersede it, and queue the new schedule. | ||
| Version++; | ||
| _owner.Statistics.RecordTimerChange(); | ||
| _owner.Reschedule(this); | ||
| _owner.Enqueue(this); | ||
| _owner.Statistics.RecordQueueEnqueue(_owner._queue.Count); | ||
| return true; | ||
| } | ||
| } | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[P2] Bound stale-entry retention when timers move earlier
Root-only cleanup leaves stale entries at their original immutable priority. Keep one 1-second periodic timer alive, then repeatedly create a disabled one-shot, Change it to 1 second, and Advance by 1 second. Every one-shot fires and is disposed, but its original
long.MaxValueentry remains behind the heartbeat's next occurrence forever, strongly retaining its callback/state. After 1,000 iterations and full GC, this head retained all 1,000 payloads and had 1,001 queue entries; the base retained zero payloads and only the heartbeat entry. Long-lived simulations therefore grow memory with total completed timers rather than live timers, and heap operations grow with that accumulated history. Use bounded stale-entry compaction or removal/replacement of superseded entries; also avoid enqueuing disabled schedules where possible. Add a regression that keeps a live periodic timer throughout the workload.Minimal workload (supply a fresh state payload per iteration to observe retention):
The same retention mechanism applies to far-future finite schedules moved earlier; merely omitting Infinite entries does not bound that case.