Provide syncer session call back handlers that can be intercepted

Author: mjfh

execution_chain/sync/beacon.nim

@@ -15,9 +15,10 @@ import
   pkg/stew/[interval_set, sorted_set],
   ../core/chain,
   ../networking/p2p,
-  ./beacon/worker/headers/headers_target,
   ./beacon/[beacon_desc, worker],
-  ./beacon/worker/classify,
+  ./beacon/worker/blocks/[blocks_fetch, blocks_import],
+  ./beacon/worker/headers/[headers_fetch, headers_target],
+  ./beacon/worker/[classify, update],
   ./[sync_sched, wire_protocol]
 
 export
@@ -26,34 +27,64 @@ export
 logScope:
   topics = "beacon sync"
 
+# ------------------------------------------------------------------------------
+# Interceptable handlers
+# ------------------------------------------------------------------------------
+
+proc schedDaemonCB(
+    ctx: BeaconCtxRef;
+      ): Future[Duration]
+      {.async: (raises: []).} =
+  return worker.runDaemon(ctx, "Daemon") # async/template
+
+proc schedStartCB(buddy: BeaconBuddyRef): bool =
+  return worker.start(buddy, "Start")
+
+proc schedStopCB(buddy: BeaconBuddyRef) =
+  worker.stop(buddy, "Stop")
+
+proc schedPoolCB(buddy: BeaconBuddyRef; last: bool; laps: int): bool =
+  return worker.runPool(buddy, last, laps, "SyncMode")
+
+proc schedPeerCB(
+    buddy: BeaconBuddyRef;
+    rank: PeerRanking;
+      ): Future[Duration]
+      {.async: (raises: []).} =
+  return worker.runPeer(buddy, rank, "Peer") # async/template
+
+proc noOpFn(buddy: BeaconBuddyRef) = discard
+proc noOpEx(self: BeaconHandlersSyncRef) = discard
+
 # ------------------------------------------------------------------------------
 # Virtual methods/interface, `mixin` functions
 # ------------------------------------------------------------------------------
 
 proc runSetup(ctx: BeaconCtxRef): bool =
-  worker.setup(ctx, "Setup")
+  return worker.setup(ctx, "Setup")
 
 proc runRelease(ctx: BeaconCtxRef) =
   worker.release(ctx, "Release")
 
-proc runDaemon(ctx: BeaconCtxRef): Future[Duration] {.async: (raises: []).} =
-  return worker.runDaemon(ctx, "Daemon")
-
 proc runTicker(ctx: BeaconCtxRef) =
   worker.runTicker(ctx, "Ticker")
 
+
+proc runDaemon(ctx: BeaconCtxRef): Future[Duration] {.async: (raises: []).} =
+  return await ctx.handler.schedDaemon(ctx)
+
 proc runStart(buddy: BeaconBuddyRef): bool =
-  worker.start(buddy, "Start")
+  return buddy.ctx.handler.schedStart(buddy)
 
 proc runStop(buddy: BeaconBuddyRef) =
-  worker.stop(buddy, "Stop")
+  buddy.ctx.handler.schedStop(buddy)
 
 proc runPool(buddy: BeaconBuddyRef; last: bool; laps: int): bool =
-  worker.runPool(buddy, last, laps, "SyncMode")
+  return buddy.ctx.handler.schedPool(buddy, last, laps)
 
 proc runPeer(buddy: BeaconBuddyRef): Future[Duration] {.async: (raises: []).} =
   let rank = buddy.classifyForFetching()
-  return worker.runPeer(buddy, rank, "Peer")
+  return await buddy.ctx.handler.schedPeer(buddy, rank)
 
 # ------------------------------------------------------------------------------
 # Public functions
@@ -85,6 +116,25 @@ proc config*(
   desc.initSync(ethNode, maxPeers)
   desc.ctx.pool.chain = chain
 
+  # Set up handlers so they can be overlayed
+  desc.ctx.pool.handlers = BeaconHandlersSyncRef(
+    version:          0,
+    activate:         updateActivateCB,
+    suspend:          updateSuspendCB,
+    schedDaemon:      schedDaemonCB,
+    schedStart:       schedStartCB,
+    schedStop:        schedStopCB,
+    schedPool:        schedPoolCB,
+    schedPeer:        schedPeerCB,
+    getBlockHeaders:  getBlockHeadersCB,
+    syncBlockHeaders: noOpFn,
+    getBlockBodies:   getBlockBodiesCB,
+    syncBlockBodies:  noOpFn,
+    importBlock:      importBlockCB,
+    syncImportBlock:  noOpFn,
+    startSync:        noOpEx,
+    stopSync:         noOpEx)
+
   if not desc.lazyConfigHook.isNil:
     desc.lazyConfigHook(desc)
     desc.lazyConfigHook = nil
@@ -101,10 +151,16 @@ proc configTarget*(desc: BeaconSyncRef; hex: string; isFinal: bool): bool =
 
 proc start*(desc: BeaconSyncRef): bool =
   doAssert not desc.ctx.isNil
-  desc.startSync()
+  if desc.startSync():
+    let w = BeaconHandlersSyncRef(desc.ctx.pool.handlers)
+    w.startSync(w)
+    return true
+  # false
 
 proc stop*(desc: BeaconSyncRef) {.async.} =
   doAssert not desc.ctx.isNil
+  let w = BeaconHandlersSyncRef(desc.ctx.pool.handlers)
+  w.stopSync(w)
   await desc.stopSync()
 
 # ------------------------------------------------------------------------------

execution_chain/sync/beacon/beacon_desc.nim

@@ -25,4 +25,10 @@ type
     ## Instance descriptor, extends scheduler object
     lazyConfigHook*: BeaconSyncConfigHook
 
+  BeaconHandlersSyncRef* = ref object of BeaconHandlersRef
+    ## Add start/stop helpers to function list. By default, this functiona
+    ## are no-ops.
+    startSync*: proc(self: BeaconHandlersSyncRef) {.gcsafe, raises: [].}
+    stopSync*: proc(self: BeaconHandlersSyncRef) {.gcsafe, raises: [].}
+
 # End

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

@@ -17,12 +17,27 @@ import
   ../../../../networking/p2p,
   ../../../wire_protocol/types,
   ../[update, worker_desc],
-  ./[blocks_fetch, blocks_helpers, blocks_import, blocks_unproc]
+  ./[blocks_fetch, blocks_helpers, blocks_unproc]
 
 # ------------------------------------------------------------------------------
 # Private helpers
 # ------------------------------------------------------------------------------
 
+template importBlock(
+     buddy: BeaconBuddyRef;
+     blk: EthBlock;
+     effPeerID: Hash;
+       ): Result[Duration,BeaconError] =
+  ## Async/template
+  ##
+  ## Wrapper around `importBlock()` handler
+  ##
+  let
+    ctx = buddy.ctx
+    rc = await ctx.handler.importBlock(buddy, blk, effPeerID)
+  ctx.handler.syncImportBlock(buddy) # debugging, trace, replay
+  rc
+
 proc getNthHash(ctx: BeaconCtxRef; blocks: seq[EthBlock]; n: int): Hash32 =
   ctx.hdrCache.getHash(blocks[n].header.number).valueOr:
     return zeroHash32
@@ -202,7 +217,7 @@ template blocksImport*(
 
       for n in 0 ..< blocks.len:
         let nBn = blocks[n].header.number
-        discard (await buddy.importBlock(blocks[n], peerID)).valueOr:
+        buddy.importBlock(blocks[n], peerID).isOkOr:
           if error.excp != ECancelledError:
             isError = true
 

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

@@ -18,11 +18,30 @@ import
   ../worker_desc,
   ./blocks_helpers
 
+logScope:
+  topics = "beacon sync"
+
+# ------------------------------------------------------------------------------
+# Private helper
+# -----------------------------------------------------------------------------
+
+template getBlockBodies(
+    buddy: BeaconBuddyRef;
+    req: BlockBodiesRequest;
+       ): Result[FetchBodiesData,BeaconError] =
+  ## Async/template
+  ##
+  ## Wrapper around `getBlockBodies()` handler
+  ##
+  let rc = await buddy.ctx.handler.getBlockBodies(buddy, req)
+  buddy.ctx.handler.syncBlockBodies(buddy) # debugging, sync, replay
+  rc
+
 # ------------------------------------------------------------------------------
-# Private helpers
+# Public handler
 # -----------------------------------------------------------------------------
 
-proc getBlockBodies(
+proc getBlockBodiesCB*(
     buddy: BeaconBuddyRef;
     req: BlockBodiesRequest;
       ): Future[Result[FetchBodiesData,BeaconError]]
@@ -75,7 +94,7 @@ template fetchBodies*(
     trace trEthSendSendingGetBlockBodies,
       peer, nReq, nErrors=buddy.nErrors.fetch.bdy
 
-    let rc = await buddy.getBlockBodies(request)
+    let rc = buddy.getBlockBodies(request)
     var elapsed: Duration
     if rc.isOk:
       elapsed = rc.value.elapsed

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

@@ -16,11 +16,14 @@ import
   ../../../wire_protocol,
   ../worker_desc
 
+logScope:
+  topics = "beacon sync"
+
 # ------------------------------------------------------------------------------
-# Public function
+# Public handler
 # ------------------------------------------------------------------------------
 
-proc importBlock*(
+proc importBlockCB*(
     buddy: BeaconBuddyRef;
     blk: EthBlock;
     effPeerID: Hash;

execution_chain/sync/beacon/worker/headers/headers_fetch.nim

@@ -18,11 +18,31 @@ import
   ../worker_desc,
   ./headers_helpers
 
+logScope:
+  topics = "beacon sync"
+
 # ------------------------------------------------------------------------------
 # Private helpers
+# -----------------------------------------------------------------------------
+
+template getBlockHeaders(
+    buddy: BeaconBuddyRef;
+    req: BlockHeadersRequest;
+    bn: BlockNumber;
+       ): Result[FetchHeadersData,BeaconError] =
+  ## Async/template
+  ##
+  ## Wrapper around `getBlockHeaders()` handler
+  ##
+  let rc = await buddy.ctx.handler.getBlockHeaders(buddy, req, bn)
+  buddy.ctx.handler.syncBlockHeaders(buddy) # debugging, sync, replay
+  rc
+
+# ------------------------------------------------------------------------------
+# Public handler
 # ------------------------------------------------------------------------------
 
-proc getBlockHeaders(
+proc getBlockHeadersCB*(
     buddy: BeaconBuddyRef;
     req: BlockHeadersRequest;
     bn: BlockNumber;
@@ -94,7 +114,7 @@ template fetchHeadersReversed*(
     trace trEthSendSendingGetBlockHeaders & " reverse", peer, req=ivReq,
       nReq=req.maxResults, hash=topHash.toStr, nErrors=buddy.nErrors.fetch.hdr
 
-    let rc = await buddy.getBlockHeaders(req, BlockNumber ivReq.maxPt)
+    let rc = buddy.getBlockHeaders(req, BlockNumber ivReq.maxPt)
     var elapsed: Duration
     if rc.isOk:
       elapsed = rc.value.elapsed

execution_chain/sync/beacon/worker/start_stop.nim

@@ -14,7 +14,7 @@ import
   pkg/[chronicles, chronos, eth/common, metrics],
   ../../../networking/p2p,
   ../../wire_protocol,
-  ./[blocks, headers, update, worker_desc]
+  ./[blocks, headers, worker_desc]
 
 type
   SyncStateData = tuple
@@ -59,8 +59,8 @@ proc setupServices*(ctx: BeaconCtxRef; info: static[string]) =
 
   # Set up the notifier informing when a new syncer session has started.
   ctx.hdrCache.start proc() =
-    # Activates the syncer. Work will be picked up by peers when available.
-    ctx.updateActivateSyncer()
+    # This directive captures `ctx` for calling the activation handler.
+    ctx.handler.activate(ctx)
 
   # Provide progress info call back handler
   ctx.pool.chain.com.beaconSyncerProgress = proc(): SyncStateData =

execution_chain/sync/beacon/worker/update.nim

@@ -30,23 +30,6 @@ declareGauge nec_sync_head, "" &
 # Private functions, state handler helpers
 # ------------------------------------------------------------------------------
 
-proc updateSuspendSyncer(ctx: BeaconCtxRef) =
-  ## Clean up sync target buckets, stop syncer activity, and and get ready
-  ## for awaiting a new request from the `CL`.
-  ##
-  ctx.hdrCache.clear()
-
-  ctx.pool.failedPeers.clear()
-  ctx.pool.seenData = false
-
-  ctx.hibernate = true
-
-  metrics.set(nec_sync_last_block_imported, 0)
-  metrics.set(nec_sync_head, 0)
-
-  info "Suspending syncer", base=ctx.chain.baseNumber.bnStr,
-    head=ctx.chain.latestNumber.bnStr, nSyncPeers=ctx.pool.nBuddies
-
 proc commitCollectHeaders(ctx: BeaconCtxRef; info: static[string]): bool =
   ## Link header chain into `FC` module. Gets ready for block import.
   ##
@@ -227,7 +210,7 @@ proc updateSyncState*(ctx: BeaconCtxRef; info: static[string]) =
 
   # Final sync scrum layout reached or inconsistent/impossible state
   if newState == idle:
-    ctx.updateSuspendSyncer()
+    ctx.handler.suspend(ctx)
 
 
 proc updateLastBlockImported*(ctx: BeaconCtxRef; bn: BlockNumber) =
@@ -238,7 +221,7 @@ proc updateLastBlockImported*(ctx: BeaconCtxRef; bn: BlockNumber) =
 # Public functions, call-back handler ready
 # ------------------------------------------------------------------------------
 
-proc updateActivateSyncer*(ctx: BeaconCtxRef) =
+proc updateActivateCB*(ctx: BeaconCtxRef) =
   ## If in hibernate mode, accept a cache session and activate syncer
   ##
   if ctx.hibernate and                          # only in idle mode
@@ -278,6 +261,24 @@ proc updateActivateSyncer*(ctx: BeaconCtxRef) =
   # Failed somewhere on the way
   ctx.hdrCache.clear()
 
+
+proc updateSuspendCB*(ctx: BeaconCtxRef) =
+  ## Clean up sync target buckets, stop syncer activity, and and get ready
+  ## for a new sync request from the `CL`.
+  ##
+  ctx.hdrCache.clear()
+
+  ctx.pool.failedPeers.clear()
+  ctx.pool.seenData = false
+
+  ctx.hibernate = true
+
+  metrics.set(nec_sync_last_block_imported, 0)
+  metrics.set(nec_sync_head, 0)
+
+  info "Suspending syncer", base=ctx.chain.baseNumber.bnStr,
+    head=ctx.chain.latestNumber.bnStr, nSyncPeers=ctx.pool.nBuddies
+
 # ------------------------------------------------------------------------------
 # End
 # ------------------------------------------------------------------------------