Beacon sync update peer connection mgmnt (#3776)

* Update error register names

why
  More intuitive names for assigned error types

* Cosmetics, update logging

* Update `worker` module (with basic tasks managed by scheduler)

why
  Explicitly check for block headers or blocks to unstage queued block
  headers or blocks. Not explicitly unstaging might lead to unnecessary
  (maybe small) delays when ending the header or block sync phase.

* Classify syncer peers and select `faster` ones for downloading

why
  When there are more than enough peers available to fetch data
  simultaneously, only those are selected that are unknown or
  have shown higher throughput when fetching.

* Relax fetch response timeouts and error threshold

why
  There is no need to zombify sync peers unless they deliver ostensibly
  bogus data.

  With zero response, only throughput statistics will be affected if a
  peer responses in time. Low throughput will leave a peer available
  but ignored if there are other peers with higher throughput unless
  all peers can run parallel.

* Fix fringe case when all higher throughput measurements are equal

why
  When most higher throughputs are equal then case all ranks must
  be the topmost rank (an not the least.) Otherwise one might have
  delays until the situation is resolved.

* Degrade ranks of useless peers as well as increasing their error count

why
  Previously, syncer peers that did not deliver and ran in a timeout
  were not marked degraded (for ranking) but rather the error count
  increased.

  As a consequence, these peers did not lose with ranking. This led
  these peers linger on and tried again until the error count threshold
  was reached.

---------

Co-authored-by: jordan <jordan@curd>

Author: mjfh

execution_chain/sync/beacon/worker.nim

@@ -18,19 +18,7 @@ import
   ../../networking/p2p,
   ./worker/headers/headers_target,
   ./worker/update/metrics,
-  ./worker/[blocks, headers, start_stop, update, worker_desc]
-
-# ------------------------------------------------------------------------------
-# Private functions
-# ------------------------------------------------------------------------------
-
-proc somethingToCollect(buddy: BeaconBuddyRef): bool =
-  if buddy.ctx.hibernate:                        # not activated yet?
-    return false
-  if buddy.headersCollectOk() or                 # something on TODO list
-     buddy.blocksCollectOk():
-    return true
-  false
+  ./worker/[blocks, classify, headers, start_stop, update, worker_desc]
 
 # ------------------------------------------------------------------------------
 # Public start/stop and admin functions
@@ -68,7 +56,7 @@ proc stop*(buddy: BeaconBuddyRef; info: static[string]) =
   ## Clean up this peer
   if not buddy.ctx.hibernate: debug info & ": release peer", peer=buddy.peer,
     throughput=buddy.only.thruPutStats.toMeanVar.psStr,
-    nSyncPeers=(buddy.ctx.pool.nBuddies-1), syncState=($buddy.syncState)
+    nSyncPeers=(buddy.ctx.pool.nBuddies-1), state=($buddy.syncState)
   buddy.stopBuddy()
 
 # ------------------------------------------------------------------------------
@@ -151,53 +139,67 @@ template runPeer*(buddy: BeaconBuddyRef; info: static[string]): Duration =
   ##
   var bodyRc = chronos.nanoseconds(0)
   block body:
-    if buddy.somethingToCollect():
-
-      # Download and process headers and blocks
-      while buddy.headersCollectOk():
+    if buddy.somethingToCollectOrUnstage():
 
-        # Collect headers and either stash them on the header chain cache
-        # directly, or stage on the header queue to get them serialised and
-        # stashed, later.
-        buddy.headersCollect info # async/template
+      # Classify sync peer (aka buddy) performance
+      let (fetchPerf {.inject.}, rank) = buddy.classifyForFetching()
 
-        # Store serialised headers from the `staged` queue onto the header
-        # chain cache.
-        if not buddy.headersUnstage info:
-          # Need to proceed with another peer (e.g. gap between queue and
-          # header chain cache.)
-          bodyRc = workerIdleWaitInterval
-          break body
+      trace info & ": start processing", peer=buddy.peer,
+        throughput=buddy.only.thruPutStats.toMeanVar.psStr,
+        fetchPerf, rank=(if rank < 0: "n/a" else: $rank),
+        nSyncPeers=buddy.ctx.pool.nBuddies, state=($buddy.syncState)
 
-        # End `while()`
+      if fetchPerf == rankingTooLow:
+        bodyRc = workerIdleWaitInterval
+        break body                                # done, exit
 
-      # Fetch bodies and combine them with headers to blocks to be staged.
-      # These staged blocks are then excuted by the daemon process (no `peer`
-      # needed.)
-      while buddy.blocksCollectOk():
-
-        # Collect bodies and either import them via `FC` module, or stage on
-        # the blocks queue to get them serialised and imported, later.
-        buddy.blocksCollect info # async/template
-
-        # Import bodies from the `staged` queue.
-        if not buddy.blocksUnstage info: # async/template
-          # Need to proceed with another peer (e.g. gap between top imported
-          # block and blocks queue.)
-          bodyRc = workerIdleWaitInterval
-          break body
-
-        # End `while()`
+      # Download and process headers and blocks
+      block downloadAndProcess:
+        while buddy.headersCollectOk():
+
+          # Collect headers and either stash them on the header chain cache
+          # directly, or stage on the header queue to get them serialised and
+          # stashed, later.
+          buddy.headersCollect info               # async/template
+
+          # Store serialised headers from the `staged` queue onto the header
+          # chain cache.
+          if not buddy.headersUnstage info:       # async/template
+            # Need to proceed with another peer (e.g. gap between queue and
+            # header chain cache.)
+            bodyRc = workerIdleWaitInterval
+            break downloadAndProcess
+
+          # End `while()`
+
+        # Fetch bodies and combine them with headers to blocks to be staged.
+        # These staged blocks are then excuted by the daemon process (no `peer`
+        # needed.)
+        while buddy.blocksCollectOk():
+          # Collect bodies and either import them via `FC` module, or stage on
+          # the blocks queue to get them serialised and imported, later.
+          buddy.blocksCollect info                # async/template
+
+          # Import bodies from the `staged` queue.
+          if not buddy.blocksUnstage info:        # async/template
+            # Need to proceed with another peer (e.g. gap between top imported
+            # block and blocks queue.)
+            bodyRc = workerIdleWaitInterval
+            break downloadAndProcess
+
+          # End `while()`
+
+      # End block: `actionLoop`
 
     else:
-      # Potential manual target set up
+      # Potentially a manual sync target set up
       buddy.headersTargetActivate info
 
-    # End block: `body`
+    # Idle sleep unless there is something to do
+    if not buddy.somethingToCollectOrUnstage():
+      bodyRc = workerIdleWaitInterval
 
-  # Idle sleep unless there is something to do
-  if not buddy.somethingToCollect():
-    bodyRc = workerIdleWaitInterval
+    # End block: `body`
 
   bodyRc
 

execution_chain/sync/beacon/worker/blocks.nim

@@ -191,7 +191,7 @@ template blocksCollect*(
 
       debug info & ": no blocks yet (failed peer)", peer,
         failedPeers=ctx.pool.failedPeers.len,
-        syncState=($buddy.syncState), bdyErrors=buddy.bdyErrors
+        state=($buddy.syncState), nErrors=buddy.blkErrors()
       break body                                    # return
 
     # This message might run in addition to the `chronicles.info` part
@@ -205,9 +205,10 @@ template blocksCollect*(
 
 # --------------
 
-proc blocksUnstageOk*(ctx: BeaconCtxRef): bool =
+proc blocksUnstageOk*(buddy: BeaconBuddyRef): bool =
   ## Check whether import processing is possible
   ##
+  let ctx = buddy.ctx
   not ctx.poolMode and
   0 < ctx.blk.staged.len
 

execution_chain/sync/beacon/worker/blocks/blocks_blocks.nim

@@ -51,7 +51,7 @@ template blocksFetchCheckImpl(
       iv {.inject,used.} = iv
       peer {.inject,used.} = buddy.peer
 
-    # Preset/append headers to be completed with bodies. Also collect block
+    # Preset headers to be completed with bodies. Also collect block
     # hashes for fetching missing blocks.
     var
       request = BlockBodiesRequest(blockHashes: newSeqUninit[Hash32](iv.len))
@@ -98,7 +98,8 @@ template blocksFetchCheckImpl(
           blocks.setLen(n)                                 # curb off junk
           buddy.bdyFetchRegisterError()
           trace info & ": Cut off junk blocks", peer, iv, n=n,
-            nTxs=bodies[n].transactions.len, nBodies, bdyErrors=buddy.bdyErrors
+            nTxs=bodies[n].transactions.len, nBodies,
+            nErrors=buddy.nErrors.fetch.bdy
           break loop
 
         # In order to avoid extensive checking here and also within the `FC`
@@ -117,7 +118,7 @@ template blocksFetchCheckImpl(
     if 0 < blocks.len.uint64:
       bodyRc = Opt[seq[EthBlock]].ok(blocks)               # return ok()
 
-    buddy.only.nProcErrors.blk.inc
+    buddy.nErrors.apply.blk.inc
     break body                                             # return err()
 
   bodyRc # return
@@ -141,7 +142,7 @@ template blocksFetch*(
   block body:
     # Make sure that this sync peer is not banned from block processing,
     # already.
-    if nProcBlocksErrThreshold < buddy.only.nProcErrors.blk:
+    if nProcBlocksErrThreshold < buddy.nErrors.apply.blk:
       buddy.ctrl.zombie = true
       break body                                      # return err()
 
@@ -195,7 +196,7 @@ template blocksImport*(
 
     var isError = false
     block loop:
-      trace info & ": Start importing blocks", peer, iv,
+      trace info & ": start importing blocks", peer, iv,
         nBlocks=iv.len, base=ctx.chain.baseNumber.bnStr,
         head=ctx.chain.latestNumber.bnStr
 
@@ -208,7 +209,7 @@ template blocksImport*(
             # Mark peer that produced that unusable headers list as a zombie
             let srcPeer = buddy.getPeer peerID
             if not srcPeer.isNil:
-              srcPeer.only.nProcErrors.blk = nProcBlocksErrThreshold + 1
+              srcPeer.only.nErrors.apply.blk = nProcBlocksErrThreshold + 1
 
             # Check whether it is enough to skip the current blocks list, only
             if ctx.subState.procFailNum != nBn:
@@ -248,7 +249,7 @@ template blocksImport*(
     if not isError:
       let srcPeer = buddy.getPeer peerID
       if not srcPeer.isNil:
-        srcPeer.only.nProcErrors.blk = 0
+        srcPeer.only.nErrors.apply.blk = 0
 
     nBlocks = ctx.subState.top - iv.minPt + 1      # number of blocks imported
 

execution_chain/sync/beacon/worker/blocks/blocks_fetch.nim

@@ -69,7 +69,7 @@ template fetchBodies*(
       nReq {.inject,used.} = request.blockHashes.len
 
     trace trEthSendSendingGetBlockBodies,
-      peer, nReq, bdyErrors=buddy.bdyErrors
+      peer, nReq, nErrors=buddy.nErrors.fetch.bdy
 
     let rc = await buddy.getBlockBodies(request)
     var elapsed: Duration
@@ -82,47 +82,53 @@ template fetchBodies*(
         of ENoException:
           break evalError
         of EPeerDisconnected, ECancelledError:
-          buddy.only.nRespErrors.blk.inc
+          buddy.nErrors.fetch.bdy.inc
           buddy.ctrl.zombie = true
         of ECatchableError:
           buddy.bdyFetchRegisterError()
 
         chronicles.info trEthRecvReceivedBlockBodies & " error", peer, nReq,
-          elapsed=rc.error.elapsed.toStr, syncState=($buddy.syncState),
-          error=rc.error.name, msg=rc.error.msg, bdyErrors=buddy.bdyErrors
+          elapsed=rc.error.elapsed.toStr, state=($buddy.syncState),
+          error=rc.error.name, msg=rc.error.msg, nErrors=buddy.nErrors.fetch.bdy
         break body                                  # return err()
 
     # Evaluate result
     if rc.isErr or buddy.ctrl.stopped:
       buddy.bdyFetchRegisterError()
       trace trEthRecvReceivedBlockBodies, peer, nReq, nResp=0,
-        elapsed=elapsed.toStr, syncState=($buddy.syncState),
-        bdyErrors=buddy.bdyErrors
+        elapsed=elapsed.toStr, state=($buddy.syncState),
+        nErrors=buddy.nErrors.fetch.bdy
       break body                                    # return err()
 
+    # Verify the correct number of block bodies received
     let b = rc.value.packet.bodies
     if b.len == 0 or nReq < b.len:
-      buddy.bdyFetchRegisterError()
+      if nReq < b.len:
+        # Bogus peer returning additional rubbish
+        buddy.bdyFetchRegisterError(forceZombie=true)
+      else:
+        # Data not avail but fast enough answer: degrade througput stats only
+        discard buddy.only.thruPutStats.blk.bpsSample(elapsed, 0)
+        if fetchBodiesErrTimeout <= elapsed:
+          buddy.bdyFetchRegisterError(slowPeer=true)
       trace trEthRecvReceivedBlockBodies, peer, nReq, nResp=b.len,
-        elapsed=elapsed.toStr, syncState=($buddy.syncState),
-        nRespErrors=buddy.only.nRespErrors.blk
+        elapsed=elapsed.toStr, state=($buddy.syncState),
+        nErrors=buddy.nErrors.fetch.bdy
       break body                                    # return err()
 
     # Update download statistics
     let bps = buddy.only.thruPutStats.blk.bpsSample(elapsed, b.getEncodedLength)
 
-    # Ban an overly slow peer for a while when seen in a row. Also there is a
-    # mimimum share of the number of requested headers expected, typically 10%.
-    if fetchBodiesErrTimeout < elapsed or
-       b.len.uint64 * 100 < nReq.uint64 * fetchBodiesMinResponsePC:
+    # Ban an overly slow peer for a while when observed consecutively.
+    if fetchBodiesErrTimeout < elapsed:
       buddy.bdyFetchRegisterError(slowPeer=true)
     else:
-      buddy.only.nRespErrors.blk = 0                # reset error count
+      buddy.nErrors.fetch.bdy = 0                   # reset error count
       buddy.ctx.pool.lastSlowPeer = Opt.none(Hash)  # not last one or not error
 
     trace trEthRecvReceivedBlockBodies, peer, nReq, nResp=b.len,
       elapsed=elapsed.toStr, throughput=(bps.toIECb(1) & "ps"),
-      syncState=($buddy.syncState), bdyErrors=buddy.bdyErrors
+      state=($buddy.syncState), nErrors=buddy.nErrors.fetch.bdy
 
     bodyRc = Opt[seq[BlockBody]].ok(b)
 

execution_chain/sync/beacon/worker/blocks/blocks_helpers.nim

@@ -17,13 +17,16 @@ import
 # Public functions
 # ------------------------------------------------------------------------------
 
-func bdyErrors*(buddy: BeaconBuddyRef): string =
-  $buddy.only.nRespErrors.blk & "/" & $buddy.only.nProcErrors.blk
-
-proc bdyFetchRegisterError*(buddy: BeaconBuddyRef, slowPeer = false) =
-  buddy.only.nRespErrors.blk.inc
-  if nFetchBodiesErrThreshold < buddy.only.nRespErrors.blk:
-    if buddy.ctx.pool.nBuddies == 1 and slowPeer:
+func blkErrors*(buddy: BeaconBuddyRef): string =
+  $buddy.nErrors.fetch.bdy & "/" & $buddy.nErrors.apply.blk
+
+proc bdyFetchRegisterError*(
+    buddy: BeaconBuddyRef;
+    slowPeer = false;
+    forceZombie = false) =
+  buddy.nErrors.fetch.bdy.inc
+  if nFetchBodiesErrThreshold < buddy.nErrors.fetch.bdy:
+    if not forceZombie and buddy.ctx.pool.nBuddies == 1 and slowPeer:
       # Remember that the current peer is the last one and is lablelled slow.
       # It would have been zombified if it were not the last one. This can be
       # used in functions -- depending on context -- that will trigger if the