From aed2cbcfe4e397b3306754a2d38a0974522a1aa6 Mon Sep 17 00:00:00 2001
From: Aaron Stannard <aaron@petabridge.com>
Date: Mon, 19 Apr 2021 17:30:49 -0500
Subject: [PATCH] Turned `HeatbeatNodeRing` into `struct` (#4944)

* added benchmark for HeartbeatNodeRing performance

* switched to local function

No perf change

* approve Akka.Benchmarks friend assembly for Akka.Cluster

* remove HeartbeatNodeRing.NodeRing() allocation and make field immutable

* made it so Akka.Util.Internal.ArrayExtensions.From no longer allocates (much)

* added some descriptive comments on HeartbeatNodeRing.Receivers

* Replaced `Lazy<T>` with `Option<T>` and a similar lazy initialization check

 Improved throughput by ~10% on larger collections and further reduced memory allocation.

* changed return types to `IImmutableSet`

Did this in order to reduce allocations from constantly converting back and forth from `ImmutableSortedSet<T>` and `ImmutableHashSet<T>` - that way we can just use whatever the underlying collection type is.

* converted `HeartbeatNodeRing` into a `struct`

improved performance some, but I don't want to lump it in with other changes just in case
---
 src/core/Akka.Cluster/ClusterHeartbeat.cs | 43 ++++++++++++-----------
 1 file changed, 23 insertions(+), 20 deletions(-)

diff --git a/src/core/Akka.Cluster/ClusterHeartbeat.cs b/src/core/Akka.Cluster/ClusterHeartbeat.cs
index 035cffc3ace..0ed134262fa 100644
--- a/src/core/Akka.Cluster/ClusterHeartbeat.cs
+++ b/src/core/Akka.Cluster/ClusterHeartbeat.cs
@@ -552,7 +552,7 @@ public ClusterHeartbeatSenderState HeartbeatRsp(UniqueAddress from)
         /// <param name="oldReceiversNowUnreachable">TBD</param>
         /// <param name="failureDetector">TBD</param>
         /// <returns>TBD</returns>
-        public ClusterHeartbeatSenderState Copy(HeartbeatNodeRing ring = null, ImmutableHashSet<UniqueAddress> oldReceiversNowUnreachable = null, IFailureDetectorRegistry<Address> failureDetector = null)
+        public ClusterHeartbeatSenderState Copy(HeartbeatNodeRing? ring = null, ImmutableHashSet<UniqueAddress> oldReceiversNowUnreachable = null, IFailureDetectorRegistry<Address> failureDetector = null)
         {
             return new ClusterHeartbeatSenderState(ring ?? Ring, oldReceiversNowUnreachable ?? OldReceiversNowUnreachable, failureDetector ?? FailureDetector);
         }
@@ -566,7 +566,7 @@ public ClusterHeartbeatSenderState Copy(HeartbeatNodeRing ring = null, Immutable
     /// 
     /// It is immutable, i.e. the methods all return new instances.
     /// </summary>
-    internal sealed class HeartbeatNodeRing
+    internal struct HeartbeatNodeRing
     {
         private readonly bool _useAllAsReceivers;
         private Option<IImmutableSet<UniqueAddress>> _myReceivers;
@@ -656,37 +656,40 @@ public IImmutableSet<UniqueAddress> Receivers(UniqueAddress sender)
                 // The reason for not limiting it to strictly monitoredByNrOfMembers is that the leader must
                 // be able to continue its duties (e.g. removal of downed nodes) when many nodes are shutdown
                 // at the same time and nobody in the remaining cluster is monitoring some of the shutdown nodes.
-                (int, ImmutableSortedSet<UniqueAddress>) Take(int n, IEnumerator<UniqueAddress> iter, ImmutableSortedSet<UniqueAddress> acc)
+                (int, ImmutableSortedSet<UniqueAddress>) Take(int n, IEnumerator<UniqueAddress> iter, ImmutableSortedSet<UniqueAddress> acc, ImmutableHashSet<UniqueAddress> unreachable, int monitoredByNumberOfNodes)
                 {
-                    if (iter.MoveNext() == false || n == 0)
+                    while (true)
                     {
-                        iter.Dispose(); // dispose enumerator
-                        return (n, acc);
-                    }
-                    else
-                    {
-                        var next = iter.Current;
-                        var isUnreachable = Unreachable.Contains(next);
-                        if (isUnreachable && acc.Count >= MonitoredByNumberOfNodes)
-                        {
-                            return Take(n, iter, acc); // skip the unreachable, since we have already picked `MonitoredByNumberOfNodes`
-                        }
-                        else if (isUnreachable)
+                        if (iter.MoveNext() == false || n == 0)
                         {
-                            return Take(n, iter, acc.Add(next)); // include the unreachable, but don't count it
+                            iter.Dispose(); // dispose enumerator
+                            return (n, acc);
                         }
                         else
                         {
-                            return Take(n - 1, iter, acc.Add(next)); // include the reachable
+                            var next = iter.Current;
+                            var isUnreachable = unreachable.Contains(next);
+                            if (isUnreachable && acc.Count >= monitoredByNumberOfNodes)
+                            {
+                            }
+                            else if (isUnreachable)
+                            {
+                                acc = acc.Add(next);
+                            }
+                            else
+                            {
+                                n = n - 1;
+                                acc = acc.Add(next);
+                            }
                         }
                     }
                 }
 
-                var (remaining, slice1) = Take(MonitoredByNumberOfNodes, NodeRing.From(sender).Skip(1).GetEnumerator(), ImmutableSortedSet<UniqueAddress>.Empty);
+                var (remaining, slice1) = Take(MonitoredByNumberOfNodes, NodeRing.From(sender).Skip(1).GetEnumerator(), ImmutableSortedSet<UniqueAddress>.Empty, Unreachable, MonitoredByNumberOfNodes);
 
                 IImmutableSet<UniqueAddress> slice = remaining == 0 
                     ? slice1 // or, wrap-around
-                    : Take(remaining, NodeRing.TakeWhile(x => x != sender).GetEnumerator(), slice1).Item2;
+                    : Take(remaining, NodeRing.TakeWhile(x => x != sender).GetEnumerator(), slice1, Unreachable, MonitoredByNumberOfNodes).Item2;
 
                 return slice;
             }