Fix PeerPool to use getFuture() only once when adding peer.

Add isAlive(peer) procedure.
Add join(peer) procedure.
Remove useless metric.
Add notifyAndWait() procedure which will help to avoid race while disconnecting.

Author: cheatfate

beacon_chain/eth2_network.nim

@@ -275,12 +275,6 @@ declarePublicCounter nbc_successful_discoveries,
 declarePublicCounter nbc_failed_discoveries,
   "Number of failed discoveries"
 
-const delayBuckets = [1.0, 5.0, 10.0, 20.0, 40.0, 60.0]
-
-declareHistogram nbc_resolve_time,
-  "Time(s) used while resolving peer information",
-   buckets = delayBuckets
-
 const
   snappy_implementation {.strdefine.} = "libp2p"
 
@@ -348,6 +342,9 @@ proc getFuture*(peer: Peer): Future[void] {.inline.} =
     peer.disconnectedFut = newFuture[void]("Peer.disconnectedFut")
   peer.disconnectedFut
 
+proc isAlive*(peer: Peer): bool =
+  peer.connectionState notin {Disconnecting, Disconnected}
+
 proc getScore*(a: Peer): int =
   ## Returns current score value for peer ``peer``.
   a.score
@@ -358,6 +355,41 @@ proc updateScore*(peer: Peer, score: int) {.inline.} =
   if peer.score > PeerScoreHighLimit:
     peer.score = PeerScoreHighLimit
 
+proc join*(peer: Peer): Future[void] =
+  var retFuture = newFuture[void]("peer.lifetime.join")
+  let peerFut = peer.getFuture()
+  let alreadyFinished = peerFut.finished()
+
+  proc continuation(udata: pointer) {.gcsafe.} =
+    if not(retFuture.finished()):
+      retFuture.complete()
+
+  proc cancellation(udata: pointer) {.gcsafe.} =
+    if not(alreadyFinished):
+      peerFut.removeCallback(continuation)
+
+  if alreadyFinished:
+    # All the `peer.disconnectedFut` callbacks are already scheduled in current
+    # `poll()` call, to avoid race we going to finish only in next `poll()`
+    # call.
+    callSoon(continuation, cast[pointer](retFuture))
+  else:
+    # `peer.disconnectedFut` is not yet finished, but we want to be scheduled
+    # after all callbacks.
+    peerFut.addCallback(continuation)
+
+  return retFuture
+
+proc notifyAndWait*(peer: Peer): Future[void] =
+  ## Notify all the waiters that peer life is finished and wait until all
+  ## callbacks will be processed.
+  let joinFut = peer.join()
+  let fut = peer.disconnectedFut
+  peer.connectionState = Disconnecting
+  fut.complete()
+  peer.disconnectedFut = nil
+  joinFut
+
 proc calcThroughput(dur: Duration, value: uint64): float =
   let secs = float(chronos.seconds(1).nanoseconds)
   if isZero(dur):
@@ -941,17 +973,16 @@ proc resolvePeer(peer: Peer) =
       discard peer.info.peerId.extractPublicKey(key)
       keys.PublicKey.fromRaw(key.skkey.getBytes()).get().toNodeId()
 
-  debug "Peer's ENR recovery task started", node_id = $nodeId
-
   # This is "fast-path" for peers which was dialed. In this case discovery
   # already has most recent ENR information about this peer.
   let gnode = peer.network.discovery.getNode(nodeId)
   if gnode.isSome():
     peer.enr = some(gnode.get().record)
     inc(nbc_successful_discoveries)
-    let delay = now(chronos.Moment) - startTime
-    nbc_resolve_time.observe(delay.toFloatSeconds())
-    debug "Peer's ENR recovered", delay = $delay
+    debug "Peer's ENR recovered"
+  else:
+    inc(nbc_failed_discoveries)
+    debug "Peer's ENR could not be recovered"
 
 proc handlePeer*(peer: Peer) {.async.} =
   let res = peer.network.peerPool.addPeerNoWait(peer, peer.direction)
@@ -980,8 +1011,7 @@ proc handlePeer*(peer: Peer) {.async.} =
     # Peer was added to PeerPool.
     peer.score = NewPeerScore
     peer.connectionState = Connected
-    # We spawn task which will obtain ENR for this peer.
-    resolvePeer(peer)
+    peer.resolvePeer()
     debug "Peer successfully connected", peer = peer,
                                          connections = peer.connections
 
@@ -1048,10 +1078,8 @@ proc onConnEvent(node: Eth2Node, peerId: PeerID, event: ConnEvent) {.async.} =
       # Whatever caused disconnection, avoid connection spamming
       node.addSeen(peerId, SeenTableTimeReconnect)
 
-      let fut = peer.disconnectedFut
-      if not(isNil(fut)):
-        fut.complete()
-        peer.disconnectedFut = nil
+      if not(isNil(peer.disconnectedFut)):
+        await peer.notifyAndWait()
       else:
         # TODO (cheatfate): This could be removed when bug will be fixed inside
         # `nim-libp2p`.

beacon_chain/peer_pool.nim

@@ -339,6 +339,7 @@ proc deletePeer*[A, B](pool: PeerPool[A, B], peer: A, force = false): bool =
 
 proc addPeerImpl[A, B](pool: PeerPool[A, B], peer: A, peerKey: B,
                        peerType: PeerType) =
+  mixin getFuture
   proc onPeerClosed(udata: pointer) {.gcsafe.} =
     discard pool.deletePeer(peer)
 
@@ -367,13 +368,13 @@ proc checkPeer*[A, B](pool: PeerPool[A, B], peer: A): PeerStatus {.inline.} =
   ## * Peer's lifetime future is not finished yet - (PeerStatus.DeadPeerError)
   ##
   ## If peer could be added to PeerPool procedure returns (PeerStatus.Success)
-  mixin getKey, getFuture
+  mixin getKey, isAlive
   if not(pool.checkPeerScore(peer)):
     PeerStatus.LowScoreError
   else:
     let peerKey = getKey(peer)
     if not(pool.registry.hasKey(peerKey)):
-      if not(peer.getFuture().finished):
+      if peer.isAlive():
         PeerStatus.Success
       else:
         PeerStatus.DeadPeerError
@@ -393,7 +394,7 @@ proc addPeerNoWait*[A, B](pool: PeerPool[A, B],
   ##     (PeerStatus.NoSpaceError)
   ##
   ## Procedure returns (PeerStatus.Success) on success.
-  mixin getKey, getFuture
+  mixin getKey
   let res = pool.checkPeer(peer)
   if res != PeerStatus.Success:
     res