Beacon sync maint update (#3520)

* Move thread switch nano-sleep into `importBlock()` function

why
  That way it becomes easier to capture internal states for
  trace/replay.

  Previously, the nano-sleep thread switch was stated after
  `importBlock()` as a separate directive. So there is no functional
  change for the production environment.

* Move async idle times from `runDaemon()` and `runPeer()` to scheduler

why
  The sleep function extends the minimum idle wait imposed by the
  scheduler if there is nothing to do.

  By moving it there, the `runPeer()` directive becomes more environment
  capture friendly for trace/replace.

  The functionality of the production system is not affected.

* Dissolve generic async functions as templates

why
  This change is inspired by trace/replay considerations.

  For the production system, this change reduces the number of
  *floating* async closures which are unnecessary and reduces the
  `chronos` overhead.

  These *floating* async closures appear when writing nested functions
  (rather than a big one) where some leaf functions call async closures
  as it is common with network communication.

* Move out most metrics gauge handling to dedicated source files

why
  Reduces latency when updated right with the state/counter changes.

todo
  Move some gauges to `FC` module

Author: mjfh

execution_chain/core/chain/header_chain_cache.nim

@@ -60,10 +60,8 @@
 {.push raises:[].}
 
 import
+  pkg/[chronicles, metrics, stew/endians2],
   pkg/eth/[common, rlp],
-  pkg/results,
-  pkg/chronicles,
-  pkg/stew/endians2,
   "../.."/[common, db/core_db, db/storage_types],
   ../../db/[kvt, kvt_cf],
   ../../db/kvt/[kvt_utils, kvt_tx_frame],
@@ -73,6 +71,12 @@ import
 logScope:
   topics = "hc-cache"
 
+declareGauge nec_sync_dangling, "" &
+  "Least block number for header chain already fetched"
+
+declareGauge nec_sync_consensus_head, "" &
+  "Block number of latest consensus head"
+
 type
   HccDbInfo = object
     ## For database table storage and clean up
@@ -318,13 +322,15 @@ proc headUpdateFromCL(hc: HeaderChainRef; h: Header; f: Hash32) =
         headHash: h.computeBlockHash())
 
       hc.kvt.putHeader(h)
+      metrics.set(nec_sync_dangling, h.number.int64)
 
       # Inform client app about that a new session has started.
       hc.notify()
       hc.chain.pendingFCU = f
 
     # For logging and metrics
     hc.session.consHeadNum = h.number
+    metrics.set(nec_sync_consensus_head, h.number.int64)
 
 # ------------------------------------------------------------------------------
 # Public constructor/destructor et al.
@@ -349,9 +355,10 @@ proc clear*(hc: HeaderChainRef) =
   ## accepted session (via `accept()`) or a mere notified session (via `notify`
   ## call back argument from `start()`.)
   ##
-  hc.session.reset                   # clear session state object
-  hc.persistClear()                  # clear database
-
+  hc.session.reset                        # clear session state object
+  hc.persistClear()                       # clear database
+  metrics.set(nec_sync_consensus_head, 0) # clear metrics register
+  metrics.set(nec_sync_dangling, 0)       # ditto
 
 proc stop*(hc: HeaderChainRef) =
   ## Stop updating the client cache. Will automatically be called by the
@@ -543,6 +550,7 @@ proc put*(
 
     # Set new antecedent `ante` and save to disk (if any)
     hc.session.ante = rev[revTopInx]
+    metrics.set(nec_sync_dangling, hc.session.ante.number.int64)
 
     # Save updates. persist to DB
     hc.persistInfo()
@@ -582,6 +590,7 @@ proc commit*(hc: HeaderChainRef): Result[void,string] =
       if hc.chain.hashToBlock.hasKey(hc.chain.pendingFCU):
         hc.session.ante = fin
         hc.session.mode = locked                      # update internal state
+        metrics.set(nec_sync_dangling, fin.number.int64)
         return ok()
 
       # Impossible situation!

execution_chain/sync/beacon.nim

@@ -35,8 +35,8 @@ proc runSetup(ctx: BeaconCtxRef): bool =
 proc runRelease(ctx: BeaconCtxRef) =
   worker.release(ctx, "RunRelease")
 
-proc runDaemon(ctx: BeaconCtxRef) {.async: (raises: []).} =
-  await worker.runDaemon(ctx, "RunDaemon")
+proc runDaemon(ctx: BeaconCtxRef): Future[Duration] {.async: (raises: []).} =
+  return worker.runDaemon(ctx, "RunDaemon")
 
 proc runTicker(ctx: BeaconCtxRef) =
   worker.runTicker(ctx, "RunTicker")
@@ -50,8 +50,8 @@ proc runStop(buddy: BeaconBuddyRef) =
 proc runPool(buddy: BeaconBuddyRef; last: bool; laps: int): bool =
   worker.runPool(buddy, last, laps, "RunPool")
 
-proc runPeer(buddy: BeaconBuddyRef) {.async: (raises: []).} =
-  await worker.runPeer(buddy, "RunPeer")
+proc runPeer(buddy: BeaconBuddyRef): Future[Duration] {.async: (raises: []).} =
+  return worker.runPeer(buddy, "RunPeer")
 
 # ------------------------------------------------------------------------------
 # Public functions

execution_chain/sync/beacon/worker.nim

@@ -24,21 +24,13 @@ import
 # Private functions
 # ------------------------------------------------------------------------------
 
-proc napUnlessSomethingToCollect(
-    buddy: BeaconBuddyRef;
-      ): Future[bool] {.async: (raises: []).} =
-  ## When idle, save cpu cycles waiting for something to do.
-  if buddy.ctx.hibernate or                      # not activated yet?
-     not (buddy.headersCollectOk() or            # something on TODO list
-          buddy.blocksCollectOk()):
-    try:
-      await sleepAsync workerIdleWaitInterval
-    except CancelledError:
-      buddy.ctrl.stopped = true
-    return true
-  else:
-    # Returning `false` => no need to check for shutdown
+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
 
 # ------------------------------------------------------------------------------
 # Public start/stop and admin functions
@@ -107,35 +99,39 @@ proc runTicker*(ctx: BeaconCtxRef; info: static[string]) =
   ctx.updateTicker()
 
 
-proc runDaemon*(
-    ctx: BeaconCtxRef;
-    info: static[string];
-      ) {.async: (raises: []).} =
+template runDaemon*(ctx: BeaconCtxRef; info: static[string]): Duration =
+  ## Async/template
+  ##
   ## Global background job that will be re-started as long as the variable
   ## `ctx.daemon` is set `true` which corresponds to `ctx.hibernating` set
   ## to false.
   ##
   ## On a fresh start, the flag `ctx.daemon` will not be set `true` before the
   ## first usable request from the CL (via RPC) stumbles in.
   ##
-  # Check for a possible header layout and body request changes
-  ctx.updateSyncState info
-  if ctx.hibernate:
-    return
+  ## The template returns a suggested idle time for after this task.
+  ##
+  var bodyRc = chronos.nanoseconds(0)
+  block body:
+    # Check for a possible header layout and body request changes
+    ctx.updateSyncState info
+    if ctx.hibernate:
+      break body             # return
+
+    # Execute staged block records.
+    if ctx.blocksUnstageOk():
 
-  # Execute staged block records.
-  if ctx.blocksUnstageOk():
+      # Import bodies from the `staged` queue.
+      discard ctx.blocksUnstage info # async/template
 
-    # Import bodies from the `staged` queue.
-    discard await ctx.blocksUnstage info
+      if not ctx.daemon or   # Implied by external sync shutdown?
+         ctx.poolMode:       # Oops, re-org needed?
+        break body           # return
 
-    if not ctx.daemon or   # Implied by external sync shutdown?
-       ctx.poolMode:       # Oops, re-org needed?
-      return
+    # # At the end of the cycle, leave time to trigger refill headers/blocks
+    bodyRc = daemonWaitInterval
 
-  # At the end of the cycle, leave time to trigger refill headers/blocks
-  try: await sleepAsync daemonWaitInterval
-  except CancelledError: discard
+  bodyRc
 
 
 proc runPool*(
@@ -163,47 +159,59 @@ proc runPool*(
   true # stop
 
 
-proc runPeer*(
-    buddy: BeaconBuddyRef;
-    info: static[string];
-      ) {.async: (raises: []).} =
+template runPeer*(buddy: BeaconBuddyRef; info: static[string]): Duration =
+  ## Async/template
+  ##
   ## This peer worker method is repeatedly invoked (exactly one per peer) while
   ## the `buddy.ctrl.poolMode` flag is set `false`.
   ##
-  if not await buddy.napUnlessSomethingToCollect():
-
-    # Download and process headers and blocks
-    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.
-      await buddy.headersCollect info
-
-      # 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.)
-        break
-
-      # 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.
-      await buddy.blocksCollect info
-
-      # Import bodies from the `staged` queue.
-      if not await buddy.blocksUnstage info:
-        # Need to proceed with another peer (e.g. gap between top imported
-        # block and blocks queue.)
-        break
-
-      # End `while()`
+  ## The template returns a suggested idle time for after this task.
+  ##
+  var bodyRc = chronos.nanoseconds(0)
+  block body:
+    if buddy.somethingToCollect():
+
+      # Download and process headers and blocks
+      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:
+          # Need to proceed with another peer (e.g. gap between queue and
+          # header chain cache.)
+          bodyRc = workerIdleWaitInterval
+          break body
+
+        # 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 body
+
+        # End `while()`
+
+  # Idle sleep unless there is something to do
+  if not buddy.somethingToCollect():
+    bodyRc = workerIdleWaitInterval
+
+  bodyRc
 
 # ------------------------------------------------------------------------------
 # End