Extract BySpeedStrategy logic in PostMergeBlocksSyncPeerAllocationStr…

…ategy to BySpeedStrategy (NethermindEth#4546) (NethermindEth#4659)

Co-authored-by: Amirul Ashraf <asdacap@gmail.com>

src/Nethermind/Nethermind.Merge.Plugin/Synchronization/PostMergeBlocksSyncPeerAllocationStrategy.cs

@@ -17,6 +17,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using Nethermind.Blockchain;
 using Nethermind.Stats;
 using Nethermind.Synchronization.Peers;
@@ -31,6 +32,8 @@ public class PostMergeBlocksSyncPeerAllocationStrategy : IPeerAllocationStrategy
 
     private const decimal MinDiffPercentageForSpeedSwitch = 0.10m;
     private const int MinDiffForSpeedSwitch = 10;
+    private readonly BySpeedStrategy _innerStrategy =
+        new(TransferSpeedType.Bodies, true, MinDiffPercentageForSpeedSwitch, MinDiffForSpeedSwitch);
 
     public PostMergeBlocksSyncPeerAllocationStrategy(long? minBlocksAhead, IBeaconPivot beaconPivot)
     {
@@ -40,68 +43,27 @@ public PostMergeBlocksSyncPeerAllocationStrategy(long? minBlocksAhead, IBeaconPi
 
     public bool CanBeReplaced => true;
 
-    private long? GetSpeed(INodeStatsManager nodeStatsManager, PeerInfo peerInfo)
-    {
-        long? bodiesSpeed = nodeStatsManager.GetOrAdd(peerInfo.SyncPeer.Node)
-            .GetAverageTransferSpeed(TransferSpeedType.Bodies);
-        if (bodiesSpeed == null)
-        {
-            return null;
-        }
-
-        return bodiesSpeed ?? 0;
-    }
-
     public PeerInfo? Allocate(PeerInfo? currentPeer, IEnumerable<PeerInfo> peers, INodeStatsManager nodeStatsManager,
         IBlockTree blockTree)
     {
-        int nullSpeed = -1;
-        int peersCount = 0;
-
-        bool wasNull = currentPeer == null;
-
-        long currentSpeed = wasNull
-            ? nullSpeed
-            : GetSpeed(nodeStatsManager, currentPeer!) ?? nullSpeed;
-        (PeerInfo? Info, long TransferSpeed) fastestPeer = (currentPeer, currentSpeed);
-
-        foreach (PeerInfo info in peers)
+        IEnumerable<PeerInfo> filteredPeers = peers.Where((info) =>
         {
-            (this as IPeerAllocationStrategy).CheckAsyncState(info);
-            peersCount++;
-
             if (_beaconPivot.BeaconPivotExists())
             {
                 if (info.HeadNumber < _beaconPivot.PivotNumber - 1)
                 {
                     // we need to guarantee the peer can have all the block prior to beacon pivot
-                    continue;
+                    return false;
                 }
             }
             else if (info.HeadNumber < (blockTree.BestSuggestedBody?.Number ?? 0) + (_minBlocksAhead ?? 1))
             {
-                continue;
-            }
-
-            long averageTransferSpeed = GetSpeed(nodeStatsManager, info) ?? 0;
-            if (averageTransferSpeed > fastestPeer.TransferSpeed)
-            {
-                fastestPeer = (info, averageTransferSpeed);
+                return false;
             }
-        }
 
-        if (peersCount == 0)
-        {
-            return currentPeer;
-        }
-
-        decimal speedRatio = fastestPeer.TransferSpeed / (decimal)Math.Max(1L, currentSpeed);
-        if (speedRatio > 1m + MinDiffPercentageForSpeedSwitch
-            && fastestPeer.TransferSpeed - currentSpeed > MinDiffForSpeedSwitch)
-        {
-            return fastestPeer.Info;
-        }
+            return true;
+        });
 
-        return currentPeer ?? fastestPeer.Info;
+        return _innerStrategy.Allocate(currentPeer, filteredPeers, nodeStatsManager, blockTree);
     }
 }

src/Nethermind/Nethermind.Synchronization.Test/AllocationStrategies/BySpeedStrategyTests.cs

@@ -0,0 +1,79 @@
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Security.Cryptography.X509Certificates;
+using FluentAssertions;
+using Nethermind.Blockchain.Synchronization;
+using Nethermind.Core.Crypto;
+using Nethermind.Core.Extensions;
+using Nethermind.Core.Test.Builders;
+using Nethermind.Crypto;
+using Nethermind.Overseer.Test.JsonRpc.Dto;
+using Nethermind.Stats;
+using Nethermind.Stats.Model;
+using Nethermind.Synchronization.Peers;
+using Nethermind.Synchronization.Peers.AllocationStrategies;
+using NSubstitute;
+using NUnit.Framework;
+using PublicKey = Nethermind.Core.Crypto.PublicKey;
+
+namespace Nethermind.Synchronization.Test.AllocationStrategies;
+
+public class BySpeedStrategyTests
+{
+    private static PublicKey TestPublicKey = new(Bytes.FromHexString(
+        "0x13a1107b6f78a4977222d2d5a4cd05a8a042b75222c8ec99129b83793eda3d214208d4e835617512fc8d148d3d1b4d89530861644f531675b1fb64b785c6c152"));
+
+    [TestCase(1, 0, 0, 2)]
+    [TestCase(2, 0, 0, 2)]
+    [TestCase(3, 0, 0, 2)]
+    [TestCase(null, 0, 0, 2)]
+    [TestCase(1, 0.5, 0, 1)]
+    [TestCase(1, 0.0, 50, 1)]
+    [TestCase(1, 0.0, 10, 2)]
+    [TestCase(1, 0.1, 0, 2)]
+    public void TestShouldSelectHighestSpeed(int? currentPeerIdx, decimal minDiffPercentageForSpeedSwitch, int minDiffSpeed, int expectedSelectedPeerIdx)
+    {
+        long[] peerSpeeds = new long[]
+        {
+            100,
+            90,
+            120,
+            50
+        };
+
+        INodeStatsManager nodeStatsManager = Substitute.For<INodeStatsManager>();
+
+        List<PeerInfo> peers = new();
+        for (int i = 0; i < peerSpeeds.Length; i++)
+        {
+            PeerInfo pInfo = CreatePeerInfoWithSpeed(peerSpeeds[i], nodeStatsManager);
+            peers.Add(pInfo);
+        }
+
+        BySpeedStrategy strategy = new(TransferSpeedType.Bodies, true, minDiffPercentageForSpeedSwitch, minDiffSpeed);
+
+        PeerInfo? currentPeer = null;
+        if (currentPeerIdx != null) currentPeer = peers[currentPeerIdx.Value];
+
+        PeerInfo? selectedPeer = strategy.Allocate(currentPeer, peers, nodeStatsManager, Build.A.BlockTree().TestObject);
+
+        int selectedPeerIdx = peers.IndexOf(selectedPeer);
+        selectedPeerIdx.Should().Be(expectedSelectedPeerIdx);
+    }
+
+    private static PeerInfo CreatePeerInfoWithSpeed(long speed, INodeStatsManager nodeStatsManager)
+    {
+        ISyncPeer syncPeer = Substitute.For<ISyncPeer>();
+        Node node = new(TestPublicKey, IPEndPoint.Parse("127.0.0.1"));
+        syncPeer.Node.Returns(node);
+        syncPeer.IsInitialized.Returns(true);
+
+        PeerInfo pInfo = new(syncPeer);
+
+        INodeStats nodeStats = Substitute.For<INodeStats>();
+        nodeStats.GetAverageTransferSpeed(Arg.Any<TransferSpeedType>()).Returns(speed);
+        nodeStatsManager.GetOrAdd(Arg.Is<Node>(n => ReferenceEquals(n, node))).Returns(nodeStats);
+        return pInfo;
+    }
+}

src/Nethermind/Nethermind.Synchronization/Peers/AllocationStrategies/BySpeedStrategy.cs

@@ -1,19 +1,20 @@
 //  Copyright (c) 2021 Demerzel Solutions Limited
 //  This file is part of the Nethermind library.
-// 
+//
 //  The Nethermind library is free software: you can redistribute it and/or modify
 //  it under the terms of the GNU Lesser General Public License as published by
 //  the Free Software Foundation, either version 3 of the License, or
 //  (at your option) any later version.
-// 
+//
 //  The Nethermind library is distributed in the hope that it will be useful,
 //  but WITHOUT ANY WARRANTY; without even the implied warranty of
 //  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 //  GNU Lesser General Public License for more details.
-// 
+//
 //  You should have received a copy of the GNU Lesser General Public License
 //  along with the Nethermind. If not, see <http://www.gnu.org/licenses/>.
 
+using System;
 using System.Collections.Generic;
 using Nethermind.Blockchain;
 using Nethermind.Stats;
@@ -24,23 +25,34 @@ public class BySpeedStrategy : IPeerAllocationStrategy
     {
         private readonly TransferSpeedType _speedType;
         private readonly bool _priority;
+        private readonly decimal _minDiffPercentageForSpeedSwitch;
+        private readonly int _minDiffForSpeedSwitch;
 
-        public BySpeedStrategy(TransferSpeedType speedType, bool priority)
+        public BySpeedStrategy(
+            TransferSpeedType speedType,
+            bool priority,
+            decimal minDiffPercentageForSpeedSwitch = 0.0m,
+            int minDiffForSpeedSwitch = 0
+        )
         {
             _speedType = speedType;
             _priority = priority;
+            _minDiffPercentageForSpeedSwitch = minDiffPercentageForSpeedSwitch;
+            _minDiffForSpeedSwitch = minDiffForSpeedSwitch;
         }
 
         public bool CanBeReplaced => false;
 
         public PeerInfo? Allocate(PeerInfo? currentPeer, IEnumerable<PeerInfo> peers, INodeStatsManager nodeStatsManager, IBlockTree blockTree)
         {
             long nullSpeed = _priority ? -1 : long.MaxValue;
+            long peerCount = 0;
             long currentSpeed = currentPeer == null ? nullSpeed : nodeStatsManager.GetOrAdd(currentPeer.SyncPeer.Node).GetAverageTransferSpeed(_speedType) ?? nullSpeed;
             (PeerInfo? Info, long TransferSpeed) bestPeer = (currentPeer, currentSpeed);
 
             foreach (PeerInfo info in peers)
             {
+                peerCount++;
                 (this as IPeerAllocationStrategy).CheckAsyncState(info);
 
                 long averageTransferSpeed = nodeStatsManager.GetOrAdd(info.SyncPeer.Node).GetAverageTransferSpeed(_speedType) ?? 0;
@@ -50,7 +62,19 @@ public BySpeedStrategy(TransferSpeedType speedType, bool priority)
                 }
             }
 
-            return bestPeer.Info;
+            if (peerCount == 0)
+            {
+                return currentPeer;
+            }
+
+            decimal speedRatio = bestPeer.TransferSpeed / (decimal)Math.Max(1L, currentSpeed);
+            if (speedRatio > 1m + _minDiffPercentageForSpeedSwitch
+                && bestPeer.TransferSpeed - currentSpeed > _minDiffForSpeedSwitch)
+            {
+                return bestPeer.Info;
+            }
+
+            return currentPeer ?? bestPeer.Info;
         }
     }
 }