enable ChannelTaskScheduler to work inside Akka.Cluster without cau…

…sing errors inside `/system` actors (#5861) * close #5498 enable `ChannelTaskScheduler` to work inside Akka.Cluster without causing errors inside `/system` actors * fix `HeartbeatSender` * cleaned up SBR internals (style) * cleaned up some comments * asynchronously attempt to acquire `Cluster` inside SBR * fixed SBR compilation * Update SplitBrainResolver.cs * subscribe on PreStart
akkadotnet · May 5, 2022 · f60f585 · f60f585
1 parent 520a4af
commit f60f585
Show file tree

Hide file tree

Showing 9 changed files with 84 additions and 87 deletions.
diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
@@ -1,3 +1,6 @@
+#### 1.4.38 April 14 2022 ####
+**Placeholder for nightlies**
+
 #### 1.4.37 April 14 2022 ####
 Akka.NET v1.4.37 is a minor release that contains some minor bug fixes.
 

diff --git a/src/common.props b/src/common.props
@@ -2,7 +2,7 @@
   <PropertyGroup>
     <Copyright>Copyright © 2013-2021 Akka.NET Team</Copyright>
     <Authors>Akka.NET Team</Authors>
-    <VersionPrefix>1.4.34</VersionPrefix>
+    <VersionPrefix>1.4.38</VersionPrefix>
     <PackageIcon>akkalogo.png</PackageIcon>
     <PackageProjectUrl>https://github.com/akkadotnet/akka.net</PackageProjectUrl>
     <PackageLicenseUrl>https://github.com/akkadotnet/akka.net/blob/master/LICENSE</PackageLicenseUrl>

diff --git a/src/core/Akka.Cluster.Tests/ClusterHeartbeatReceiverSpec.cs b/src/core/Akka.Cluster.Tests/ClusterHeartbeatReceiverSpec.cs
@@ -27,7 +27,7 @@ public ClusterHeartbeatReceiverSpec(ITestOutputHelper output)
         [Fact]
         public void ClusterHeartbeatReceiver_should_respond_to_heartbeats_with_same_SeqNo_and_SendTime()
         {
-            var heartbeater = Sys.ActorOf(ClusterHeartbeatReceiver.Props(() => Cluster.Get(Sys)));
+            var heartbeater = Sys.ActorOf(ClusterHeartbeatReceiver.Props(Cluster.Get(Sys)));
             heartbeater.Tell(new Heartbeat(Cluster.Get(Sys).SelfAddress, 1, 2));
             ExpectMsg<HeartbeatRsp>(new HeartbeatRsp(Cluster.Get(Sys).SelfUniqueAddress, 1, 2));
         }

diff --git a/src/core/Akka.Cluster.Tests/ClusterHeartbeatSenderSpec.cs b/src/core/Akka.Cluster.Tests/ClusterHeartbeatSenderSpec.cs
@@ -23,7 +23,7 @@ class TestClusterHeartbeatSender : ClusterHeartbeatSender
         {
             private readonly TestProbe _probe;
 
-            public TestClusterHeartbeatSender(TestProbe probe)
+            public TestClusterHeartbeatSender(TestProbe probe) : base(Cluster.Get(Context.System))
             {
                 _probe = probe;
             }

diff --git a/src/core/Akka.Cluster/Cluster.cs b/src/core/Akka.Cluster/Cluster.cs
@@ -520,7 +520,7 @@ public ImmutableHashSet<string> SelfRoles
         public DefaultFailureDetectorRegistry<Address> FailureDetector { get; }
 
         /// <summary>
-        /// TBD
+        /// The downing provider used to execute automatic downing inside Akka.Cluster.
         /// </summary>
         public IDowningProvider DowningProvider => _downingProvider.Value;
 
@@ -535,7 +535,6 @@ public ImmutableHashSet<string> SelfRoles
 
         private static IScheduler CreateScheduler(ActorSystem system)
         {
-            //TODO: Whole load of stuff missing here!
             return system.Scheduler;
         }
 

diff --git a/src/core/Akka.Cluster/ClusterDaemon.cs b/src/core/Akka.Cluster/ClusterDaemon.cs
@@ -895,7 +895,7 @@ private void CreateChildren(Cluster cluster)
             _cluster = cluster;
             _coreSupervisor = Context.ActorOf(Props.Create<ClusterCoreSupervisor>(), "core");
 
-            Context.ActorOf(ClusterHeartbeatReceiver.Props(() => _cluster), "heartbeatReceiver");
+            Context.ActorOf(ClusterHeartbeatReceiver.Props(cluster), "heartbeatReceiver");
         }
 
         protected override void PostStop()
@@ -1333,7 +1333,7 @@ private void BecomeInitialized()
         {
             // start heartbeatSender here, and not in constructor to make sure that
             // heartbeating doesn't start before Welcome is received
-            Context.ActorOf(Props.Create<ClusterHeartbeatSender>().WithDispatcher(_cluster.Settings.UseDispatcher),
+            Context.ActorOf(Props.Create(() => new ClusterHeartbeatSender(_cluster)).WithDispatcher(_cluster.Settings.UseDispatcher),
                 "heartbeatSender");
             // make sure that join process is stopped
             StopSeedNodeProcess();

diff --git a/src/core/Akka.Cluster/ClusterHeartbeat.cs b/src/core/Akka.Cluster/ClusterHeartbeat.cs
@@ -26,16 +26,16 @@ internal sealed class ClusterHeartbeatReceiver : UntypedActor
     {
         // Important - don't use Cluster.Get(Context.System) in constructor because that would
         // cause deadlock. See startup sequence in ClusterDaemon.
-        private readonly Lazy<Cluster> _cluster;
+        private readonly Cluster _cluster;
 
-        public bool VerboseHeartbeat => _cluster.Value.Settings.VerboseHeartbeatLogging;
+        public bool VerboseHeartbeat => _cluster.Settings.VerboseHeartbeatLogging;
 
         /// <summary>
         /// TBD
         /// </summary>
-        public ClusterHeartbeatReceiver(Func<Cluster> getCluster)
+        public ClusterHeartbeatReceiver(Cluster cluster)
         {
-            _cluster = new Lazy<Cluster>(getCluster);
+            _cluster = cluster;
         }
 
         protected override void OnReceive(object message)
@@ -44,8 +44,8 @@ protected override void OnReceive(object message)
             {
                 case ClusterHeartbeatSender.Heartbeat hb:
                     // TODO log the sequence nr once serializer is enabled
-                    if(VerboseHeartbeat) _cluster.Value.CurrentInfoLogger.LogDebug("Heartbeat from [{0}]", hb.From);
-                    Sender.Tell(new ClusterHeartbeatSender.HeartbeatRsp(_cluster.Value.SelfUniqueAddress,
+                    if(VerboseHeartbeat) _cluster.CurrentInfoLogger.LogDebug("Heartbeat from [{0}]", hb.From);
+                    Sender.Tell(new ClusterHeartbeatSender.HeartbeatRsp(_cluster.SelfUniqueAddress,
                         hb.SequenceNr, hb.CreationTimeNanos));
                     break;
                 default:
@@ -54,7 +54,7 @@ protected override void OnReceive(object message)
             }
         }
 
-        public static Props Props(Func<Cluster> getCluster)
+        public static Props Props(Cluster getCluster)
         {
             return Akka.Actor.Props.Create(() => new ClusterHeartbeatReceiver(getCluster));
         }
@@ -76,11 +76,12 @@ internal class ClusterHeartbeatSender : ReceiveActor
         private DateTime _tickTimestamp;
 
         /// <summary>
-        /// TBD
+        /// Create a new instance of the <see cref="ClusterHeartbeatSender"/> and pass in a reference to the <see cref="Cluster"/>
+        /// to which it belongs.
         /// </summary>
-        public ClusterHeartbeatSender()
+        public ClusterHeartbeatSender(Cluster cluster)
         {
-            _cluster = Cluster.Get(Context.System);
+            _cluster = cluster;
             var tickInitialDelay = _cluster.Settings.PeriodicTasksInitialDelay.Max(_cluster.Settings.HeartbeatInterval);
             _tickTimestamp = DateTime.UtcNow + tickInitialDelay;