More efficient implementation of the 'POST beacon_committee_subscriptions' API (#3153)

Author: zah

beacon_chain/consensus_object_pools/sync_committee_msg_pool.nim

@@ -8,7 +8,8 @@
 {.push raises: [Defect].}
 
 import
-  std/[hashes, sets, tables],
+  std/[sets, tables],
+  stew/shims/hashes,
   chronicles,
   ../spec/digest,
   ../spec/datatypes/altair
@@ -53,7 +54,7 @@ type
     onContributionReceived*: OnSyncContributionCallback
 
 func hash*(x: SyncCommitteeMsgKey): Hash =
-  hashData(unsafeAddr x, sizeof(x))
+  hashAllFields(x)
 
 func init*(T: type SyncCommitteeMsgPool,
            onSyncContribution: OnSyncContributionCallback = nil

beacon_chain/eth1/eth1_monitor.nim

@@ -8,13 +8,13 @@
 {.push raises: [Defect].}
 
 import
-  std/[deques, hashes, options, strformat, strutils, sequtils, tables,
+  std/[deques, options, strformat, strutils, sequtils, tables,
        typetraits, uri, json],
   # Nimble packages:
   chronos, json, metrics, chronicles/timings,
   web3, web3/ethtypes as web3Types, web3/ethhexstrings, web3/engine_api,
   eth/common/eth_types,
-  eth/async_utils, stew/[objects, byteutils],
+  eth/async_utils, stew/[objects, byteutils, shims/hashes],
   # Local modules:
   ../spec/[eth2_merkleization, forks, helpers],
   ../spec/datatypes/[base, merge],
@@ -356,7 +356,7 @@ proc addBlock*(chain: var Eth1Chain, newBlock: Eth1Block) =
   eth1_chain_len.set chain.blocks.len.int64
 
 func hash*(x: Eth1Data): Hash =
-  hashData(unsafeAddr x, sizeof(x))
+  hash(x.block_hash)
 
 template hash*(x: Eth1Block): Hash =
   hash(x.voteData)

beacon_chain/rpc/rest_validator_api.nim

@@ -503,30 +503,38 @@ proc installValidatorApiHandlers*(router: var RestRouter, node: BeaconNode) =
     if not(node.isSynced(node.dag.head)):
       return RestApiResponse.jsonError(Http503, BeaconNodeInSyncError)
 
+    let
+      wallSlot = node.beaconClock.now.slotOrZero
+      wallEpoch = wallSlot.epoch
+      head = node.dag.head
+
+    var currentEpoch, nextEpoch: Option[EpochRef]
+    template getAndCacheEpochRef(epochRefVar: var Option[EpochRef],
+                                 epoch: Epoch): EpochRef =
+      if epochRefVar.isNone:
+        epochRefVar = some node.dag.getEpochRef(head, epoch)
+      epochRefVar.get
+
     for request in requests:
       if uint64(request.committee_index) >= uint64(MAX_COMMITTEES_PER_SLOT):
         return RestApiResponse.jsonError(Http400,
                                          InvalidCommitteeIndexValueError)
       if uint64(request.validator_index) >=
                   lenu64(getStateField(node.dag.headState.data, validators)):
         return RestApiResponse.jsonError(Http400,
-                                          InvalidValidatorIndexValueError)
-
-      let wallSlot = node.beaconClock.now.slotOrZero
+                                         InvalidValidatorIndexValueError)
       if wallSlot > request.slot + 1:
         return RestApiResponse.jsonError(Http400, SlotFromThePastError)
+
       let epoch = request.slot.epoch
-      if epoch >= wallSlot.epoch and epoch - wallSlot.epoch > 1:
+      let epochRef = if epoch == wallEpoch:
+        currentEpoch.getAndCacheEpochRef(wallEpoch)
+      elif epoch == wallEpoch + 1:
+        nextEpoch.getAndCacheEpochRef(wallEpoch + 1)
+      else:
         return RestApiResponse.jsonError(Http400,
                                          SlotNotInNextWallSlotEpochError)
-      let head =
-        block:
-          let res = node.getCurrentHead(epoch)
-          if res.isErr():
-            return RestApiResponse.jsonError(Http400, NoHeadForSlotError,
-                                             $res.error())
-          res.get()
-      let epochRef = node.dag.getEpochRef(head, epoch)
+
       let subnet_id = compute_subnet_for_attestation(
         get_committee_count_per_slot(epochRef), request.slot,
         request.committee_index)

beacon_chain/validators/action_tracker.nim

@@ -1,7 +1,7 @@
 import
-  std/[sequtils, sets, tables],
-  chronicles,
+  std/[tables, sequtils],
   bearssl,
+  stew/shims/[sets, hashes], chronicles,
   eth/p2p/discoveryv5/random2,
   ../spec/datatypes/base,
   ../spec/[helpers, network],
@@ -56,11 +56,14 @@ type
       ## subnet for each such validator - the slot is used to expire validators
       ## that no longer are posting duties
 
-    duties*: seq[AggregatorDuty] ##\
+    duties*: HashSet[AggregatorDuty] ##\
       ## Known aggregation duties in the near future - before each such
       ## duty, we'll subscribe to the corresponding subnet to collect
       ## attestations for the aggregate
 
+func hash*(x: AggregatorDuty): Hash =
+  hashAllFields(x)
+
 # https://github.com/ethereum/consensus-specs/blob/v1.1.4/specs/phase0/validator.md#phase-0-attestation-subnet-stability
 func randomStabilitySubnet*(
     self: ActionTracker, epoch: Epoch): tuple[subnet_id: SubnetId, expiration: Epoch] =
@@ -84,12 +87,11 @@ proc registerDuty*(
   if isAggregator:
     let newDuty = AggregatorDuty(slot: slot, subnet_id: subnet_id)
 
-    for duty in tracker.duties.mitems():
-      if duty == newDuty:
-        return
+    if newDuty in tracker.duties:
+      return
 
     debug "Registering aggregation duty", slot, subnet_id, vidx
-    tracker.duties.add(newDuty)
+    tracker.duties.incl(newDuty)
 
 const allSubnetBits = block:
   var res: AttnetBits
@@ -100,7 +102,6 @@ func aggregateSubnets*(tracker: ActionTracker, wallSlot: Slot): AttnetBits =
   var res: AttnetBits
   # Subscribe to subnets for upcoming duties
   for duty in tracker.duties:
-
     if wallSlot <= duty.slot and
         wallSlot + SUBNET_SUBSCRIPTION_LEAD_TIME_SLOTS > duty.slot: