Integration of RecursivePeerRefreshState and PeerDiscoveryController

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/discovery/internal/PeerDiscoveryController.java

@@ -12,7 +12,6 @@
  */
 package tech.pegasys.pantheon.ethereum.p2p.discovery.internal;
 
-import static java.util.Collections.emptyList;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static tech.pegasys.pantheon.ethereum.p2p.discovery.internal.PeerTable.AddResult.Outcome;
@@ -39,6 +38,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.logging.log4j.LogManager;
@@ -92,6 +92,7 @@ public class PeerDiscoveryController {
 
   private static final Logger LOG = LogManager.getLogger();
   private static final long REFRESH_CHECK_INTERVAL_MILLIS = MILLISECONDS.convert(30, SECONDS);
+  private static final int PEER_REFRESH_ROUND_TIMEOUT_IN_SECONDS = 5;
   protected final TimerUtil timerUtil;
   private final PeerTable peerTable;
 
@@ -123,6 +124,8 @@ public class PeerDiscoveryController {
   // Observers for "peer bonded" discovery events.
   private final Subscribers<Consumer<PeerBondedEvent>> peerBondedObservers;
 
+  private RecursivePeerRefreshState recursivePeerRefreshState;
+
   public PeerDiscoveryController(
       final KeyPair keypair,
       final DiscoveryPeer localPeer,
@@ -154,14 +157,29 @@ public CompletableFuture<?> start() {
     }
 
     bootstrapNodes.stream()
-        .filter(node -> peerTable.tryAdd(node).getOutcome() == Outcome.ADDED)
-        .filter(node -> whitelistIfPresentIsNodePermitted(node))
-        .forEach(node -> bond(node, true));
+        .filter(this::whitelistIfPresentIsNodePermitted)
+        .forEach(peerTable::tryAdd);
+
+    recursivePeerRefreshState =
+        new RecursivePeerRefreshState(
+            peerBlacklist,
+            nodeWhitelistController,
+            this::bond,
+            this::findNodes,
+            timerUtil,
+            PEER_REFRESH_ROUND_TIMEOUT_IN_SECONDS,
+            100);
+
+    final List<DiscoveryPeer> initialDiscoveryPeers =
+        bootstrapNodes.stream()
+            .filter(this::whitelistIfPresentIsNodePermitted)
+            .collect(Collectors.toList());
+    recursivePeerRefreshState.start(initialDiscoveryPeers, localPeer.getId());
 
     final long timerId =
         timerUtil.setPeriodic(
             Math.min(REFRESH_CHECK_INTERVAL_MILLIS, tableRefreshIntervalMs),
-            () -> refreshTableIfRequired());
+            this::refreshTableIfRequired);
     tableRefreshTimerId = OptionalLong.of(timerId);
 
     return CompletableFuture.completedFuture(null);
@@ -181,7 +199,7 @@ public CompletableFuture<?> stop() {
 
   private boolean whitelistIfPresentIsNodePermitted(final DiscoveryPeer sender) {
     return nodeWhitelistController
-        .map(nodeWhitelistController1 -> nodeWhitelistController1.isPermitted(sender))
+        .map(nodeWhitelistController -> nodeWhitelistController.isPermitted(sender))
         .orElse(true);
   }
 
@@ -239,38 +257,17 @@ public void onMessage(final Packet packet, final DiscoveryPeer sender) {
                       return;
                     }
                     addToPeerTable(peer);
-
-                    // If this was a bootstrap peer, let's ask it for nodes near to us.
-                    if (interaction.isBootstrap()) {
-                      findNodes(peer, localPeer.getId());
-                    }
+                    recursivePeerRefreshState.onBondingComplete(peer);
                   });
           break;
         }
       case NEIGHBORS:
         LOG.trace("Received NEIGHBORS packet from {}", sender.getEnodeURI());
         matchInteraction(packet)
             .ifPresent(
-                interaction -> {
-                  // Extract the peers from the incoming packet.
-                  final List<DiscoveryPeer> neighbors =
-                      packet
-                          .getPacketData(NeighborsPacketData.class)
-                          .map(NeighborsPacketData::getNodes)
-                          .orElse(emptyList());
-
-                  for (final DiscoveryPeer neighbor : neighbors) {
-                    // If the peer is not whitelisted, is blacklisted, is already known, or
-                    // represents this node, skip bonding
-                    if (!whitelistIfPresentIsNodePermitted(neighbor)
-                        || peerBlacklist.contains(neighbor)
-                        || peerTable.get(neighbor).isPresent()
-                        || neighbor.getId().equals(localPeer.getId())) {
-                      continue;
-                    }
-                    bond(neighbor, false);
-                  }
-                });
+                interaction ->
+                    recursivePeerRefreshState.onNeighboursPacketReceived(
+                        peer, packet.getPacketData(NeighborsPacketData.class).orElse(null)));
         break;
 
       case FIND_NEIGHBORS:
@@ -341,24 +338,29 @@ private void refreshTableIfRequired() {
     }
   }
 
+  @VisibleForTesting
+  RecursivePeerRefreshState getRecursivePeerRefreshState() {
+    return recursivePeerRefreshState;
+  }
+
   /**
    * Refreshes the peer table by generating a random ID and interrogating the closest nodes for it.
    * Currently the refresh process is NOT recursive.
    */
   private void refreshTable() {
     final BytesValue target = Peer.randomId();
-    peerTable.nearestPeers(Peer.randomId(), 16).forEach((peer) -> findNodes(peer, target));
+    final List<DiscoveryPeer> initialPeers = peerTable.nearestPeers(Peer.randomId(), 16);
+    recursivePeerRefreshState.start(initialPeers, target);
     lastRefreshTime = System.currentTimeMillis();
   }
 
   /**
    * Initiates a bonding PING-PONG cycle with a peer.
    *
    * @param peer The targeted peer.
-   * @param bootstrap Whether this is a bootstrap interaction.
    */
   @VisibleForTesting
-  void bond(final DiscoveryPeer peer, final boolean bootstrap) {
+  void bond(final DiscoveryPeer peer) {
     peer.setFirstDiscovered(System.currentTimeMillis());
     peer.setStatus(PeerDiscoveryStatus.BONDING);
 
@@ -383,7 +385,7 @@ void bond(final DiscoveryPeer peer, final boolean bootstrap) {
 
     // The filter condition will be updated as soon as the action is performed.
     final PeerInteractionState ping =
-        new PeerInteractionState(action, PacketType.PONG, (packet) -> false, true, bootstrap);
+        new PeerInteractionState(action, PacketType.PONG, (packet) -> false, true);
     dispatchInteraction(peer, ping);
   }
 
@@ -414,7 +416,7 @@ private void findNodes(final DiscoveryPeer peer, final BytesValue target) {
           sendPacket(peer, PacketType.FIND_NEIGHBORS, data);
         };
     final PeerInteractionState interaction =
-        new PeerInteractionState(action, PacketType.NEIGHBORS, packet -> true, true, false);
+        new PeerInteractionState(action, PacketType.NEIGHBORS, packet -> true, true);
     dispatchInteraction(peer, interaction);
   }
 
@@ -483,22 +485,18 @@ private class PeerInteractionState implements Predicate<Packet> {
     private Predicate<Packet> filter;
     /** Whether the action associated to this state is retryable or not. */
     private final boolean retryable;
-    /** Whether this is an entry for a bootstrap peer. */
-    private final boolean bootstrap;
     /** Timers associated with this entry. */
     private OptionalLong timerId = OptionalLong.empty();
 
     PeerInteractionState(
         final Consumer<PeerInteractionState> action,
         final PacketType expectedType,
         final Predicate<Packet> filter,
-        final boolean retryable,
-        final boolean bootstrap) {
+        final boolean retryable) {
       this.action = action;
       this.expectedType = expectedType;
       this.filter = filter;
       this.retryable = retryable;
-      this.bootstrap = bootstrap;
     }
 
     @Override
@@ -510,10 +508,6 @@ void updateFilter(final Predicate<Packet> filter) {
       this.filter = filter;
     }
 
-    boolean isBootstrap() {
-      return bootstrap;
-    }
-
     /**
      * Executes the action associated with this state. Sets a "boomerang" timer to itself in case
      * the action is retryable.

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/discovery/internal/PeerTable.java

@@ -19,7 +19,6 @@
 
 import tech.pegasys.pantheon.crypto.Hash;
 import tech.pegasys.pantheon.ethereum.p2p.discovery.DiscoveryPeer;
-import tech.pegasys.pantheon.ethereum.p2p.discovery.PeerDiscoveryStatus;
 import tech.pegasys.pantheon.ethereum.p2p.peers.Peer;
 import tech.pegasys.pantheon.ethereum.p2p.peers.PeerId;
 import tech.pegasys.pantheon.util.bytes.BytesValue;
@@ -180,7 +179,6 @@ private void buildBloomFilter() {
   public List<DiscoveryPeer> nearestPeers(final BytesValue target, final int limit) {
     final BytesValue keccak256 = Hash.keccak256(target);
     return getAllPeers().stream()
-        .filter(p -> p.getStatus() == PeerDiscoveryStatus.BONDED)
         .sorted(comparingInt((peer) -> distance(peer.keccak256(), keccak256)))
         .limit(limit)
         .collect(toList());