fix: Optimizes EventScheduler by using PriorityQueue (#2167)

This commit is contained in:
Kamron Batman 2025-04-30 16:33:59 -07:00 • committed by GitHub
parent 21a4092dd8
commit 415c7a6bfd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 30 additions and 61 deletions

View file

@ -86,7 +86,7 @@ public class EventSchedulerTests
Assert.True(called); Assert.True(called);
Assert.Equal(Core._now, evt.NextOccurrence); Assert.Equal(Core._now, evt.NextOccurrence);
EventScheduler.Shared.StopEvent(evt); evt.Cancel();
} }
finally finally
{ {
@ -131,9 +131,9 @@ public class EventSchedulerTests
// Verify they executed in time order, not addition order // Verify they executed in time order, not addition order
Assert.Equal([1, 2, 3], executionOrder); Assert.Equal([1, 2, 3], executionOrder);
EventScheduler.Shared.StopEvent(evt1); evt1.Cancel();
EventScheduler.Shared.StopEvent(evt2); evt2.Cancel();
EventScheduler.Shared.StopEvent(evt3); evt3.Cancel();
} }
finally finally
{ {
@ -169,7 +169,7 @@ public class EventSchedulerTests
Assert.Equal(3, callCount); Assert.Equal(3, callCount);
Assert.Equal(3, evt.CallCount); Assert.Equal(3, evt.CallCount);
EventScheduler.Shared.StopEvent(evt); evt.Cancel();
} }
finally finally
{ {
@ -211,8 +211,8 @@ public class EventSchedulerTests
Assert.True(failedEventCalled); Assert.True(failedEventCalled);
Assert.True(laterEventCalled); Assert.True(laterEventCalled);
EventScheduler.Shared.StopEvent(failedEvt); failedEvt.Cancel();
EventScheduler.Shared.StopEvent(laterEvt); laterEvt.Cancel();
} }
finally finally
{ {
@ -235,7 +235,7 @@ public class EventSchedulerTests
); );
// Remove before execution // Remove before execution
EventScheduler.Shared.StopEvent(evt); evt.Cancel();
// Advance time // Advance time
Core._now = Core._now.AddSeconds(20); Core._now = Core._now.AddSeconds(20);

View file

@ -33,15 +33,12 @@ public enum DaysOfWeek : byte
public abstract class ScheduledEvent public abstract class ScheduledEvent
{ {
private static Serial _nextSerial = (Serial)1;
// Tie-breaker for sorted set
public Serial Serial { get; }
public IRecurrencePattern Recurrence { get; } public IRecurrencePattern Recurrence { get; }
public TimeZoneInfo TimeZone { get; } public TimeZoneInfo TimeZone { get; }
public TimeOnly Time { get; } public TimeOnly Time { get; }
public DateTime EndDate { get; } public DateTime EndDate { get; }
public DateTime NextOccurrence { get; private set; } public DateTime NextOccurrence { get; private set; }
public bool Cancelled { get; private set; }
public ScheduledEvent(DateTime startOn, TimeZoneInfo timeZone = null) public ScheduledEvent(DateTime startOn, TimeZoneInfo timeZone = null)
: this(startOn, startOn, TimeOnly.FromDateTime(startOn), null, timeZone) : this(startOn, startOn, TimeOnly.FromDateTime(startOn), null, timeZone)
@ -61,7 +58,6 @@ public abstract class ScheduledEvent
TimeZoneInfo timeZone = null TimeZoneInfo timeZone = null
) )
{ {
Serial = _nextSerial++;
Time = time; Time = time;
Recurrence = recurrence; Recurrence = recurrence;
TimeZone = timeZone ?? TimeZoneInfo.Utc; TimeZone = timeZone ?? TimeZoneInfo.Utc;
@ -71,18 +67,19 @@ public abstract class ScheduledEvent
EndDate = endOn == DateTime.MaxValue || endOn.Kind == DateTimeKind.Utc ? endOn : endOn.LocalToUtc(TimeZone); EndDate = endOn == DateTime.MaxValue || endOn.Kind == DateTimeKind.Utc ? endOn : endOn.LocalToUtc(TimeZone);
} }
public bool Advance() public void Cancel() => Cancelled = true;
public DateTime Advance()
{ {
OnEvent(); OnEvent();
var next = Recurrence?.GetNextOccurrence(NextOccurrence, Time, TimeZone) ?? DateTime.MaxValue; var next = Recurrence?.GetNextOccurrence(NextOccurrence, Time, TimeZone) ?? DateTime.MaxValue;
if (next == DateTime.MaxValue || next > EndDate) if (next == DateTime.MaxValue || next > EndDate)
{ {
return false; return DateTime.MaxValue;
} }
NextOccurrence = next; return NextOccurrence = next;
return true;
} }
public abstract void OnEvent(); public abstract void OnEvent();
@ -92,7 +89,7 @@ public class EventScheduler : Timer
{ {
private static readonly ILogger logger = LogFactory.GetLogger(typeof(EventScheduler)); private static readonly ILogger logger = LogFactory.GetLogger(typeof(EventScheduler));
private readonly SortedSet<ScheduledEvent> _schedule = new(ScheduledEventComparer.Default); private readonly PriorityQueue<ScheduledEvent, DateTime> _schedule = new();
public static EventScheduler Shared { get; private set; } public static EventScheduler Shared { get; private set; }
@ -182,15 +179,7 @@ public class EventScheduler : Timer
{ {
if (entry != null) if (entry != null)
{ {
_schedule.Add(entry); _schedule.Enqueue(entry, entry.NextOccurrence);
}
}
public void StopEvent(ScheduledEvent entry)
{
if (entry != null)
{
_schedule.Remove(entry);
} }
} }
@ -200,55 +189,35 @@ public class EventScheduler : Timer
while (_schedule.Count > 0) while (_schedule.Count > 0)
{ {
var entry = _schedule.Min!; var entry = _schedule.Peek();
if (entry.NextOccurrence > now) var cancelled = entry.Cancelled;
if (!cancelled && entry.NextOccurrence > now)
{ {
break; break;
} }
_schedule.Remove(entry); _schedule.Dequeue();
bool advance; if (cancelled)
{
continue;
}
DateTime nextOccurrence;
try try
{ {
advance = entry.Advance(); nextOccurrence = entry.Advance();
} }
catch (Exception e) catch (Exception e)
{ {
logger.Error(e, "Error while executing scheduled event."); logger.Error(e, "Error while executing scheduled event.");
advance = false; nextOccurrence = DateTime.MaxValue;
} }
if (advance && entry.NextOccurrence < DateTime.MaxValue) if (nextOccurrence < DateTime.MaxValue)
{ {
_schedule.Add(entry); _schedule.Enqueue(entry, nextOccurrence);
} }
} }
} }
private sealed class ScheduledEventComparer : IComparer<ScheduledEvent>
{
public static readonly ScheduledEventComparer Default = new();
public int Compare(ScheduledEvent x, ScheduledEvent y)
{
if (x == null && y == null)
{
return 0;
}
if (x == null)
{
return 1;
}
if (y == null)
{
return -1;
}
var next = x.NextOccurrence.CompareTo(y.NextOccurrence);
return next != 0 ? next : x.Serial.CompareTo(y.Serial);
}
}
} }