ETCM-446: Use the average received response per second as priority for pruning.

Author: aakoshh

src/main/scala/io/iohk/ethereum/network/ConnectedPeers.scala

@@ -5,7 +5,6 @@ import java.net.InetSocketAddress
 import akka.actor.ActorRef
 import akka.util.ByteString
 import scala.concurrent.duration.FiniteDuration
-import scala.util.Random
 
 case class ConnectedPeers(
     private val incomingPendingPeers: Map[PeerId, Peer],
@@ -85,9 +84,10 @@ case class ConnectedPeers(
   }
 
   def prunePeers(
-      incoming: Boolean,
       minAge: FiniteDuration,
       numPeers: Int,
+      priority: PeerId => Double = _ => 0.0,
+      incoming: Boolean = true,
       currentTimeMillis: Long = System.currentTimeMillis
   ): (Seq[Peer], ConnectedPeers) = {
     val ageThreshold = currentTimeMillis - minAge.toMillis
@@ -97,7 +97,7 @@ case class ConnectedPeers(
     } else {
       val candidates = handshakedPeers.values.filter(canPrune(incoming, ageThreshold)).toSeq
 
-      val toPrune = Random.shuffle(candidates).take(numPeers)
+      val toPrune = candidates.sortBy(peer => priority(peer.id)).take(numPeers)
 
       val pruned = copy(
         pruningPeers = toPrune.foldLeft(pruningPeers) { case (acc, peer) =>

src/main/scala/io/iohk/ethereum/network/PeerManagerActor.scala

@@ -102,7 +102,8 @@ class PeerManagerActor(
     handleCommonMessages(connectedPeers) orElse
       handleBlacklistMessages orElse
       handleConnections(connectedPeers) orElse
-      handleNewNodesToConnectMessages(connectedPeers)
+      handleNewNodesToConnectMessages(connectedPeers) orElse
+      handlePruning(connectedPeers)
   }
 
   private def handleNewNodesToConnectMessages(connectedPeers: ConnectedPeers): Receive = {
@@ -270,9 +271,9 @@ class PeerManagerActor(
         handshakedPeer.ref ! PeerActor.DisconnectPeer(Disconnect.Reasons.TooManyPeers)
 
         // It looks like all incoming slots are taken; try to make some room.
-        val prunedConnectedPeers = pruneIncomingPeers(connectedPeers)
+        schedulePruningIncomingPeers()
 
-        context become listening(prunedConnectedPeers)
+        context become listening(connectedPeers)
 
       } else if (handshakedPeer.nodeId.exists(connectedPeers.hasHandshakedWith)) {
         // FIXME: peers received after handshake should always have their nodeId defined, we could maybe later distinguish
@@ -301,12 +302,39 @@ class PeerManagerActor(
     (pendingPeer, newConnectedPeers)
   }
 
+  /** Ask for statistics and try to prune incoming peers when they arrive. */
+  private def schedulePruningIncomingPeers(): Unit = {
+    implicit val timeout: Timeout = Timeout(peerConfiguration.updateNodesInterval)
+    // Picking the minimum pruning age is fair for anyone
+    val window = peerConfiguration.minPruneAge
+    (peerStatistics ? PeerStatisticsActor.GetStatsForAll(window))
+      .mapTo[PeerStatisticsActor.StatsForAll]
+      .map(PruneIncomingPeers(_))
+      .pipeTo(self)
+  }
+
+  private def handlePruning(connectedPeers: ConnectedPeers): Receive = {
+    case PruneIncomingPeers(PeerStatisticsActor.StatsForAll(stats)) =>
+      val prunedConnectedPeers = pruneIncomingPeers(connectedPeers, stats)
+
+      context become listening(prunedConnectedPeers)
+  }
+
   /** Disconnect some incoming connections so we can free up slots. */
-  private def pruneIncomingPeers(connectedPeers: ConnectedPeers): ConnectedPeers = {
+  private def pruneIncomingPeers(
+      connectedPeers: ConnectedPeers,
+      stats: Map[PeerId, PeerStatisticsActor.Stat]
+  ): ConnectedPeers = {
     val pruneCount = PeerManagerActor.numberOfIncomingConnectionsToPrune(connectedPeers, peerConfiguration)
-
+    val now = System.currentTimeMillis
     val (peersToPrune, prunedConnectedPeers) =
-      connectedPeers.prunePeers(incoming = true, minAge = peerConfiguration.minPruneAge, numPeers = pruneCount)
+      connectedPeers.prunePeers(
+        incoming = true,
+        minAge = peerConfiguration.minPruneAge,
+        numPeers = pruneCount,
+        priority = prunePriority(stats, now),
+        currentTimeMillis = now
+      )
 
     peersToPrune.foreach { peer =>
       peer.ref ! PeerActor.DisconnectPeer(Disconnect.Reasons.TooManyPeers)
@@ -477,6 +505,8 @@ object PeerManagerActor {
 
   case class PeerAddress(value: String) extends BlackListId
 
+  case class PruneIncomingPeers(stats: PeerStatisticsActor.StatsForAll)
+
   /** Number of new connections the node should try to open at any given time. */
   def outgoingConnectionDemand(
       connectedPeers: ConnectedPeers,
@@ -500,4 +530,23 @@ object PeerManagerActor {
       connectedPeers.incomingHandshakedPeersCount - connectedPeers.incomingPruningPeersCount - minIncomingPeers
     )
   }
+
+  /** Assign a priority to peers that we can use to order connections,
+    * with lower priorities being the ones to prune first.
+    */
+  def prunePriority(stats: Map[PeerId, PeerStatisticsActor.Stat], currentTimeMillis: Long)(peerId: PeerId): Double = {
+    stats
+      .get(peerId)
+      .flatMap { stat =>
+        val maybeAgeSeconds = stat.firstSeenTimeMillis
+          .map(currentTimeMillis - _)
+          .map(_ * 1000)
+          .filterNot(_ <= 0)
+
+        // Use the average number of responses per second over the lifetime of the connection
+        // as an indicator of how fruitful the peer is for us.
+        maybeAgeSeconds.map(age => stat.responsesReceived.toDouble / age)
+      }
+      .getOrElse(0.0)
+  }
 }

src/test/scala/io/iohk/ethereum/network/PeerManagerSpec.scala

@@ -25,7 +25,7 @@ import org.scalatest.concurrent.Eventually
 import org.scalatest.flatspec.AnyFlatSpecLike
 import org.scalatest.matchers.should.Matchers
 import org.scalatestplus.scalacheck.ScalaCheckDrivenPropertyChecks
-import org.scalacheck.{Arbitrary, Gen}, Arbitrary.arbitrary
+import org.scalacheck.{Arbitrary, Gen, Shrink}, Arbitrary.arbitrary
 import scala.concurrent.duration._
 
 // scalastyle:off magic.number
@@ -260,6 +260,8 @@ class PeerManagerSpec
     peerAsOutgoingProbe.expectMsg(PeerActor.DisconnectPeer(Disconnect.Reasons.AlreadyConnected))
   }
 
+  behavior of "outgoingConnectionDemand"
+
   it should "try to connect to at least min-outgoing-peers but no longer than max-outgoing-peers" in new ConnectedPeersFixture {
     forAll { (connectedPeers: ConnectedPeers) =>
       val demand = PeerManagerActor.outgoingConnectionDemand(connectedPeers, peerConfiguration)
@@ -272,6 +274,8 @@ class PeerManagerSpec
     }
   }
 
+  behavior of "numberOfIncomingConnectionsToPrune"
+
   it should "try to prune incoming connections down to the minimum allowed number" in new ConnectedPeersFixture {
     forAll { (connectedPeers: ConnectedPeers) =>
       val numPeersToPrune = PeerManagerActor.numberOfIncomingConnectionsToPrune(connectedPeers, peerConfiguration)
@@ -289,20 +293,23 @@ class PeerManagerSpec
     }
   }
 
+  behavior of "ConnectedPeers.prunePeers"
+
   // The `ConnectedPeers` is quite slow to generate, so doing a few tests in one go.
-  it should "prune peers which are old enough down to incoming number, protecting against repeated forced pruning" in new ConnectedPeersFixture {
+  it should "prune peers which are old enough, protecting against repeated forced pruning" in new ConnectedPeersFixture {
     forAll { (connectedPeers: ConnectedPeers) =>
       val numPeersToPrune = PeerManagerActor.numberOfIncomingConnectionsToPrune(connectedPeers, peerConfiguration)
 
+      val now = System.currentTimeMillis
+
       // Prune the requested number of peers.
       {
         // Pretend we are in the future so age doesn't count.
         val (maxPrunedPeers, _) =
           connectedPeers.prunePeers(
-            incoming = true,
             peerConfiguration.minPruneAge,
             numPeers = numPeersToPrune,
-            currentTimeMillis = System.currentTimeMillis + peerConfiguration.minPruneAge.toMillis + 1
+            currentTimeMillis = now + peerConfiguration.minPruneAge.toMillis + 1
           )
 
         maxPrunedPeers.size shouldBe numPeersToPrune
@@ -311,61 +318,66 @@ class PeerManagerSpec
       // Only prune peers which are old enough.
       {
         val (agedPrunedPeers, _) = connectedPeers.prunePeers(
-          incoming = true,
           peerConfiguration.minPruneAge,
           numPeers = numPeersToPrune
         )
         Inspectors.forAll(agedPrunedPeers) {
-          _.createTimeMillis shouldBe <=(System.currentTimeMillis - peerConfiguration.minPruneAge.toMillis)
+          _.createTimeMillis shouldBe <=(now - peerConfiguration.minPruneAge.toMillis)
         }
       }
 
-      // Not prune repeatedly.
+      // Not prune twice in a row within the prune cool-of time.
       {
-        val minAge = 1.day // That should include all peers in the test data
-
-        val (probe2, _) = connectedPeers.prunePeers(
-          incoming = true,
+        val now = System.currentTimeMillis
+        val minAge = 1.minute
+        // Check that we have at least 2 peers to prune.
+        val (probe, _) = connectedPeers.prunePeers(
           minAge,
-          numPeers = 2
+          numPeers = Int.MaxValue,
+          currentTimeMillis = now
         )
-        if (probe2.size == 2) {
-          val (_, pruned1) = connectedPeers.prunePeers(
-            incoming = true,
-            minAge,
-            numPeers = 1
-          )
-          val (probe0, _) = pruned1.prunePeers(
-            incoming = true,
-            minAge,
-            numPeers = 1
-          )
-          probe0 shouldBe empty
-
-          val (probe1, _) = pruned1.prunePeers(
-            incoming = true,
-            minAge,
-            numPeers = 1,
-            currentTimeMillis = System.currentTimeMillis + minAge.toMillis
-          )
-          probe1 should not be empty
+        whenever(probe.size >= 2) {
+          val (_, pruned1) = connectedPeers.prunePeers(minAge, numPeers = 1, currentTimeMillis = now)
+
+          pruned1.prunePeers(minAge, numPeers = 1, currentTimeMillis = now + 1)._1 shouldBe empty
+
+          pruned1
+            .prunePeers(
+              minAge,
+              numPeers = 1,
+              currentTimeMillis = now + minAge.toMillis
+            )
+            ._1 should not be empty
         }
       }
 
       // Not prune the same peer repeatedly.
       {
         val (peers1, pruned) = connectedPeers.prunePeers(
-          incoming = true,
           peerConfiguration.minPruneAge,
           numPeers = numPeersToPrune
         )
         val (peers2, _) = pruned.prunePeers(
-          incoming = true,
           peerConfiguration.minPruneAge,
-          numPeers = numPeersToPrune
+          numPeers = numPeersToPrune,
+          currentTimeMillis = now + peerConfiguration.minPruneAge.toMillis
         )
         peers1.toSet intersect peers2.toSet shouldBe empty
       }
+
+      // Prune peers with minimum priority first.
+      {
+        val (peers, _) = connectedPeers.prunePeers(
+          peerConfiguration.minPruneAge,
+          numPeers = numPeersToPrune,
+          priority = _.hashCode.toDouble // Dummy priority
+        )
+        whenever(peers.nonEmpty) {
+          Inspectors.forAll(peers.init zip peers.tail) { case (a, b) =>
+            a.id.hashCode shouldBe <=(b.id.hashCode)
+          }
+        }
+      }
     }
   }
 
@@ -385,7 +397,6 @@ class PeerManagerSpec
 
       // Not prune again until the peers have been disconnected.
       val (peers, pruning) = connectedPeers.prunePeers(
-        incoming = true,
         peerConfiguration.minPruneAge,
         numPeersToPrune0
       )
@@ -395,22 +406,24 @@ class PeerManagerSpec
       val pruned = peers.foldLeft(pruning) { case (ps, p) =>
         ps.removeTerminatedPeer(p.ref)._2
       }
+      // ignore should be at the minimum incoming peer count now.
       PeerManagerActor.numberOfIncomingConnectionsToPrune(pruned, peerConfiguration) shouldBe 0
 
       val replenished = newIncoming.foldLeft(pruned) { case (ps, p) =>
         ps.addNewPendingPeer(p).promotePeerToHandshaked(p)
       }
+      // ignore should be maxed out now, can prune again.
       PeerManagerActor.numberOfIncomingConnectionsToPrune(replenished, peerConfiguration) shouldBe >(0)
     }
   }
 
   trait ConnectedPeersFixture {
     case class TestConfig(
-        minOutgoingPeers: Int = 20,
+        minOutgoingPeers: Int = 10,
         maxOutgoingPeers: Int = 30,
         maxIncomingPeers: Int = 30,
         maxPendingPeers: Int = 20,
-        pruneIncomingPeers: Int = 10,
+        pruneIncomingPeers: Int = 20,
         minPruneAge: FiniteDuration = 30.minutes
     ) extends PeerManagerActor.PeerConfiguration.ConnectionLimits
 
@@ -419,6 +432,9 @@ class PeerManagerSpec
     implicit val arbConnectedPeers: Arbitrary[ConnectedPeers] = Arbitrary {
       genConnectedPeers(peerConfiguration.maxIncomingPeers, peerConfiguration.maxOutgoingPeers)
     }
+
+    implicit val noShrinkConnectedPeers: Shrink[ConnectedPeers] =
+      Shrink[ConnectedPeers](_ => Stream.empty)
   }
 
   trait TestSetup {
@@ -508,7 +524,7 @@ class PeerManagerSpec
       ip <- Gen.listOfN(4, Gen.choose(0, 255)).map(_.mkString("."))
       port <- Gen.choose(10000, 60000)
       incoming <- arbitrary[Boolean]
-      ageMillis <- Gen.choose(0, 60 * 60 * 1000)
+      ageMillis <- Gen.choose(0, 24 * 60 * 60 * 1000)
     } yield Peer(
       remoteAddress = new InetSocketAddress(ip, port),
       ref = TestProbe().ref,
@@ -531,7 +547,9 @@ class PeerManagerSpec
       incoming <- Gen.listOfN(numIncoming, genIncomingPeer)
       outgoing <- Gen.listOfN(numOutgoing, genOugoingPeer)
       connections0 = (incoming ++ outgoing).foldLeft(ConnectedPeers.empty)(_ addNewPendingPeer _)
-      handshaked <- Gen.someOf(incoming ++ outgoing)
+      ratioHandshaked <- Gen.choose(0.75, 1.0)
+      numHandshaked <- Gen.choose(0.75, 1.0).map(_ * (numIncoming + numOutgoing)).map(_.toInt)
+      handshaked <- Gen.pick(numHandshaked, incoming ++ outgoing)
       connections1 = handshaked.foldLeft(connections0)(_ promotePeerToHandshaked _)
     } yield connections1