Skip to content

Commit 9a01f26

Browse files
tgiachitgiachi
committed
Merge branch 'feature/actors-mailbox' into develop
Add SquidStd.Actors: actor/mailbox primitive over TPL Dataflow ActionBlock with FIFO single-consumer processing, TellAsync/AskAsync, configurable overflow and error policies, drain-on-dispose, and an EventBus adapter.
2 parents 303891c + 2aa2f01 commit 9a01f26

21 files changed

Lines changed: 886 additions & 101 deletions

‎SquidStd.slnx‎

Lines changed: 54 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -1,61 +1,62 @@
11
<Solution>
22
<Folder Name="/src/">
3-
<Project Path="src/SquidStd.Abstractions/SquidStd.Abstractions.csproj"/>
4-
<Project Path="src/SquidStd.Generators/SquidStd.Generators.csproj"/>
5-
<Project Path="src/SquidStd.Telemetry.Abstractions/SquidStd.Telemetry.Abstractions.csproj"/>
6-
<Project Path="src/SquidStd.Telemetry.OpenTelemetry/SquidStd.Telemetry.OpenTelemetry.csproj"/>
7-
<Project Path="src/SquidStd.Aws.Abstractions/SquidStd.Aws.Abstractions.csproj"/>
8-
<Project Path="src/SquidStd.AspNetCore/SquidStd.AspNetCore.csproj"/>
9-
<Project Path="src/SquidStd.Caching.Abstractions/SquidStd.Caching.Abstractions.csproj"/>
10-
<Project Path="src/SquidStd.Caching.Redis/SquidStd.Caching.Redis.csproj"/>
11-
<Project Path="src/SquidStd.Caching/SquidStd.Caching.csproj"/>
12-
<Project Path="src/SquidStd.Core/SquidStd.Core.csproj"/>
13-
<Project Path="src/SquidStd.Database.Abstractions/SquidStd.Database.Abstractions.csproj"/>
14-
<Project Path="src/SquidStd.Database/SquidStd.Database.csproj"/>
15-
<Project Path="src/SquidStd.Messaging.Abstractions/SquidStd.Messaging.Abstractions.csproj"/>
16-
<Project Path="src/SquidStd.Messaging.RabbitMq/SquidStd.Messaging.RabbitMq.csproj"/>
17-
<Project Path="src/SquidStd.Messaging.Sqs/SquidStd.Messaging.Sqs.csproj"/>
18-
<Project Path="src/SquidStd.Messaging/SquidStd.Messaging.csproj"/>
19-
<Project Path="src/SquidStd.Network/SquidStd.Network.csproj"/>
20-
<Project Path="src/SquidStd.Persistence.Abstractions/SquidStd.Persistence.Abstractions.csproj"/>
21-
<Project Path="src/SquidStd.Persistence/SquidStd.Persistence.csproj"/>
22-
<Project Path="src/SquidStd.Persistence.MessagePack/SquidStd.Persistence.MessagePack.csproj"/>
23-
<Project Path="src/SquidStd.Plugin.Abstractions/SquidStd.Plugin.Abstractions.csproj"/>
24-
<Project Path="src/SquidStd.Scripting.Lua/SquidStd.Scripting.Lua.csproj"/>
25-
<Project Path="src/SquidStd.Services.Core/SquidStd.Services.Core.csproj"/>
26-
<Project Path="src/SquidStd.Storage.Abstractions/SquidStd.Storage.Abstractions.csproj"/>
27-
<Project Path="src/SquidStd.Storage.S3/SquidStd.Storage.S3.csproj"/>
28-
<Project Path="src/SquidStd.Storage/SquidStd.Storage.csproj"/>
29-
<Project Path="src/SquidStd.Templating/SquidStd.Templating.csproj"/>
30-
<Project Path="src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj"/>
31-
<Project Path="src/SquidStd.Workers/SquidStd.Workers.csproj"/>
32-
<Project Path="src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj"/>
33-
<Project Path="src/SquidStd.Templates/SquidStd.Templates.csproj"/>
34-
<Project Path="src/SquidStd.Search.Abstractions/SquidStd.Search.Abstractions.csproj"/>
35-
<Project Path="src/SquidStd.Search.Elasticsearch/SquidStd.Search.Elasticsearch.csproj"/>
36-
<Project Path="src/SquidStd.Mail.Abstractions/SquidStd.Mail.Abstractions.csproj"/>
37-
<Project Path="src/SquidStd.Mail.MailKit/SquidStd.Mail.MailKit.csproj"/>
38-
<Project Path="src/SquidStd.Mail.Queue/SquidStd.Mail.Queue.csproj"/>
3+
<Project Path="src/SquidStd.Abstractions/SquidStd.Abstractions.csproj" />
4+
<Project Path="src/SquidStd.Generators/SquidStd.Generators.csproj" />
5+
<Project Path="src/SquidStd.Telemetry.Abstractions/SquidStd.Telemetry.Abstractions.csproj" />
6+
<Project Path="src/SquidStd.Telemetry.OpenTelemetry/SquidStd.Telemetry.OpenTelemetry.csproj" />
7+
<Project Path="src/SquidStd.Aws.Abstractions/SquidStd.Aws.Abstractions.csproj" />
8+
<Project Path="src/SquidStd.Actors/SquidStd.Actors.csproj" />
9+
<Project Path="src/SquidStd.AspNetCore/SquidStd.AspNetCore.csproj" />
10+
<Project Path="src/SquidStd.Caching.Abstractions/SquidStd.Caching.Abstractions.csproj" />
11+
<Project Path="src/SquidStd.Caching.Redis/SquidStd.Caching.Redis.csproj" />
12+
<Project Path="src/SquidStd.Caching/SquidStd.Caching.csproj" />
13+
<Project Path="src/SquidStd.Core/SquidStd.Core.csproj" />
14+
<Project Path="src/SquidStd.Database.Abstractions/SquidStd.Database.Abstractions.csproj" />
15+
<Project Path="src/SquidStd.Database/SquidStd.Database.csproj" />
16+
<Project Path="src/SquidStd.Messaging.Abstractions/SquidStd.Messaging.Abstractions.csproj" />
17+
<Project Path="src/SquidStd.Messaging.RabbitMq/SquidStd.Messaging.RabbitMq.csproj" />
18+
<Project Path="src/SquidStd.Messaging.Sqs/SquidStd.Messaging.Sqs.csproj" />
19+
<Project Path="src/SquidStd.Messaging/SquidStd.Messaging.csproj" />
20+
<Project Path="src/SquidStd.Network/SquidStd.Network.csproj" />
21+
<Project Path="src/SquidStd.Persistence.Abstractions/SquidStd.Persistence.Abstractions.csproj" />
22+
<Project Path="src/SquidStd.Persistence/SquidStd.Persistence.csproj" />
23+
<Project Path="src/SquidStd.Persistence.MessagePack/SquidStd.Persistence.MessagePack.csproj" />
24+
<Project Path="src/SquidStd.Plugin.Abstractions/SquidStd.Plugin.Abstractions.csproj" />
25+
<Project Path="src/SquidStd.Scripting.Lua/SquidStd.Scripting.Lua.csproj" />
26+
<Project Path="src/SquidStd.Services.Core/SquidStd.Services.Core.csproj" />
27+
<Project Path="src/SquidStd.Storage.Abstractions/SquidStd.Storage.Abstractions.csproj" />
28+
<Project Path="src/SquidStd.Storage.S3/SquidStd.Storage.S3.csproj" />
29+
<Project Path="src/SquidStd.Storage/SquidStd.Storage.csproj" />
30+
<Project Path="src/SquidStd.Templating/SquidStd.Templating.csproj" />
31+
<Project Path="src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj" />
32+
<Project Path="src/SquidStd.Workers/SquidStd.Workers.csproj" />
33+
<Project Path="src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj" />
34+
<Project Path="src/SquidStd.Templates/SquidStd.Templates.csproj" />
35+
<Project Path="src/SquidStd.Search.Abstractions/SquidStd.Search.Abstractions.csproj" />
36+
<Project Path="src/SquidStd.Search.Elasticsearch/SquidStd.Search.Elasticsearch.csproj" />
37+
<Project Path="src/SquidStd.Mail.Abstractions/SquidStd.Mail.Abstractions.csproj" />
38+
<Project Path="src/SquidStd.Mail.MailKit/SquidStd.Mail.MailKit.csproj" />
39+
<Project Path="src/SquidStd.Mail.Queue/SquidStd.Mail.Queue.csproj" />
3940
</Folder>
4041
<Folder Name="/samples/">
41-
<Project Path="samples/SquidStd.Samples.WorkerSystem/SquidStd.Samples.WorkerSystem.csproj"/>
42-
<Project Path="samples/SquidStd.Samples.Templating/SquidStd.Samples.Templating.csproj"/>
43-
<Project Path="samples/SquidStd.Samples.Storage/SquidStd.Samples.Storage.csproj"/>
44-
<Project Path="samples/SquidStd.Samples.Search/SquidStd.Samples.Search.csproj"/>
45-
<Project Path="samples/SquidStd.Samples.ScriptingLua/SquidStd.Samples.ScriptingLua.csproj"/>
46-
<Project Path="samples/SquidStd.Samples.Plugins/SquidStd.Samples.Plugins.csproj"/>
47-
<Project Path="samples/SquidStd.Samples.Networking/SquidStd.Samples.Networking.csproj"/>
48-
<Project Path="samples/SquidStd.Samples.Persistence/SquidStd.Samples.Persistence.csproj"/>
49-
<Project Path="samples/SquidStd.Samples.Commands/SquidStd.Samples.Commands.csproj"/>
50-
<Project Path="samples/SquidStd.Samples.Email/SquidStd.Samples.Email.csproj"/>
51-
<Project Path="samples/SquidStd.Samples.Database/SquidStd.Samples.Database.csproj"/>
52-
<Project Path="samples/SquidStd.Samples.AspNetCore/SquidStd.Samples.AspNetCore.csproj"/>
53-
<Project Path="samples/SquidStd.Samples.Messaging/SquidStd.Samples.Messaging.csproj"/>
54-
<Project Path="samples/SquidStd.Samples.Caching/SquidStd.Samples.Caching.csproj"/>
55-
<Project Path="samples/SquidStd.Samples.EventsJobsScheduling/SquidStd.Samples.EventsJobsScheduling.csproj"/>
56-
<Project Path="samples/SquidStd.Samples.GettingStarted/SquidStd.Samples.GettingStarted.csproj"/>
42+
<Project Path="samples/SquidStd.Samples.WorkerSystem/SquidStd.Samples.WorkerSystem.csproj" />
43+
<Project Path="samples/SquidStd.Samples.Templating/SquidStd.Samples.Templating.csproj" />
44+
<Project Path="samples/SquidStd.Samples.Storage/SquidStd.Samples.Storage.csproj" />
45+
<Project Path="samples/SquidStd.Samples.Search/SquidStd.Samples.Search.csproj" />
46+
<Project Path="samples/SquidStd.Samples.ScriptingLua/SquidStd.Samples.ScriptingLua.csproj" />
47+
<Project Path="samples/SquidStd.Samples.Plugins/SquidStd.Samples.Plugins.csproj" />
48+
<Project Path="samples/SquidStd.Samples.Networking/SquidStd.Samples.Networking.csproj" />
49+
<Project Path="samples/SquidStd.Samples.Persistence/SquidStd.Samples.Persistence.csproj" />
50+
<Project Path="samples/SquidStd.Samples.Commands/SquidStd.Samples.Commands.csproj" />
51+
<Project Path="samples/SquidStd.Samples.Email/SquidStd.Samples.Email.csproj" />
52+
<Project Path="samples/SquidStd.Samples.Database/SquidStd.Samples.Database.csproj" />
53+
<Project Path="samples/SquidStd.Samples.AspNetCore/SquidStd.Samples.AspNetCore.csproj" />
54+
<Project Path="samples/SquidStd.Samples.Messaging/SquidStd.Samples.Messaging.csproj" />
55+
<Project Path="samples/SquidStd.Samples.Caching/SquidStd.Samples.Caching.csproj" />
56+
<Project Path="samples/SquidStd.Samples.EventsJobsScheduling/SquidStd.Samples.EventsJobsScheduling.csproj" />
57+
<Project Path="samples/SquidStd.Samples.GettingStarted/SquidStd.Samples.GettingStarted.csproj" />
5758
</Folder>
5859
<Folder Name="/tests/">
59-
<Project Path="tests/SquidStd.Tests/SquidStd.Tests.csproj"/>
60+
<Project Path="tests/SquidStd.Tests/SquidStd.Tests.csproj" />
6061
</Folder>
6162
</Solution>

‎src/SquidStd.Actors/Actor.cs‎

Lines changed: 202 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,202 @@
1+
using System.Collections.Concurrent;
2+
using System.Threading.Tasks.Dataflow;
3+
using Serilog;
4+
using SquidStd.Actors.Data;
5+
using SquidStd.Actors.Interfaces;
6+
using SquidStd.Actors.Types;
7+
using ILogger = Serilog.ILogger;
8+
9+
namespace SquidStd.Actors;
10+
11+
/// <summary>
12+
/// Base class for an actor: a single-consumer mailbox that processes messages in FIFO order on one
13+
/// logical thread, so handler state is mutated without locks. Send fire-and-forget messages with
14+
/// <see cref="TellAsync" /> and request/response messages with <see cref="AskAsync{TRequest,TReply}" />.
15+
/// </summary>
16+
/// <typeparam name="TMessage">The base type of every message this actor accepts.</typeparam>
17+
public abstract class Actor<TMessage> : IAsyncDisposable
18+
{
19+
private readonly ActorOptions _options;
20+
private readonly ActionBlock<TMessage> _mailbox;
21+
private readonly CancellationTokenSource _shutdown;
22+
private readonly ConcurrentDictionary<IActorRequestCore, byte> _outstanding;
23+
private readonly ILogger _logger;
24+
private bool _disposed;
25+
26+
/// <summary>Number of messages waiting in the mailbox.</summary>
27+
public int PendingCount
28+
{
29+
get { return _mailbox.InputCount; }
30+
}
31+
32+
/// <summary>Initializes the actor and starts its mailbox consumer.</summary>
33+
/// <param name="options">Mailbox options; defaults are used when null.</param>
34+
protected Actor(ActorOptions? options = null)
35+
{
36+
_options = options ?? new ActorOptions();
37+
_shutdown = new CancellationTokenSource();
38+
_outstanding = new ConcurrentDictionary<IActorRequestCore, byte>();
39+
_logger = Log.ForContext(GetType());
40+
41+
var blockOptions = new ExecutionDataflowBlockOptions
42+
{
43+
MaxDegreeOfParallelism = 1,
44+
EnsureOrdered = true,
45+
BoundedCapacity = _options.OverflowPolicy == ActorOverflowPolicy.Unbounded
46+
? DataflowBlockOptions.Unbounded
47+
: _options.Capacity,
48+
CancellationToken = _shutdown.Token
49+
};
50+
51+
_mailbox = new ActionBlock<TMessage>(ProcessAsync, blockOptions);
52+
}
53+
54+
/// <summary>Enqueues a fire-and-forget message. Returns false only when dropped (DropNewest).</summary>
55+
/// <param name="message">The message to enqueue.</param>
56+
/// <param name="cancellationToken">Cancels the enqueue wait under the Wait policy.</param>
57+
/// <returns>True when accepted; false when dropped.</returns>
58+
public async ValueTask<bool> TellAsync(TMessage message, CancellationToken cancellationToken = default)
59+
{
60+
ArgumentNullException.ThrowIfNull(message);
61+
ThrowIfDisposed();
62+
63+
if (_options.OverflowPolicy == ActorOverflowPolicy.DropNewest)
64+
{
65+
return _mailbox.Post(message);
66+
}
67+
68+
return await _mailbox.SendAsync(message, cancellationToken);
69+
}
70+
71+
/// <summary>Enqueues a request and awaits its typed reply.</summary>
72+
/// <param name="request">The request message; it is also a <typeparamref name="TMessage" />.</param>
73+
/// <param name="cancellationToken">Acts as a timeout for the reply.</param>
74+
/// <typeparam name="TRequest">The request message type.</typeparam>
75+
/// <typeparam name="TReply">The reply type.</typeparam>
76+
/// <returns>The reply.</returns>
77+
public async Task<TReply> AskAsync<TRequest, TReply>(
78+
TRequest request, CancellationToken cancellationToken = default
79+
)
80+
where TRequest : TMessage, IActorRequest<TReply>
81+
{
82+
ArgumentNullException.ThrowIfNull(request);
83+
ThrowIfDisposed();
84+
85+
_outstanding[request] = 0;
86+
87+
using var registration = cancellationToken.Register(
88+
() => request.Fail(new OperationCanceledException(cancellationToken))
89+
);
90+
91+
try
92+
{
93+
var accepted = await TellAsync(request, cancellationToken);
94+
95+
if (!accepted)
96+
{
97+
request.Fail(new InvalidOperationException("Actor mailbox is full; the request was dropped."));
98+
}
99+
100+
return await request.Completion;
101+
}
102+
finally
103+
{
104+
_outstanding.TryRemove(request, out _);
105+
}
106+
}
107+
108+
/// <summary>Handles a single message. Runs on the mailbox consumer; mutate actor state here freely.</summary>
109+
/// <param name="message">The message to handle.</param>
110+
/// <param name="cancellationToken">Cancelled when the actor is disposed.</param>
111+
protected abstract ValueTask ReceiveAsync(TMessage message, CancellationToken cancellationToken);
112+
113+
/// <summary>Optional hook invoked after an isolated failure. Default is a no-op.</summary>
114+
/// <param name="message">The message whose handler threw.</param>
115+
/// <param name="error">The thrown exception.</param>
116+
protected virtual ValueTask OnErrorAsync(TMessage message, Exception error)
117+
{
118+
return ValueTask.CompletedTask;
119+
}
120+
121+
private async Task ProcessAsync(TMessage message)
122+
{
123+
try
124+
{
125+
await ReceiveAsync(message, _shutdown.Token);
126+
}
127+
catch (OperationCanceledException) when (_shutdown.IsCancellationRequested)
128+
{
129+
if (message is IActorRequestCore cancelledRequest)
130+
{
131+
cancelledRequest.Fail(new OperationCanceledException(_shutdown.Token));
132+
}
133+
}
134+
catch (Exception ex)
135+
{
136+
if (message is IActorRequestCore request)
137+
{
138+
request.Fail(ex);
139+
}
140+
141+
_logger.Error(ex, "Actor {ActorType} failed handling {MessageType}", GetType().Name, message?.GetType().Name);
142+
143+
try
144+
{
145+
await OnErrorAsync(message, ex);
146+
}
147+
catch (Exception hookError)
148+
{
149+
_logger.Error(hookError, "Actor {ActorType} OnError hook threw", GetType().Name);
150+
}
151+
152+
if (_options.ErrorPolicy == ActorErrorPolicy.StopOnError)
153+
{
154+
foreach (var pending in _outstanding.Keys)
155+
{
156+
pending.Fail(new InvalidOperationException("Actor stopped due to a handler failure.", ex));
157+
}
158+
159+
throw;
160+
}
161+
}
162+
}
163+
164+
private void ThrowIfDisposed()
165+
{
166+
if (_disposed)
167+
{
168+
throw new ObjectDisposedException(GetType().Name);
169+
}
170+
}
171+
172+
/// <summary>Completes the mailbox, drains in-flight work, and faults any still-pending requests.</summary>
173+
public async ValueTask DisposeAsync()
174+
{
175+
if (_disposed)
176+
{
177+
return;
178+
}
179+
180+
_disposed = true;
181+
_shutdown.Cancel();
182+
_mailbox.Complete();
183+
184+
try
185+
{
186+
await _mailbox.Completion;
187+
}
188+
catch (Exception ex)
189+
{
190+
_logger.Debug(ex, "Actor {ActorType} mailbox completed with fault during dispose", GetType().Name);
191+
}
192+
193+
foreach (var request in _outstanding.Keys)
194+
{
195+
request.Fail(new ObjectDisposedException(GetType().Name));
196+
}
197+
198+
_outstanding.Clear();
199+
_shutdown.Dispose();
200+
GC.SuppressFinalize(this);
201+
}
202+
}

‎src/SquidStd.Actors/ActorRequest.cs‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
using SquidStd.Actors.Interfaces;
2+
3+
namespace SquidStd.Actors;
4+
5+
/// <summary>
6+
/// Base request record that removes the <see cref="TaskCompletionSource{TResult}" /> boilerplate.
7+
/// Derive from it together with your actor's message interface, e.g.
8+
/// <c>record GetNick : ActorRequest&lt;string?&gt;, ISessionMessage;</c>.
9+
/// </summary>
10+
/// <typeparam name="TReply">The reply type.</typeparam>
11+
public abstract record ActorRequest<TReply> : IActorRequest<TReply>
12+
{
13+
private readonly TaskCompletionSource<TReply> _completion =
14+
new(TaskCreationOptions.RunContinuationsAsynchronously);
15+
16+
/// <inheritdoc />
17+
public Task<TReply> Completion
18+
{
19+
get { return _completion.Task; }
20+
}
21+
22+
/// <inheritdoc />
23+
public void Reply(TReply value)
24+
{
25+
_completion.TrySetResult(value);
26+
}
27+
28+
/// <inheritdoc />
29+
public void Fail(Exception error)
30+
{
31+
_completion.TrySetException(error);
32+
}
33+
}

‎src/SquidStd.Actors/Data/ActorOptions.cs‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
using SquidStd.Actors.Types;
2+
3+
namespace SquidStd.Actors.Data;
4+
5+
/// <summary>
6+
/// Configuration for an <see cref="SquidStd.Actors.Actor{TMessage}" /> mailbox.
7+
/// </summary>
8+
public sealed class ActorOptions
9+
{
10+
/// <summary>Bounded mailbox capacity (ignored when <see cref="OverflowPolicy" /> is Unbounded).</summary>
11+
public int Capacity { get; init; } = 1024;
12+
13+
/// <summary>Behavior when the mailbox is full.</summary>
14+
public ActorOverflowPolicy OverflowPolicy { get; init; } = ActorOverflowPolicy.Wait;
15+
16+
/// <summary>Behavior when a fire-and-forget handler throws.</summary>
17+
public ActorErrorPolicy ErrorPolicy { get; init; } = ActorErrorPolicy.Isolate;
18+
}

0 commit comments

Comments
 (0)