EventStream subscription does not get automatically removed on actor death causing memory leak

[harmonyitinc]

Akka.Net v1.4.34 (was discovered in v1.4.17 as that's what we were using but verified to still exist in latest)

Running on .Net Core 3.1

If an Actor subscribes to a message Type on the EventStream and doesn't unsubscribe before dying the actor is garbage collected but the subscription remains and it keeps the ActorCell alive forever along with the Props used to create it.

In our app we have a set of Job actors that are loaded on demand and stay in memory until idle for a time at which point they kill themselves. This works great except the Props (ours are unique per instance due to a GUID in the parameters) and the ActorCell stay in memory forever due to the EventStream subscription never getting freed up. The EventStream subscription dictionaries continue to grow in size as well. We have 1000s of Job actors churning 24/7 so in a 12 hour period, I measured on average 2000 live Job actors at any given moment and 68,000 dead and gone at the end of the period. However 70,000 ActorCells and Props were still alive resulting in about 30MB/24 hours memory leakage at the current churn rate. The total memory leakage depends on the ActorCell overhead and the size of the Props as the contents of the Props are kept alive too - we had 7000 extra GUIDs alive after 12 hours too due to this as they are in the Props.

To repro this I wrote a test program that slowly creates and then slowly kills 100 actors (slow so I can see what's going on and snapshot it easily) and can be configured to either not subscribe at all, subscribe in PreStart and unsubscribe in PostStop or to only subscribe and not unsubscribe. I monitored the memory with SciTech Memory Profiler snapshotting before, during and after for each case.

If set to either not subscribe at all or sub and unsub all is well. All Props and ActorCells are gone except for a few ones used by Akka internally and my Manager Actor that creates my 100 instances.

If instances subscribe and are killed without unsubscribing (like our app), all ActorCells and Props stay around forever although the Actors are gone. It's the EventStream subscriptions holding the reference to the IActorRef (ActorCell) that's doing it. As you can see below I've got 107 ActorCells left. There's just the expected 7 when I unsub manually or don't sub at all.

I ran my test app linked directly with the Akka.Net source code so I could debug into the Akka code and the message to Register to unsubscribe is never received by the EventBusUnsubscriber so it never Context.Watch the actors therefore it doesn't care if they die and are still subscribed. I verified the EventBusUnsubscriber is running it just never does anything. I am unable to figure out why the Register message is never received but I think it never gets sent due to some issue with the code in EventStream.cs RegisterWithUnsubscriber. The IF never seems to evaluate to TRUE that I've seen. I can't figure out what this code is trying to avoid doing but it seems to avoid it all the time...it's too clever for me :-)

Explicitly unsubscribing Self in the PostStop of the Actor 100% fixes the issue.

Here is the code I believe the problem lies in.
Thanks
Tom

*edit - added Repro

        private void RegisterWithUnsubscriber(IActorRef subscriber)
        {
            _initiallySubscribedOrUnsubscriber.Match().With<Left<IImmutableSet<IActorRef>>>(v =>
            {
                if (_initiallySubscribedOrUnsubscriber.CompareAndSet(v,
                    Either.Left<IImmutableSet<IActorRef>>(v.Value.Add(subscriber))))
                {
                    RegisterWithUnsubscriber(subscriber);  // <---- this line never gets called
                }

            }).With<Right<IActorRef>>(unsubscriber =>
            {
                unsubscriber.Value.Tell( new EventStreamUnsubscriber.Register(subscriber)); // <---- neither does this
            });
        }

AkkaMemoryLeakRepro.zip

[Aaronontheweb]

We automatically do this today

akka.net/src/core/Akka/Event/EventBusUnsubscriber.cs

Lines 29 to 149 in 7123d0f

	class EventStreamUnsubscriber : ActorBase
	{
	private readonly EventStream _eventStream;
	private readonly bool _debug;
	private readonly ActorSystem _system;
	/// <summary>
	/// TBD
	/// </summary>
	/// <param name="eventStream">TBD</param>
	/// <param name="system">TBD</param>
	/// <param name="debug">TBD</param>
	public EventStreamUnsubscriber(EventStream eventStream, ActorSystem system, bool debug)
	{
	_eventStream = eventStream;
	_system = system;
	_debug = debug;
	}
	/// <summary>
	/// TBD
	/// </summary>
	/// <param name="message">TBD</param>
	/// <returns>TBD</returns>
	protected override bool Receive(object message)
	{
	return message.Match().With<Register>(register =>
	{
	if (_debug)
	_eventStream.Publish(new Debug(this.GetType().Name, GetType(),
	string.Format("watching {0} in order to unsubscribe from EventStream when it terminates", register.Actor)));
	Context.Watch(register.Actor);
	}).With<UnregisterIfNoMoreSubscribedChannels>(unregister =>
	{
	if (_debug)
	_eventStream.Publish(new Debug(this.GetType().Name, GetType(),
	string.Format("unwatching {0} since has no subscriptions", unregister.Actor)));
	Context.Unwatch(unregister.Actor);
	}).With<Terminated>(terminated =>
	{
	if (_debug)
	_eventStream.Publish(new Debug(this.GetType().Name, GetType(),
	string.Format("unsubscribe {0} from {1}, because it was terminated", terminated.Actor , _eventStream )));
	_eventStream.Unsubscribe(terminated.Actor);
	})
	.WasHandled;
	}
	/// <summary>
	/// TBD
	/// </summary>
	protected override void PreStart()
	{
	if (_debug)
	_eventStream.Publish(new Debug(this.GetType().Name, GetType(),
	string.Format("registering unsubscriber with {0}", _eventStream)));
	_eventStream.InitUnsubscriber(Self, _system);
	}
	/// <summary>
	/// TBD
	/// </summary>
	internal class Register
	{
	/// <summary>
	/// TBD
	/// </summary>
	/// <param name="actor">TBD</param>
	public Register(IActorRef actor)
	{
	Actor = actor;
	}
	/// <summary>
	/// TBD
	/// </summary>
	public IActorRef Actor { get; private set; }
	}
	/// <summary>
	/// TBD
	/// </summary>
	internal class Terminated
	{
	/// <summary>
	/// TBD
	/// </summary>
	/// <param name="actor">TBD</param>
	public Terminated(IActorRef actor)
	{
	Actor = actor;
	}
	/// <summary>
	/// TBD
	/// </summary>
	public IActorRef Actor { get; private set; }
	}
	/// <summary>
	/// TBD
	/// </summary>
	internal class UnregisterIfNoMoreSubscribedChannels
	{
	/// <summary>
	/// TBD
	/// </summary>
	/// <param name="actor">TBD</param>
	public UnregisterIfNoMoreSubscribedChannels(IActorRef actor)
	{
	Actor = actor;
	}
	/// <summary>
	/// TBD
	/// </summary>
	public IActorRef Actor { get; private set; }
	}
	}

At least we are supposed to. I'll go take a look.

[Aaronontheweb]

Reproduced here: #5720

[harmonyitinc]

Great! Yes, I should have said the documentation says you do it automatically but it doesn't work...as you've now reproduced too. I couldn't figure out how the code I included above works for sending the registration message, just that it doesn't appear to ever send it to the unsubscriber...I did verify the unsubscriber actor itself is running, but it doesn't react to eventstream subscriber actor's death

[Aaronontheweb]

I did verify the unsubscriber actor itself is running, but it doesn't react to eventstream subscriber actor's death

I believe the answer for that is this code:

akka.net/src/core/Akka/Event/EventStream.cs

Lines 131 to 159 in d055f46

	return _initiallySubscribedOrUnsubscriber.Match().With<Left<IImmutableSet<IActorRef>>>(v =>
	{
	if (_initiallySubscribedOrUnsubscriber.CompareAndSet(v, Either.Right(unsubscriber)))
	{
	if (_debug)
	{
	Publish(new Debug(SimpleName(this), GetType(),
	string.Format("initialized unsubscriber to: {0} registering {1} initial subscribers with it", unsubscriber, v.Value.Count)));
	}
	v.Value.ForEach(RegisterWithUnsubscriber);
	}
	else
	{
	InitUnsubscriber(unsubscriber, system);
	}
	}).With<Right<IActorRef>>(presentUnsubscriber =>
	{
	if (_debug)
	{
	Publish(new Debug(SimpleName(this), GetType(),
	string.Format("not using unsubscriber {0}, because already initialized with {1}", unsubscriber, presentUnsubscriber)));
	}
	}).WasHandled;

I've been debugging concurrent + distributed systems full-time in .NET for about a decade - I have no idea how on earth that code is supposed to be remotely safe, let alone how it's supposed to work in a linear non-concurrent flow. I'm going to simplify it.

[Aaronontheweb]

akka.net/src/core/Akka/Event/EventBusUnsubscriber.cs

Lines 110 to 128 in d055f46

	/// <summary>
	/// TBD
	/// </summary>
	internal class Terminated
	{
	/// <summary>
	/// TBD
	/// </summary>
	/// <param name="actor">TBD</param>
	public Terminated(IActorRef actor)
	{
	Actor = actor;
	}
	/// <summary>
	/// TBD
	/// </summary>
	public IActorRef Actor { get; private set; }
	}

- found another possible cause.

The unsubscriber is handling its own custom Terminated message - not the one generated by Context.Watch.

[Aaronontheweb]

Have this all fixed now in #5720