How to tell when all actors idle?

[pwhe23]

I have an application that queues up a bunch of messages for some actors to run when it starts up. How can I tell when all the messages in the system have been processed so I can then shut it down?

[Aaronontheweb]

@pwhe23 best way to do this would be a "roll up" process:

1. Each individual actor uses a ReceiveTimeout to test for when it has become idle;
2. Parent actor manages all of the worker actors that use the ReceiveTimeout - parent actor calls Context.Watch on each child so it will receive a Terminated message when that child actor dies;
3. When each child receives its ReceiveTimeout message, it kills itself via Context.Stop - parent receives the Terminated message and then calls Context.GetChildren (or decrements an internal counter) to see how many children are still alive.
4. When the parent actor notices that the number of living children == 0, parent actor can kill itself or shut down the entire ActorSystem if that's what you want.

That would be a tidy, elastically-configurable way of solving this issue.

[Aaronontheweb]

If you have multiple areas of the actor hierarchy that need to do this, have all of them implement the pattern and then have a single actor that calls Context.Watch on the parent actor for each portion of the hierarchy.

[pwhe23]

That pattern worked pretty great. Here's a little LinqPad script I threw together to test it out for anyone interested:

//nuget:Akka.DependencyInjection
//nuget:Microsoft.Extensions.Hosting
using Akka.Actor;
using Akka.DependencyInjection;
using Akka.Event;
using Akka.Routing;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

//REF: https://getakka.net/articles/actors/dependency-injection.html
//REF: https://petabridge.com/blog/akkadotnet-aspnetcore/
async Task Main(string[] args)
{
    var host = Host.CreateDefaultBuilder(args)
        .ConfigureServices(services =>
        {
            services.AddSingleton<AkkaService>();
            services.AddHostedService(x => x.GetRequiredService<AkkaService>());
        })
        .Build();
    await host.StartAsync();

    var akka = host.Services.GetRequiredService<AkkaService>();
    foreach (var id in Enumerable.Range(1, 100))
    {
        await Task.Delay(TimeSpan.FromMilliseconds(100));
        akka.TaskActor.Tell(new Start(id));
    }

    await host.WaitForShutdownAsync();
    Console.WriteLine("[Exit]");
}

public class AkkaService : IHostedService
{
    private readonly IServiceProvider _ioc;
    private ActorSystem _akka = null!;

    public AkkaService(IServiceProvider sp)
    {
        _ioc = sp;
    }

    public IActorRef TaskActor { get; private set; } = null!;

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _akka = ActorSystem.Create("Akka", BootstrapSetup.Create().And(DependencyResolverSetup.Create(_ioc)));
        TaskActor = _akka.ActorOf(DependencyResolver.For(_akka).Props<TaskActor>(), "Parent");
        return Task.CompletedTask;
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        await _akka.Terminate();
    }
};

record Start(int Id);
record Work(int Id);

public class TaskActor : ReceiveActor
{
    public TaskActor(IHostApplicationLifetime host)
    {
        Receive<Start>(x => StartChild(x, host));
        ReceiveAsync<Work>(DoWorkAsync);
        Receive<Terminated>(_ => ChildTerminated(host));
    }

    private void StartChild(Start task, IHostApplicationLifetime host)
    {
        Console.WriteLine($"[Start] {task.Id}");
        var childActor = Context.ActorOf(Props.Create(() => new TaskActor(host)), $"Child{task.Id}");
        childActor.Tell(new Work(task.Id), Self);
        Context.Watch(childActor);
    }

    private async Task DoWorkAsync(Work task)
    {
        Console.WriteLine($"[Work] {task.Id}");
        await Task.Delay(TimeSpan.FromSeconds(Random.Shared.Next(1, 10)));
        Console.WriteLine($"[Done] {task.Id}");
        Context.Stop(Self);
    }

    private void ChildTerminated(IHostApplicationLifetime host)
    {
        if (!Context.GetChildren().Any()) host.StopApplication();
    }
};