From ca2e232705bcad5af9ea84c3d5b95b82c72e797f Mon Sep 17 00:00:00 2001
From: Petri Kero <petrikero2@gmail.com>
Date: Mon, 9 Mar 2020 21:25:24 +0200
Subject: [PATCH] Fix atomic updates of AddressTerminatedTopic._subscribers.
 Also, optimize the implementation to not take copies on Subscribe() or
 Unsubscribe().

---
 src/core/Akka/Event/AddressTerminatedTopic.cs | 30 ++++++++-----------
 1 file changed, 12 insertions(+), 18 deletions(-)

diff --git a/src/core/Akka/Event/AddressTerminatedTopic.cs b/src/core/Akka/Event/AddressTerminatedTopic.cs
index b13d366acdc..d16fd7cf756 100644
--- a/src/core/Akka/Event/AddressTerminatedTopic.cs
+++ b/src/core/Akka/Event/AddressTerminatedTopic.cs
@@ -5,7 +5,9 @@
 // </copyright>
 //-----------------------------------------------------------------------
 
+using System;
 using System.Collections.Generic;
+using System.Linq;
 using Akka.Actor;
 using Akka.Util;
 
@@ -35,7 +37,7 @@ public override AddressTerminatedTopic CreateExtension(ExtendedActorSystem syste
     /// </summary>
     internal sealed class AddressTerminatedTopic : IExtension
     {
-        private readonly AtomicReference<HashSet<IActorRef>> _subscribers = new AtomicReference<HashSet<IActorRef>>(new HashSet<IActorRef>());
+        private readonly HashSet<IActorRef> _subscribers = new HashSet<IActorRef>();
 
         /// <summary>
         /// Retrieves the extension from the specified actor system.
@@ -53,13 +55,8 @@ public static AddressTerminatedTopic Get(ActorSystem system)
         /// <param name="subscriber">The actor that is registering for notifications.</param>
         public void Subscribe(IActorRef subscriber)
         {
-            while (true)
-            {
-                var current = _subscribers;
-                if (!_subscribers.CompareAndSet(current, new HashSet<IActorRef>(current.Value) {subscriber}))
-                    continue;
-                break;
-            }
+            lock (_subscribers)
+                _subscribers.Add(subscriber);
         }
 
         /// <summary>
@@ -68,15 +65,8 @@ public void Subscribe(IActorRef subscriber)
         /// <param name="subscriber">The actor that is unregistering for notifications.</param>
         public void Unsubscribe(IActorRef subscriber)
         {
-            while (true)
-            {
-                var current = _subscribers;
-                var newSet = new HashSet<IActorRef>(_subscribers.Value);
-                newSet.Remove(subscriber);
-                if (!_subscribers.CompareAndSet(current, newSet))
-                    continue;
-                break;
-            }
+            lock (_subscribers)
+                _subscribers.Remove(subscriber);
         }
 
         /// <summary>
@@ -85,7 +75,11 @@ public void Unsubscribe(IActorRef subscriber)
         /// <param name="msg">The message that is sent to all subscribers.</param>
         public void Publish(AddressTerminated msg)
         {
-            foreach (var subscriber in _subscribers.Value)
+            List<IActorRef> subscribers;
+            lock(_subscribers)
+                subscribers = _subscribers.ToList();
+
+            foreach (var subscriber in subscribers)
             {
                 subscriber.Tell(msg, ActorRefs.NoSender);
             }