attestation processing speedups

* avoid creating indexed attestation just to check signatures - above all, don't create it when not checking signatures ;) * avoid pointer op when adding attestation to pool * better iterator for yielding attestations * add metric / log for attestation packing time
2021-04-14 16:43:29 +02:00 · 2021-04-14 16:43:29 +02:00 · f1f424cc2d
parent 6806ffe1c8
commit f1f424cc2d
5 changed files with 135 additions and 108 deletions
--- a/beacon_chain/consensus_object_pools/attestation_pool.nim
+++ b/beacon_chain/consensus_object_pools/attestation_pool.nim
@ -11,18 +11,22 @@ import
  # Standard libraries
  std/[options, tables, sequtils],
  # Status libraries
+  metrics,
  chronicles, stew/byteutils, json_serialization/std/sets as jsonSets,
  # Internal
  ../spec/[beaconstate, datatypes, crypto, digest, validator],
  ../ssz/merkleization,
  "."/[spec_cache, blockchain_dag, block_quarantine],
-  ../beacon_node_types, ../extras,
+  ".."/[beacon_clock, beacon_node_types, extras],
  ../fork_choice/fork_choice

 export beacon_node_types

 logScope: topics = "attpool"

+declareGauge attestation_pool_block_attestation_packing_time,
+  "Time it took to create list of attestations for block"
+
 proc init*(T: type AttestationPool, chainDag: ChainDAGRef, quarantine: QuarantineRef): T =
  ## Initialize an AttestationPool from the chainDag `headState`
  ## The `finalized_root` works around the finalized_checkpoint of the genesis block
@ -102,12 +106,12 @@ proc addForkChoiceVotes(
      # hopefully the fork choice will heal itself over time.
      error "Couldn't add attestation to fork choice, bug?", err = v.error()

-func candidateIdx(pool: AttestationPool, slot: Slot): Option[uint64] =
+func candidateIdx(pool: AttestationPool, slot: Slot): Option[int] =
  if slot >= pool.startingSlot and
      slot < (pool.startingSlot + pool.candidates.lenu64):
-    some(slot mod pool.candidates.lenu64)
+    some(int(slot mod pool.candidates.lenu64))
  else:
-    none(uint64)
+    none(int)

 proc updateCurrent(pool: var AttestationPool, wallSlot: Slot) =
  if wallSlot + 1 < pool.candidates.lenu64:
@ -210,6 +214,52 @@ func updateAggregates(entry: var AttestationEntry) =
            inc j
        inc i

+proc addAttestation(entry: var AttestationEntry,
+                    attestation: Attestation,
+                    signature: CookedSig): bool =
+  logScope:
+    attestation = shortLog(attestation)
+
+  let
+    singleIndex = oneIndex(attestation.aggregation_bits)
+
+  if singleIndex.isSome():
+    if singleIndex.get() in entry.singles:
+      trace "Attestation already seen",
+        singles = entry.singles.len(),
+        aggregates = entry.aggregates.len()
+
+      return false
+
+    debug "Attestation resolved",
+      singles = entry.singles.len(),
+      aggregates = entry.aggregates.len()
+
+    entry.singles[singleIndex.get()] = signature
+  else:
+    # More than one vote in this attestation
+    for i in 0..<entry.aggregates.len():
+      if attestation.aggregation_bits.isSubsetOf(entry.aggregates[i].aggregation_bits):
+        trace "Aggregate already seen",
+          singles = entry.singles.len(),
+          aggregates = entry.aggregates.len()
+        return false
+
+    # Since we're adding a new aggregate, we can now remove existing
+    # aggregates that don't add any new votes
+    entry.aggregates.keepItIf(
+      not it.aggregation_bits.isSubsetOf(attestation.aggregation_bits))
+
+    entry.aggregates.add(Validation(
+      aggregation_bits: attestation.aggregation_bits,
+      aggregate_signature: AggregateSignature.init(signature)))
+
+    debug "Aggregate resolved",
+      singles = entry.singles.len(),
+      aggregates = entry.aggregates.len()
+
+  true
+
 proc addAttestation*(pool: var AttestationPool,
                     attestation: Attestation,
                     participants: seq[ValidatorIndex],
@ -234,50 +284,23 @@ proc addAttestation*(pool: var AttestationPool,
      startingSlot = pool.startingSlot
    return

-  let
-    singleIndex = oneIndex(attestation.aggregation_bits)
-    root = hash_tree_root(attestation.data)
-    # Careful with pointer, candidate table must not be touched after here
-    entry = addr pool.candidates[candidateIdx.get].mGetOrPut(
-      root,
-      AttestationEntry(
-        data: attestation.data,
-        committee_len: attestation.aggregation_bits.len()))
-
-  if singleIndex.isSome():
-    if singleIndex.get() in entry[].singles:
-      trace "Attestation already seen",
-        singles = entry[].singles.len(),
-        aggregates = entry[].aggregates.len()
+  let attestation_data_root = hash_tree_root(attestation.data)

+  # TODO withValue is an abomination but hard to use anything else too without
+  #      creating an unnecessary AttestationEntry on the hot path and avoiding
+  #      multiple lookups
+  pool.candidates[candidateIdx.get()].withValue(attestation_data_root, entry) do:
+    if not addAttestation(entry[], attestation, signature):
+      return
+  do:
+    if not addAttestation(
+        pool.candidates[candidateIdx.get()].mGetOrPut(
+          attestation_data_root,
+          AttestationEntry(
+            data: attestation.data,
+            committee_len: attestation.aggregation_bits.len())),
+        attestation, signature):
      return
-
-    debug "Attestation resolved",
-      singles = entry[].singles.len(),
-      aggregates = entry[].aggregates.len()
-
-    entry[].singles[singleIndex.get()] = signature
-  else:
-    # More than one vote in this attestation
-    for i in 0..<entry[].aggregates.len():
-      if attestation.aggregation_bits.isSubsetOf(entry[].aggregates[i].aggregation_bits):
-        trace "Aggregate already seen",
-          singles = entry[].singles.len(),
-          aggregates = entry[].aggregates.len()
-        return
-
-    # Since we're adding a new aggregate, we can now remove existing
-    # aggregates that don't add any new votes
-    entry[].aggregates.keepItIf(
-      not it.aggregation_bits.isSubsetOf(attestation.aggregation_bits))
-
-    entry[].aggregates.add(Validation(
-      aggregation_bits: attestation.aggregation_bits,
-      aggregate_signature: AggregateSignature.init(signature)))
-
-    debug "Aggregate resolved",
-      singles = entry[].singles.len(),
-      aggregates = entry[].aggregates.len()

  pool.addForkChoiceVotes(
    attestation.data.slot, participants, attestation.data.beacon_block_root,
@ -301,8 +324,18 @@ proc addForkChoice*(pool: var AttestationPool,

 iterator attestations*(pool: AttestationPool, slot: Option[Slot],
                       index: Option[CommitteeIndex]): Attestation =
-  template processTable(table: AttestationTable) =
-    for _, entry in table:
+  let candidateIndices =
+    if slot.isSome():
+      let candidateIdx = pool.candidateIdx(slot.get())
+      if candidateIdx.isSome():
+        candidateIdx.get() .. candidateIdx.get()
+      else:
+        1 .. 0
+    else:
+      0 ..< pool.candidates.len()
+
+  for candidateIndex in candidateIndices:
+    for _, entry in pool.candidates[candidateIndex]:
      if index.isNone() or entry.data.index == index.get().uint64:
        var singleAttestation = Attestation(
          aggregation_bits: CommitteeValidatorsBits.init(entry.committee_len),
@ -317,14 +350,6 @@ iterator attestations*(pool: AttestationPool, slot: Option[Slot],
        for v in entry.aggregates:
          yield entry.toAttestation(v)

-  if slot.isSome():
-    let candidateIdx = pool.candidateIdx(slot.get())
-    if candidateIdx.isSome():
-      processTable(pool.candidates[candidateIdx.get()])
-  else:
-    for i in 0..<pool.candidates.len():
-      processTable(pool.candidates[i])
-
 type
  AttestationCacheKey* = (Slot, uint64)
  AttestationCache = Table[AttestationCacheKey, CommitteeValidatorsBits] ##\
@ -393,6 +418,7 @@ proc getAttestationsForBlock*(pool: var AttestationPool,
    # Attestations produced in a particular slot are added to the block
    # at the slot where at least MIN_ATTESTATION_INCLUSION_DELAY have passed
    maxAttestationSlot = newBlockSlot - MIN_ATTESTATION_INCLUSION_DELAY
+    startPackingTime = Moment.now()

  var
    candidates: seq[tuple[
@ -422,9 +448,8 @@ proc getAttestationsForBlock*(pool: var AttestationPool,
        # Attestations are checked based on the state that we're adding the
        # attestation to - there might have been a fork between when we first
        # saw the attestation and the time that we added it
-        # TODO avoid creating a full attestation here and instead do the checks
-        #      based on the attestation data and bits
-        if not check_attestation(state, attestation, {skipBlsValidation}, cache).isOk():
+        if not check_attestation(
+              state, attestation, {skipBlsValidation}, cache).isOk():
          continue

        let score = attCache.score(
@ -453,7 +478,7 @@ proc getAttestationsForBlock*(pool: var AttestationPool,
      state.previous_epoch_attestations.maxLen - state.previous_epoch_attestations.len()

  var res: seq[Attestation]
-
+  let totalCandidates = candidates.len()
  while candidates.len > 0 and res.lenu64() < MAX_ATTESTATIONS:
    block:
      # Find the candidate with the highest score - slot is used as a
@ -491,6 +516,14 @@ proc getAttestationsForBlock*(pool: var AttestationPool,
        # Only keep candidates that might add coverage
        it.score > 0

+  let
+    packingTime = Moment.now() - startPackingTime
+
+  debug "Packed attestations for block",
+    newBlockSlot, packingTime, totalCandidates, attestations = res.len()
+  attestation_pool_block_attestation_packing_time.set(
+    packingTime.toFloatSeconds())
+
  res

 func bestValidation(aggregates: openArray[Validation]): (int, int) =
--- a/beacon_chain/consensus_object_pools/spec_cache.nim
+++ b/beacon_chain/consensus_object_pools/spec_cache.nim
@ -145,24 +145,29 @@ proc is_valid_indexed_attestation*(
 # https://github.com/ethereum/eth2.0-specs/blob/v1.0.1/specs/phase0/beacon-chain.md#is_valid_indexed_attestation
 proc is_valid_indexed_attestation*(
    fork: Fork, genesis_validators_root: Eth2Digest,
-    epochRef: EpochRef, attesting_indices: auto,
+    epochRef: EpochRef,
    attestation: SomeAttestation, flags: UpdateFlags): Result[void, cstring] =
  # This is a variation on `is_valid_indexed_attestation` that works directly
  # with an attestation instead of first constructing an `IndexedAttestation`
  # and then validating it - for the purpose of validating the signature, the
  # order doesn't matter and we can proceed straight to validating the
  # signature instead
-  if attesting_indices.len == 0:
-    return err("indexed_attestation: no attesting indices")
+  let sigs = attestation.aggregation_bits.countOnes()
+  if sigs == 0:
+    return err("is_valid_indexed_attestation: no attesting indices")

  # Verify aggregate signature
  if not (skipBLSValidation in flags or attestation.signature is TrustedSig):
-    let pubkeys = mapIt(
-      attesting_indices, epochRef.validator_keys[it])
+    var
+      pubkeys = newSeqOfCap[ValidatorPubKey](sigs)
+    for index in get_attesting_indices(
+        epochRef, attestation.data, attestation.aggregation_bits):
+      pubkeys.add(epochRef.validator_keys[index])
+
    if not verify_attestation_signature(
        fork, genesis_validators_root, attestation.data,
        pubkeys, attestation.signature):
-      return err("indexed attestation: signature verification failure")
+      return err("is_valid_indexed_attestation: signature verification failure")

  ok()

--- a/beacon_chain/gossip_processing/gossip_validation.nim
+++ b/beacon_chain/gossip_processing/gossip_validation.nim
@ -277,8 +277,7 @@ proc validateAttestation*(
  block:
    # First pass - without cryptography
    let v = is_valid_indexed_attestation(
-        fork, genesis_validators_root, epochRef, attesting_indices,
-        attestation,
+        fork, genesis_validators_root, epochRef, attestation,
        {skipBLSValidation})
    if v.isErr():
      return err((ValidationResult.Reject, v.error))
--- a/beacon_chain/spec/beaconstate.nim
+++ b/beacon_chain/spec/beaconstate.nim
@ -489,53 +489,44 @@ iterator get_attesting_indices*(state: BeaconState,
                                bits: CommitteeValidatorsBits,
                                cache: var StateCache): ValidatorIndex =
  ## Return the set of attesting indices corresponding to ``data`` and ``bits``.
-  if bits.lenu64 != get_beacon_committee_len(state, data.slot, data.index.CommitteeIndex, cache):
+  if bits.lenu64 != get_beacon_committee_len(
+      state, data.slot, data.index.CommitteeIndex, cache):
    trace "get_attesting_indices: inconsistent aggregation and committee length"
  else:
    var i = 0
-    for index in get_beacon_committee(state, data.slot, data.index.CommitteeIndex, cache):
+    for index in get_beacon_committee(
+        state, data.slot, data.index.CommitteeIndex, cache):
      if bits[i]:
        yield index
      inc i

-iterator get_sorted_attesting_indices*(state: BeaconState,
-                                       data: AttestationData,
-                                       bits: CommitteeValidatorsBits,
-                                       cache: var StateCache): ValidatorIndex =
-  var heap = initHeapQueue[ValidatorIndex]()
-  for index in get_attesting_indices(state, data, bits, cache):
-    heap.push(index)
+proc is_valid_indexed_attestation*(
+    state: BeaconState, attestation: SomeAttestation, flags: UpdateFlags,
+    cache: var StateCache): Result[void, cstring] =
+  # This is a variation on `is_valid_indexed_attestation` that works directly
+  # with an attestation instead of first constructing an `IndexedAttestation`
+  # and then validating it - for the purpose of validating the signature, the
+  # order doesn't matter and we can proceed straight to validating the
+  # signature instead

-  while heap.len > 0:
-    yield heap.pop()
+  let sigs = attestation.aggregation_bits.countOnes()
+  if sigs == 0:
+    return err("is_valid_indexed_attestation: no attesting indices")

-func get_sorted_attesting_indices_list*(
-    state: BeaconState, data: AttestationData, bits: CommitteeValidatorsBits,
-    cache: var StateCache): List[uint64, Limit MAX_VALIDATORS_PER_COMMITTEE] =
-  for index in get_sorted_attesting_indices(state, data, bits, cache):
-    if not result.add index.uint64:
-      raiseAssert "The `result` list has the same max size as the sorted `bits` input"
+  # Verify aggregate signature
+  if not (skipBLSValidation in flags or attestation.signature is TrustedSig):
+    var
+      pubkeys = newSeqOfCap[ValidatorPubKey](sigs)
+    for index in get_attesting_indices(
+        state, attestation.data, attestation.aggregation_bits, cache):
+      pubkeys.add(state.validators[index].pubkey)

-# https://github.com/ethereum/eth2.0-specs/blob/v1.0.1/specs/phase0/beacon-chain.md#get_indexed_attestation
-func get_indexed_attestation(state: BeaconState, attestation: Attestation,
-    cache: var StateCache): IndexedAttestation =
-  ## Return the indexed attestation corresponding to ``attestation``.
-  IndexedAttestation(
-    attesting_indices: get_sorted_attesting_indices_list(
-      state, attestation.data, attestation.aggregation_bits, cache),
-    data: attestation.data,
-    signature: attestation.signature
-  )
+    if not verify_attestation_signature(
+        state.fork, state.genesis_validators_root, attestation.data,
+        pubkeys, attestation.signature):
+      return err("indexed attestation: signature verification failure")

-func get_indexed_attestation(state: BeaconState, attestation: TrustedAttestation,
-    cache: var StateCache): TrustedIndexedAttestation =
-  ## Return the indexed attestation corresponding to ``attestation``.
-  TrustedIndexedAttestation(
-    attesting_indices: get_sorted_attesting_indices_list(
-      state, attestation.data, attestation.aggregation_bits, cache),
-    data: attestation.data,
-    signature: attestation.signature
-  )
+  ok()

 # Attestation validation
 # ------------------------------------------------------------------------------------------
@ -610,8 +601,7 @@ proc check_attestation*(
    if not (data.source == state.previous_justified_checkpoint):
      return err("FFG data not matching previous justified epoch")

-  ? is_valid_indexed_attestation(
-      state, get_indexed_attestation(state, attestation, cache), flags)
+  ? is_valid_indexed_attestation(state, attestation, flags, cache)

  ok()

--- a/beacon_chain/validators/validator_duties.nim
+++ b/beacon_chain/validators/validator_duties.nim
@ -218,7 +218,7 @@ proc createAndSendAttestation(node: BeaconNode,
  let deadline = attestationData.slot.toBeaconTime() +
                 seconds(int(SECONDS_PER_SLOT div 3))

-  let (delayStr, delayMillis) =
+  let (delayStr, delaySecs) =
    if wallTime < deadline:
      ("-" & $(deadline - wallTime), -toFloatSeconds(deadline - wallTime))
    else:
@ -228,7 +228,7 @@ proc createAndSendAttestation(node: BeaconNode,
                             validator = shortLog(validator), delay = delayStr,
                             indexInCommittee = indexInCommittee

-  beacon_attestation_sent_delay.observe(delayMillis)
+  beacon_attestation_sent_delay.observe(delaySecs)

 proc getBlockProposalEth1Data*(node: BeaconNode,
                               stateData: StateData): BlockProposalEth1Data =