fix attestation handling bottlenecks in slot processing and quadratic-time inclusion delay rewards/palties in epoch processing; create infrastructure for dynamically (un)subscribing to/from attestation topics (#1294)

This commit is contained in:
tersec 2020-07-10 09:24:04 +00:00 committed by GitHub
parent 396825344c
commit 5d5957fd7e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 50 additions and 57 deletions

View File

@ -76,5 +76,5 @@ task test, "Run all tests":
buildAndRunBinary "all_fixtures_require_ssz", "tests/official/", "-d:chronicles_log_level=TRACE -d:const_preset=mainnet" buildAndRunBinary "all_fixtures_require_ssz", "tests/official/", "-d:chronicles_log_level=TRACE -d:const_preset=mainnet"
# State and block sims; getting to 4th epoch triggers consensus checks # State and block sims; getting to 4th epoch triggers consensus checks
buildAndRunBinary "state_sim", "research/", "-d:const_preset=mainnet", "--validators=2000 --slots=128" buildAndRunBinary "state_sim", "research/", "-d:const_preset=mainnet", "--validators=3000 --slots=128"
buildAndRunBinary "block_sim", "research/", "-d:const_preset=mainnet", "--validators=2000 --slots=128" buildAndRunBinary "block_sim", "research/", "-d:const_preset=mainnet", "--validators=3000 --slots=128"

View File

@ -9,7 +9,7 @@
import import
# Standard libraries # Standard libraries
deques, sequtils, tables, options, algorithm, deques, sequtils, tables, options,
# Status libraries # Status libraries
chronicles, stew/[byteutils], json_serialization/std/sets, chronicles, stew/[byteutils], json_serialization/std/sets,
# Internal # Internal
@ -274,8 +274,6 @@ proc addResolved(pool: var AttestationPool, blck: BlockRef, attestation: Attesta
# sets by virtue of not overlapping with some other attestation # sets by virtue of not overlapping with some other attestation
# and therefore being useful after all? # and therefore being useful after all?
trace "Ignoring subset attestation", trace "Ignoring subset attestation",
existingParticipants = get_attesting_indices_seq(
state, a.data, v.aggregation_bits, cache),
newParticipants = participants, newParticipants = participants,
cat = "filtering" cat = "filtering"
found = true found = true
@ -286,10 +284,6 @@ proc addResolved(pool: var AttestationPool, blck: BlockRef, attestation: Attesta
# can now be removed per same logic as above # can now be removed per same logic as above
trace "Removing subset attestations", trace "Removing subset attestations",
existingParticipants = a.validations.filterIt(
it.aggregation_bits.isSubsetOf(validation.aggregation_bits)
).mapIt(get_attesting_indices_seq(
state, a.data, it.aggregation_bits, cache)),
newParticipants = participants, newParticipants = participants,
cat = "pruning" cat = "pruning"

View File

@ -417,7 +417,7 @@ proc onSlotStart(node: BeaconNode, lastSlot, scheduledSlot: Slot) {.gcsafe, asyn
epoch = scheduledSlot.compute_epoch_at_slot(), epoch = scheduledSlot.compute_epoch_at_slot(),
slot = scheduledSlot slot = scheduledSlot
# Brute-force, but ensure it's reliably enough to run in CI. # Brute-force, but ensure it's reliable enough to run in CI.
quit(0) quit(0)
if not wallSlot.afterGenesis or (wallSlot.slot < lastSlot): if not wallSlot.afterGenesis or (wallSlot.slot < lastSlot):

View File

@ -460,7 +460,6 @@ type
StateCache* = object StateCache* = object
shuffled_active_validator_indices*: shuffled_active_validator_indices*:
Table[Epoch, seq[ValidatorIndex]] Table[Epoch, seq[ValidatorIndex]]
committee_count_cache*: Table[Epoch, uint64]
beacon_proposer_indices*: Table[Slot, Option[ValidatorIndex]] beacon_proposer_indices*: Table[Slot, Option[ValidatorIndex]]
func shortValidatorKey*(state: BeaconState, validatorIdx: int): string = func shortValidatorKey*(state: BeaconState, validatorIdx: int): string =

View File

@ -22,7 +22,7 @@
{.push raises: [Defect].} {.push raises: [Defect].}
import import
math, sequtils, tables, math, sequtils, tables, algorithm,
stew/[bitops2], chronicles, json_serialization/std/sets, stew/[bitops2], chronicles, json_serialization/std/sets,
metrics, ../extras, ../ssz/merkleization, metrics, ../extras, ../ssz/merkleization,
beaconstate, crypto, datatypes, digest, helpers, validator, beaconstate, crypto, datatypes, digest, helpers, validator,
@ -330,41 +330,41 @@ func get_inclusion_delay_deltas(
state: BeaconState, total_balance: Gwei, cache: var StateCache): state: BeaconState, total_balance: Gwei, cache: var StateCache):
seq[Gwei] = seq[Gwei] =
# Return proposer and inclusion delay micro-rewards/penalties for each validator. # Return proposer and inclusion delay micro-rewards/penalties for each validator.
var rewards = repeat(0'u64, len(state.validators)) var
let rewards = repeat(0'u64, len(state.validators))
matching_source_attestations = matching_source_attestations =
get_matching_source_attestations(state, get_previous_epoch(state)) get_matching_source_attestations(state, get_previous_epoch(state))
source_attestation_attesting_indices = mapIt(
matching_source_attestations,
get_attesting_indices(state, it.data, it.aggregation_bits, cache))
for index in get_unslashed_attesting_indices(state, matching_source_attestations, cache): # Translation of attestation = min([...])
# Translation of attestation = min([...]) # The spec (pseudo)code defines this in terms of Python's min(), which per
# Start by filtering the right attestations # https://docs.python.org/3/library/functions.html#min:
var filtered_matching_source_attestations: seq[PendingAttestation] # If multiple items are minimal, the function returns the first one
# encountered.
# Therefore, this approach depends on Nim's default sort being stable, per
# https://nim-lang.org/docs/algorithm.html#sort,openArray[T],proc(T,T) via
# "The sorting is guaranteed to be stable and the worst case is guaranteed
# to be O(n log n)."
matching_source_attestations.sort do (x, y: PendingAttestation) -> int:
cmp(x.inclusion_delay, y.inclusion_delay)
for source_attestation_index, a in matching_source_attestations: # Order/indices in source_attestation_attesting_indices matches sorted order
if index notin let source_attestation_attesting_indices = mapIt(
matching_source_attestations,
get_attesting_indices(state, it.data, it.aggregation_bits, cache))
for index in get_unslashed_attesting_indices(
state, matching_source_attestations, cache):
for source_attestation_index, attestation in matching_source_attestations:
if index in
source_attestation_attesting_indices[source_attestation_index]: source_attestation_attesting_indices[source_attestation_index]:
continue rewards[attestation.proposer_index] +=
filtered_matching_source_attestations.add a get_proposer_reward(state, index, total_balance)
let max_attester_reward =
if filtered_matching_source_attestations.len == 0: get_base_reward(state, index, total_balance) -
continue get_proposer_reward(state, index, total_balance)
rewards[index] +=
# The first filtered attestation serves as min until we find something Gwei(max_attester_reward div attestation.inclusion_delay)
# better break
var attestation = filtered_matching_source_attestations[0]
for source_attestation_index, a in filtered_matching_source_attestations:
if a.inclusion_delay < attestation.inclusion_delay:
attestation = a
rewards[attestation.proposer_index] +=
get_proposer_reward(state, index, total_balance)
let max_attester_reward =
get_base_reward(state, index, total_balance) -
get_proposer_reward(state, index, total_balance)
rewards[index] += Gwei(max_attester_reward div attestation.inclusion_delay)
# No penalties associated with inclusion delay # No penalties associated with inclusion delay
# Spec constructs both and returns both; this doesn't # Spec constructs both and returns both; this doesn't

View File

@ -9,7 +9,7 @@
{.push raises: [Defect].} {.push raises: [Defect].}
import import
algorithm, options, sequtils, math, tables, algorithm, options, sequtils, math, tables, sets,
./datatypes, ./digest, ./helpers ./datatypes, ./digest, ./helpers
# https://github.com/ethereum/eth2.0-specs/blob/v0.12.1/specs/phase0/beacon-chain.md#compute_shuffled_index # https://github.com/ethereum/eth2.0-specs/blob/v0.12.1/specs/phase0/beacon-chain.md#compute_shuffled_index
@ -140,18 +140,15 @@ func get_beacon_committee*(
cache.shuffled_active_validator_indices[epoch] = cache.shuffled_active_validator_indices[epoch] =
get_shuffled_active_validator_indices(state, epoch) get_shuffled_active_validator_indices(state, epoch)
# Constant throughout an epoch
if epoch notin cache.committee_count_cache:
cache.committee_count_cache[epoch] =
get_committee_count_at_slot(state, slot)
try: try:
let committee_count = get_committee_count_at_slot(
cache.shuffled_active_validator_indices[epoch].len)
compute_committee( compute_committee(
cache.shuffled_active_validator_indices[epoch], cache.shuffled_active_validator_indices[epoch],
get_seed(state, epoch, DOMAIN_BEACON_ATTESTER), get_seed(state, epoch, DOMAIN_BEACON_ATTESTER),
(slot mod SLOTS_PER_EPOCH) * cache.committee_count_cache[epoch] + (slot mod SLOTS_PER_EPOCH) * committee_count +
index.uint64, index.uint64,
cache.committee_count_cache[epoch] * SLOTS_PER_EPOCH committee_count * SLOTS_PER_EPOCH
) )
except KeyError: except KeyError:
raiseAssert "values are added to cache before using them" raiseAssert "values are added to cache before using them"
@ -160,7 +157,6 @@ func get_beacon_committee*(
func get_empty_per_epoch_cache*(): StateCache = func get_empty_per_epoch_cache*(): StateCache =
result.shuffled_active_validator_indices = result.shuffled_active_validator_indices =
initTable[Epoch, seq[ValidatorIndex]]() initTable[Epoch, seq[ValidatorIndex]]()
result.committee_count_cache = initTable[Epoch, uint64]()
# https://github.com/ethereum/eth2.0-specs/blob/v0.12.1/specs/phase0/beacon-chain.md#compute_proposer_index # https://github.com/ethereum/eth2.0-specs/blob/v0.12.1/specs/phase0/beacon-chain.md#compute_proposer_index
func compute_proposer_index(state: BeaconState, indices: seq[ValidatorIndex], func compute_proposer_index(state: BeaconState, indices: seq[ValidatorIndex],
@ -234,7 +230,8 @@ func get_beacon_proposer_index*(state: BeaconState, cache: var StateCache):
# https://github.com/ethereum/eth2.0-specs/blob/v0.12.1/specs/phase0/validator.md#validator-assignments # https://github.com/ethereum/eth2.0-specs/blob/v0.12.1/specs/phase0/validator.md#validator-assignments
func get_committee_assignment*( func get_committee_assignment*(
state: BeaconState, epoch: Epoch, validator_index: ValidatorIndex): state: BeaconState, epoch: Epoch,
validator_indices: HashSet[ValidatorIndex]):
Option[tuple[a: seq[ValidatorIndex], b: CommitteeIndex, c: Slot]] {.used.} = Option[tuple[a: seq[ValidatorIndex], b: CommitteeIndex, c: Slot]] {.used.} =
# Return the committee assignment in the ``epoch`` for ``validator_index``. # Return the committee assignment in the ``epoch`` for ``validator_index``.
# ``assignment`` returned is a tuple of the following form: # ``assignment`` returned is a tuple of the following form:
@ -242,6 +239,9 @@ func get_committee_assignment*(
# * ``assignment[1]`` is the index to which the committee is assigned # * ``assignment[1]`` is the index to which the committee is assigned
# * ``assignment[2]`` is the slot at which the committee is assigned # * ``assignment[2]`` is the slot at which the committee is assigned
# Return None if no assignment. # Return None if no assignment.
#
# Slightly adapted from spec version to support multiple validator indices,
# since each beacon_node supports many validators.
let next_epoch = get_current_epoch(state) + 1 let next_epoch = get_current_epoch(state) + 1
doAssert epoch <= next_epoch doAssert epoch <= next_epoch
@ -251,9 +251,8 @@ func get_committee_assignment*(
for slot in start_slot ..< start_slot + SLOTS_PER_EPOCH: for slot in start_slot ..< start_slot + SLOTS_PER_EPOCH:
for index in 0 ..< get_committee_count_at_slot(state, slot): for index in 0 ..< get_committee_count_at_slot(state, slot):
let idx = index.CommitteeIndex let idx = index.CommitteeIndex
let committee = let committee = get_beacon_committee(state, slot, idx, cache)
get_beacon_committee(state, slot, idx, cache) if not disjoint(validator_indices, toHashSet(committee)):
if validator_index in committee:
return some((committee, idx, slot)) return some((committee, idx, slot))
none(tuple[a: seq[ValidatorIndex], b: CommitteeIndex, c: Slot]) none(tuple[a: seq[ValidatorIndex], b: CommitteeIndex, c: Slot])

View File

@ -7,7 +7,7 @@
import import
# Standard library # Standard library
tables, strutils, parseutils, sequtils, tables, strutils, parseutils, sequtils, sets,
# Nimble packages # Nimble packages
stew/[byteutils, objects], stew/[byteutils, objects],
@ -356,7 +356,8 @@ proc installValidatorApiHandlers*(rpcServer: RpcServer, node: BeaconNode) =
let idx = state.validators.asSeq.findIt(it.pubKey == pubkey) let idx = state.validators.asSeq.findIt(it.pubKey == pubkey)
if idx == -1: if idx == -1:
continue continue
let ca = state.get_committee_assignment(epoch, idx.ValidatorIndex) let ca = state.get_committee_assignment(
epoch, toHashSet([idx.ValidatorIndex]))
if ca.isSome: if ca.isSome:
result.add((public_key: pubkey, result.add((public_key: pubkey,
committee_index: ca.get.b, committee_index: ca.get.b,