How to avoid message loss while sending via a consistent hash based router with local actors? #6944
Replies: 3 comments 11 replies
-
|
In Akka.NET, in order to use a consistent hashing router you must either:
Which of these are you using? |
Beta Was this translation helpful? Give feedback.
-
|
Just tried to reproduce it here: async Task Main()
{
var myActorSystem = ActorSystem.Create("Repro");
var poolSize = 25;
var chr = myActorSystem.ActorOf(Props.Create(() => new ChrTestActor()).WithRouter(new ConsistentHashingPool(poolSize)), "ChrRouters");
var chrCounter = myActorSystem.ActorOf(Props.Create(() => new ChrTestResultsGatherer(chr, poolSize)), "testResults");
var testMessage = new PerformReadAfterWriteLatencyTest(Random.Shared.Next(1000, 10000), 1, 10, 1, 1);
for(int i = 0; i < 10_000; i++){
chr.Tell(testMessage with { CampaignId = Random.Shared.Next(1000, 10000) });
}
chrCounter.Tell(Finished.Instance);
await myActorSystem.WhenTerminated;
}
internal sealed record PerformReadAfterWriteLatencyTest(int CampaignId, int PublisherId, int SpendAmount, int UnixTimeStampSinceEpoch, int SharedBudgetId) : IConsistentHashable
{
public object ConsistentHashKey => CampaignId;
}
internal sealed class Finished{
public static readonly Finished Instance = new();
private Finished(){}
}
internal sealed class ChrTestResultsGatherer : ReceiveActor{
private readonly int _actorCount;
private readonly IActorRef _chrRouter;
private readonly ILoggingAdapter _log = Context.GetLogger();
public ChrTestResultsGatherer(IActorRef chrRouter, int actorCount){
_actorCount = actorCount;
_chrRouter = chrRouter;
Waiting();
}
public void Waiting(){
Receive<Finished>(f => {
// ping all recipients
_chrRouter.Tell(new Broadcast(Finished.Instance));
Become(() => Counting(_actorCount));
});
}
public void Counting(int remainingActors){
var dict = new Dictionary<IActorRef, int>();
ReceiveAsync<int>(async i => {
dict[Sender] = i;
if (dict.Count == _actorCount)
{
var c = 0;
foreach (var d in dict) {
_log.Info("Actor [{0}]: {1}", c, d.Value);
c++;
}
var totalMsgsProcessed = dict.Values.Sum();
_log.Info("Total messages processed: [{0}]", totalMsgsProcessed);
await Context.System.Terminate();
}
});
}
}
internal sealed class ChrTestActor : ReceiveActor{
private int _count = 0;
private readonly ILoggingAdapter _log = Context.GetLogger();
public ChrTestActor(){
Receive<PerformReadAfterWriteLatencyTest>(r => {
_count++;
});
Receive<Finished>(_ => {
Sender.Tell(_count);
});
}
}Produced the following: |
Beta Was this translation helpful? Give feedback.
-
|
Like I said - very unlikely this is an Akka.NET issue; that code has been used very heavily for many years. |
Beta Was this translation helpful? Give feedback.
-
|
If we assume that due to some issue, one of worker of CHR is restarted, wouldn't CHR create a new worker and process remaining messages? or we just lose all message pending in queue of worker that stopped? Also I reading between lines, it seems that it is guarantee that no message is lost when sent to local actor either directly or via router? Is this right assumption? It is quite important for me as I am trying to build a distributed application that keep track of Budget spend in real time. |
Beta Was this translation helpful? Give feedback.
-
|
Which version of Akka.NET are you using? Because there was a router / routee supervision bug we fixed late last year in Akka.NET v1.4.50: #6295 - you could be running into that issue if you are using an older version of Akka.NET. |
Beta Was this translation helpful? Give feedback.
-
Messages can only be lost, when working with local actors, if the actor is terminated before it has a chance to process the message (via |
Beta Was this translation helpful? Give feedback.
-
|
I am using 1.5.12.
So does it mean that all enqueued messages are lost too on restart? I
though restart use same message queue, so messages are not actually lost
(except atmost 1 in flight message). This is interesting if restart means
to have a new queue and hence all messages are lost
…On Wed, Oct 4, 2023 at 1:08 PM Aaron Stannard ***@***.***> wrote:
Also I reading between lines, it seems that it is guarantee that no
message is lost when sent to local actor either directly or via router? Is
this right assumption? It is quite important for me as I am trying to build
a distributed application that keep track of Budget spend in real time.
Messages can only be lost, when working with local actors, if the actor is
terminated before it has a chance to process the message (via Context.Stop
or a Directive.Stop or by a parent actor restarting, which was the router
issue we fixed in v1.4.50.) All in-memory messaging is facilitated by
ConcurrentQueue<T> under the covers.
—
Reply to this email directly, view it on GitHub
<#6944 (reply in thread)>,
or unsubscribe
<https://github.com/notifications/unsubscribe-auth/AEJNUH6XFPMHV66463RHZI3X5W645AVCNFSM6AAAAAA5RU5KCWVHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM3TCOJRGA3TI>
.
You are receiving this because you were mentioned.Message ID:
***@***.***>
|
Beta Was this translation helpful? Give feedback.
-
No - messages are not lost on restart: https://www.youtube.com/watch?v=CC2XdYlpmvo |
Beta Was this translation helpful? Give feedback.
-
|
Restart != Terminated Restart == transient error handling + recovery |
Beta Was this translation helpful? Give feedback.
-
|
When a parent actor restarts, by default of its children will be terminated - thus, having parent actors carry out work that can result in exceptions being thrown is considered to be code smell in Akka.NET usually. |
Beta Was this translation helpful? Give feedback.
-
|
Thank you for the clarification and appreciate the quick responses . Let me
add more logging to find if something is causing termination and thus
message loss.
…On Wed, Oct 4, 2023 at 1:33 PM Aaron Stannard ***@***.***> wrote:
When a parent actor restarts, by default of its children will be
terminated - thus, having parent actors carry out work that can result in
exceptions being thrown is considered to be code smell in Akka.NET usually.
—
Reply to this email directly, view it on GitHub
<#6944 (reply in thread)>,
or unsubscribe
<https://github.com/notifications/unsubscribe-auth/AEJNUH6T73LBLYWX7SOG2P3X5XB2RAVCNFSM6AAAAAA5RU5KCWVHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM3TCOJRGI2DI>
.
You are receiving this because you were mentioned.Message ID:
***@***.***>
|
Beta Was this translation helpful? Give feedback.
-
Watch the YouTube video - we explicitly show in there that restarting doesn't affect messages that are in the queue. |
Beta Was this translation helpful? Give feedback.
-
I am using Akka.NET to create a consistent hash based router . The router is created locally and use local actors. There is no remote actor, all actors are on a single machine and system. However, I noticed that I am losing 50% of the messages sent to the local actor. This is surprising to me, because I read in this blog post from petabridge that “All in-memory message passing within one application, for instance, is going to be guaranteed unless you hit an OutOfMemoryException or any of the other standard CLR failures”. I do not see any exception in my application.
How can I ensure that local messages are always delivered exactly once on my local actor?
Here is code snippet that reproduce this issue
` var consistentHashRouter = mySystem.ActorOf(
Props.Create(randomNodeRouter, metricCollectorActor)
.WithRouter(new ConsistentHashingPool(workerCount)),
"ConcurrentPool");
// generate 100 random campaign ids
Random random = new Random();
for (int j = 0; j < numberOfTimesToRun; j++)
{
var testMessage = new PerformReadAfterWriteLatencyTest(random.Next(1000, 10000), 1, 10);
consistentHashRouter.Tell(testMessage);
Console.WriteLine($"Sent message for test run {j}: {testMessage.ToString()}");
}`
Beta Was this translation helpful? Give feedback.
All reactions