Skip to content

Commit f37b607

Browse files
tgiachitgiachi
authored
Merge pull request #30 from tgiachi/feature/eventloop
feat(loop): add EventLoopService (dispatcher drain + timer wheel)
2 parents 1091e95 + 3d3a9db commit f37b607

10 files changed

Lines changed: 385 additions & 3 deletions

File tree

‎src/SquidStd.Core/Data/EventLoop/EventLoopConfig.cs‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
namespace SquidStd.Core.Data.EventLoop;
2+
3+
/// <summary>
4+
/// Configuration for the event loop thread.
5+
/// </summary>
6+
public sealed class EventLoopConfig
7+
{
8+
/// <summary>
9+
/// When <c>true</c>, the loop sleeps <see cref="IdleSleepMs" /> milliseconds whenever a tick
10+
/// processes zero work units. When <c>false</c>, the loop spins continuously.
11+
/// </summary>
12+
public bool IdleCpuEnabled { get; set; } = true;
13+
14+
/// <summary>Milliseconds to sleep when a tick finds no work. Default 1 ms.</summary>
15+
public int IdleSleepMs { get; set; } = 1;
16+
17+
/// <summary>Ticks slower than this threshold (ms) emit a warning. Default 250 ms.</summary>
18+
public double SlowTickThresholdMs { get; set; } = 250;
19+
}

‎src/SquidStd.Core/Interfaces/Threading/IEventLoopService.cs‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
namespace SquidStd.Core.Interfaces.Threading;
2+
3+
/// <summary>
4+
/// Exposes runtime metrics of the event loop. The loop lifecycle is provided by the implementation
5+
/// (which also implements the service lifecycle interface); this interface stays free of an
6+
/// Abstractions dependency because <c>SquidStd.Core</c> does not reference it.
7+
/// </summary>
8+
public interface IEventLoopService
9+
{
10+
/// <summary>Number of loop iterations performed since start.</summary>
11+
long TickCount { get; }
12+
13+
/// <summary>Exponential moving average of per-tick elapsed time in milliseconds.</summary>
14+
double AverageTickMs { get; }
15+
16+
/// <summary>Worst observed tick elapsed time in milliseconds.</summary>
17+
double MaxTickMs { get; }
18+
}

‎src/SquidStd.Core/Interfaces/Timing/ITimerWheelDriver.cs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
namespace SquidStd.Core.Interfaces.Timing;
2+
3+
/// <summary>
4+
/// Marker for a service that advances the timer wheel (<see cref="ITimerService.UpdateTicksDelta" />).
5+
/// At most one driver must be registered at a time, so that the wheel is not advanced twice per elapsed
6+
/// time. Implemented by the timer-wheel pump and by the event loop, which are mutually exclusive.
7+
/// </summary>
8+
public interface ITimerWheelDriver
9+
{
10+
}

‎src/SquidStd.Mail.MailKit/Extensions/MailRegistrationExtensions.cs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using SquidStd.Abstractions.Extensions.Config;
33
using SquidStd.Abstractions.Extensions.Services;
44
using SquidStd.Core.Data.Timing;
5+
using SquidStd.Core.Interfaces.Timing;
56
using SquidStd.Mail.Abstractions.Data.Config;
67
using SquidStd.Mail.Abstractions.Interfaces;
78
using SquidStd.Mail.Abstractions.Types.Mail;
@@ -43,10 +44,11 @@ public IContainer AddMail(MailOptions options)
4344

4445
container.RegisterStdService<MailPollingService, MailPollingService>(100);
4546

46-
if (!container.IsRegistered<TimerWheelPumpService>())
47+
if (!container.IsRegistered<ITimerWheelDriver>())
4748
{
4849
container.RegisterConfigSection("timerWheelPump", static () => new TimerWheelPumpConfig(), -90);
4950
container.RegisterStdService<TimerWheelPumpService, TimerWheelPumpService>(-1);
51+
container.RegisterMapping<ITimerWheelDriver, TimerWheelPumpService>();
5052
}
5153

5254
return container;

‎src/SquidStd.Services.Core/Extensions/RegisterSchedulerServicesExtension.cs‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,13 @@
11
using DryIoc;
22
using SquidStd.Abstractions.Extensions.Config;
33
using SquidStd.Abstractions.Extensions.Services;
4+
using SquidStd.Core.Data.EventLoop;
45
using SquidStd.Core.Data.Timing;
6+
using SquidStd.Core.Interfaces.Metrics;
57
using SquidStd.Core.Interfaces.Scheduling;
8+
using SquidStd.Core.Interfaces.Threading;
9+
using SquidStd.Core.Interfaces.Timing;
10+
using SquidStd.Services.Core.Services.EventLoop;
611
using SquidStd.Services.Core.Services.Scheduling;
712

813
namespace SquidStd.Services.Core.Extensions;
@@ -24,9 +29,34 @@ public IContainer RegisterSchedulerServices()
2429
{
2530
container.RegisterConfigSection("timerWheelPump", static () => new TimerWheelPumpConfig(), -90);
2631
container.RegisterStdService<TimerWheelPumpService, TimerWheelPumpService>(-1);
32+
container.RegisterMapping<ITimerWheelDriver, TimerWheelPumpService>();
2733
container.RegisterStdService<ICronScheduler, CronSchedulerService>(-1);
2834

2935
return container;
3036
}
37+
38+
/// <summary>
39+
/// Registers the event loop (dedicated thread draining the dispatcher and advancing the timer
40+
/// wheel). Mutually exclusive with the timer-wheel pump: throws if a timer-wheel driver is
41+
/// already registered. Must be called after <c>RegisterCoreServices</c> so that
42+
/// <c>IMainThreadDispatcher</c> and <c>ITimerService</c> exist.
43+
/// </summary>
44+
/// <returns>The same container for chaining.</returns>
45+
public IContainer RegisterEventLoop()
46+
{
47+
if (container.IsRegistered<ITimerWheelDriver>())
48+
{
49+
throw new InvalidOperationException(
50+
"A timer wheel driver is already registered (EventLoop or TimerWheelPump). They are mutually exclusive."
51+
);
52+
}
53+
54+
container.RegisterConfigSection("eventLoop", static () => new EventLoopConfig(), -90);
55+
container.RegisterStdService<IEventLoopService, EventLoopService>(-1);
56+
container.RegisterMapping<ITimerWheelDriver, IEventLoopService>();
57+
container.RegisterMapping<IMetricProvider, IEventLoopService>();
58+
59+
return container;
60+
}
3161
}
3262
}

‎src/SquidStd.Services.Core/Services/EventLoop/EventLoopService.cs‎

Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
using System.Diagnostics;
2+
using Serilog;
3+
using SquidStd.Abstractions.Interfaces.Services;
4+
using SquidStd.Core.Data.EventLoop;
5+
using SquidStd.Core.Data.Metrics;
6+
using SquidStd.Core.Interfaces.Metrics;
7+
using SquidStd.Core.Interfaces.Threading;
8+
using SquidStd.Core.Interfaces.Timing;
9+
using SquidStd.Core.Types.Metrics;
10+
11+
namespace SquidStd.Services.Core.Services.EventLoop;
12+
13+
/// <summary>
14+
/// General-purpose event loop. Owns a dedicated background thread that each frame drains the
15+
/// <see cref="IMainThreadDispatcher" /> and advances the timer wheel (<see cref="ITimerService" />),
16+
/// sleeps when idle, and exposes tick metrics. Mutually exclusive with the timer-wheel pump.
17+
/// </summary>
18+
public sealed class EventLoopService : IEventLoopService, ISquidStdService, IMetricProvider, ITimerWheelDriver, IDisposable
19+
{
20+
private readonly IMainThreadDispatcher _dispatcher;
21+
private readonly ITimerService _timer;
22+
private readonly EventLoopConfig _config;
23+
private readonly CancellationTokenSource _cts = new();
24+
private readonly ILogger _logger = Log.ForContext<EventLoopService>();
25+
private readonly Lock _metricsSync = new();
26+
private double _averageTickMs;
27+
private long _idleSleepCount;
28+
private double _maxTickMs;
29+
private Thread? _thread;
30+
private long _tickCount;
31+
32+
/// <inheritdoc />
33+
public long TickCount => Interlocked.Read(ref _tickCount);
34+
35+
/// <inheritdoc />
36+
public double AverageTickMs
37+
{
38+
get
39+
{
40+
lock (_metricsSync)
41+
{
42+
return _averageTickMs;
43+
}
44+
}
45+
}
46+
47+
/// <inheritdoc />
48+
public double MaxTickMs
49+
{
50+
get
51+
{
52+
lock (_metricsSync)
53+
{
54+
return _maxTickMs;
55+
}
56+
}
57+
}
58+
59+
/// <inheritdoc />
60+
public string ProviderName => "eventloop";
61+
62+
public EventLoopService(IMainThreadDispatcher dispatcher, ITimerService timer, EventLoopConfig config)
63+
{
64+
_dispatcher = dispatcher;
65+
_timer = timer;
66+
_config = config;
67+
}
68+
69+
/// <inheritdoc />
70+
public ValueTask StartAsync(CancellationToken cancellationToken = default)
71+
{
72+
_thread = new Thread(RunLoop)
73+
{
74+
IsBackground = true,
75+
Name = "SquidStd-EventLoop"
76+
};
77+
_thread.Start();
78+
79+
return ValueTask.CompletedTask;
80+
}
81+
82+
/// <inheritdoc />
83+
public ValueTask StopAsync(CancellationToken cancellationToken = default)
84+
{
85+
_cts.Cancel();
86+
_thread?.Join(TimeSpan.FromSeconds(5));
87+
88+
return ValueTask.CompletedTask;
89+
}
90+
91+
/// <inheritdoc />
92+
public ValueTask<IReadOnlyList<MetricSample>> CollectAsync(CancellationToken cancellationToken = default)
93+
{
94+
double avg;
95+
double max;
96+
97+
lock (_metricsSync)
98+
{
99+
avg = _averageTickMs;
100+
max = _maxTickMs;
101+
}
102+
103+
IReadOnlyList<MetricSample> samples =
104+
[
105+
new("tick_count", Interlocked.Read(ref _tickCount), Type: MetricType.Counter, Help: "Total event loop iterations"),
106+
new("tick_avg_ms", avg, Help: "EMA tick elapsed in ms"),
107+
new("tick_max_ms", max, Help: "Worst tick elapsed in ms"),
108+
new("idle_sleeps_total", Interlocked.Read(ref _idleSleepCount), Type: MetricType.Counter, Help: "Total idle sleeps")
109+
];
110+
111+
return new ValueTask<IReadOnlyList<MetricSample>>(samples);
112+
}
113+
114+
internal int Tick()
115+
{
116+
var start = Stopwatch.GetTimestamp();
117+
var work = _dispatcher.DrainPending();
118+
var nowMs = (long)Math.Floor(Stopwatch.GetTimestamp() * 1000.0 / Stopwatch.Frequency);
119+
work += _timer.UpdateTicksDelta(nowMs);
120+
var elapsed = Stopwatch.GetElapsedTime(start);
121+
122+
UpdateMetrics(elapsed);
123+
124+
if (elapsed.TotalMilliseconds >= _config.SlowTickThresholdMs)
125+
{
126+
_logger.Warning(
127+
"Slow tick: {Elapsed:0.###}ms work={Work} pending={Pending}",
128+
elapsed.TotalMilliseconds,
129+
work,
130+
_dispatcher.PendingCount
131+
);
132+
}
133+
134+
return work;
135+
}
136+
137+
private void RunLoop()
138+
{
139+
while (!_cts.IsCancellationRequested)
140+
{
141+
var work = Tick();
142+
143+
if (_config.IdleCpuEnabled && work == 0)
144+
{
145+
Thread.Sleep(_config.IdleSleepMs);
146+
Interlocked.Increment(ref _idleSleepCount);
147+
}
148+
}
149+
}
150+
151+
private void UpdateMetrics(TimeSpan elapsed)
152+
{
153+
Interlocked.Increment(ref _tickCount);
154+
155+
lock (_metricsSync)
156+
{
157+
// Exponential moving average: 0.95 weight to history, 0.05 to current sample.
158+
_averageTickMs = _averageTickMs * 0.95 + elapsed.TotalMilliseconds * 0.05;
159+
_maxTickMs = Math.Max(_maxTickMs, elapsed.TotalMilliseconds);
160+
}
161+
}
162+
163+
/// <inheritdoc />
164+
public void Dispose()
165+
{
166+
_cts.Dispose();
167+
GC.SuppressFinalize(this);
168+
}
169+
}

‎src/SquidStd.Services.Core/Services/Scheduling/TimerWheelPumpService.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ namespace SquidStd.Services.Core.Services.Scheduling;
1111
/// (non-game-loop) application. Drives <see cref="ITimerService.UpdateTicksDelta" /> on a
1212
/// background loop.
1313
/// </summary>
14-
public sealed class TimerWheelPumpService : ISquidStdService, IDisposable
14+
public sealed class TimerWheelPumpService : ISquidStdService, ITimerWheelDriver, IDisposable
1515
{
1616
private readonly CancellationTokenSource _cts = new();
1717
private readonly ILogger _logger = Log.ForContext<TimerWheelPumpService>();

‎src/SquidStd.Workers.Manager/Extensions/WorkerManagerRegistrationExtensions.cs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using SquidStd.Abstractions.Extensions.Config;
33
using SquidStd.Abstractions.Extensions.Services;
44
using SquidStd.Core.Data.Timing;
5+
using SquidStd.Core.Interfaces.Timing;
56
using SquidStd.Services.Core.Services.Scheduling;
67
using SquidStd.Workers.Manager.Data.Config;
78
using SquidStd.Workers.Manager.Interfaces;
@@ -33,10 +34,11 @@ public IContainer AddWorkerManager()
3334
container.RegisterStdService<HeartbeatCollectorService, HeartbeatCollectorService>(100);
3435
container.RegisterStdService<WorkerOfflineSweepService, WorkerOfflineSweepService>(110);
3536

36-
if (!container.IsRegistered<TimerWheelPumpService>())
37+
if (!container.IsRegistered<ITimerWheelDriver>())
3738
{
3839
container.RegisterConfigSection("timerWheelPump", static () => new TimerWheelPumpConfig(), -90);
3940
container.RegisterStdService<TimerWheelPumpService, TimerWheelPumpService>(-1);
41+
container.RegisterMapping<ITimerWheelDriver, TimerWheelPumpService>();
4042
}
4143

4244
return container;

‎tests/SquidStd.Tests/EventLoop/EventLoopRegistrationTests.cs‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
using DryIoc;
2+
using SquidStd.Abstractions.Extensions.Config;
3+
using SquidStd.Abstractions.Extensions.Services;
4+
using SquidStd.Core.Interfaces.Threading;
5+
using SquidStd.Core.Interfaces.Timing;
6+
using SquidStd.Services.Core.Extensions;
7+
using SquidStd.Services.Core.Services;
8+
using SquidStd.Services.Core.Services.Scheduling;
9+
10+
namespace SquidStd.Tests.EventLoop;
11+
12+
public class EventLoopRegistrationTests
13+
{
14+
[Fact]
15+
public void RegisterEventLoop_RegistersDriverAndMetricProvider()
16+
{
17+
using var container = new Container();
18+
container.RegisterCoreServices("squidstd", Path.GetTempPath());
19+
20+
container.RegisterEventLoop();
21+
22+
Assert.True(container.IsRegistered<IEventLoopService>());
23+
Assert.True(container.IsRegistered<ITimerWheelDriver>());
24+
}
25+
26+
[Fact]
27+
public void RegisterEventLoop_AfterPump_Throws()
28+
{
29+
using var container = new Container();
30+
container.RegisterCoreServices("squidstd", Path.GetTempPath());
31+
container.RegisterConfigSection("timerWheelPump", static () => new SquidStd.Core.Data.Timing.TimerWheelPumpConfig(), -90);
32+
container.RegisterStdService<TimerWheelPumpService, TimerWheelPumpService>(-1);
33+
container.RegisterMapping<ITimerWheelDriver, TimerWheelPumpService>();
34+
35+
Assert.Throws<InvalidOperationException>(() => container.RegisterEventLoop());
36+
}
37+
38+
[Fact]
39+
public void EventLoopService_IsATimerWheelDriver()
40+
{
41+
Assert.True(typeof(ITimerWheelDriver).IsAssignableFrom(typeof(SquidStd.Services.Core.Services.EventLoop.EventLoopService)));
42+
}
43+
44+
[Fact]
45+
public void TimerWheelPumpService_IsATimerWheelDriver()
46+
{
47+
Assert.True(typeof(ITimerWheelDriver).IsAssignableFrom(typeof(TimerWheelPumpService)));
48+
}
49+
}

0 commit comments

Comments
 (0)