Bloom filter acceleration for deposit processing

beacon_chain/bloomfilter.nim

@@ -0,0 +1,46 @@
+# beacon_chain
+# Copyright (c) 2024 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+{.push raises: [].}
+
+import "."/spec/crypto
+
+from "."/spec/datatypes/base import Validator, pubkey
+from "."/spec/helpers import bytes_to_uint32
+
+const
+  # https://hur.st/bloomfilter/?n=4M&p=&m=8MiB&k=
+  pubkeyBloomFilterScale = 23   # 21 too small, 22 borderline, 24 also ok
+
+type
+  PubkeyBloomFilter* = object
+    bloomFilter: array[1 shl pubkeyBloomFilterScale, byte]
+
+from stew/bitops2 import getBit, setBit
+
+iterator bloomFilterHashes(pubkey: ValidatorPubKey): auto =
+  const pubkeyBloomFilterMask = (1 shl pubkeyBloomFilterScale) - 1
+  for r in countup(0'u32, 20'u32, 4'u32):
+    # ValidatorPubKeys have fairly uniform entropy; using enough hash
+    # functions also reduces risk of low-entropy portions
+    yield pubkey.blob.toOpenArray(r, r+3).bytes_to_uint32 and
+      pubkeyBloomFilterMask
+
+func constructBloomFilter*(x: openArray[Validator]): auto =
+  let res = new PubkeyBloomFilter
+  for m in x:
+    for bloomFilterHash in bloomFilterHashes(m.pubkey):
+      setBit(res[].bloomFilter, bloomFilterHash)
+  res
+
+func mightContain*(bf: PubkeyBloomFilter, pubkey: ValidatorPubKey): bool =
+  # Might return false positive, but never false negative
+  for bloomFilterHash in bloomFilterHashes(pubkey):
+    if not getBit(bf.bloomFilter, bloomFilterHash):
+      return false
+
+  true

beacon_chain/spec/state_transition_block.nim

@@ -287,10 +287,13 @@ func findValidatorIndex*(state: ForkyBeaconState, pubkey: ValidatorPubKey):
     if state.validators.asSeq[vidx - 1].pubkey == pubkey:
       return Opt[ValidatorIndex].ok((vidx - 1).ValidatorIndex)
 
-proc process_deposit*(cfg: RuntimeConfig,
-                      state: var ForkyBeaconState,
-                      deposit: Deposit,
-                      flags: UpdateFlags): Result[void, cstring] =
+from ".."/bloomfilter import
+  PubkeyBloomFilter, constructBloomFilter, mightContain
+
+proc process_deposit*(
+    cfg: RuntimeConfig, state: var ForkyBeaconState,
+    bloom_filter: var PubkeyBloomFilter, deposit: Deposit, flags: UpdateFlags):
+    Result[void, cstring] =
   ## Process an Eth1 deposit, registering a validator or increasing its balance.
 
   # Verify the Merkle branch
@@ -309,7 +312,11 @@ proc process_deposit*(cfg: RuntimeConfig,
   let
     pubkey = deposit.data.pubkey
     amount = deposit.data.amount
-    index = findValidatorIndex(state, pubkey)
+    index =
+      if bloom_filter.mightContain(pubkey):
+        findValidatorIndex(state, pubkey)
+      else:
+        Opt.none(ValidatorIndex)
 
   if index.isSome():
     # Increase balance by deposit amount
@@ -452,8 +459,10 @@ proc process_operations(cfg: RuntimeConfig,
     ? process_attester_slashing(cfg, state, op, flags, cache)
   for op in body.attestations:
     ? process_attestation(state, op, flags, base_reward_per_increment, cache)
-  for op in body.deposits:
-    ? process_deposit(cfg, state, op, flags)
+  if body.deposits.len > 0:
+    let bloom_filter = constructBloomFilter(state.validators.asSeq)
+    for op in body.deposits:
+      ? process_deposit(cfg, state, bloom_filter[], op, flags)
   for op in body.voluntary_exits:
     ? process_voluntary_exit(cfg, state, op, flags, cache)
   when typeof(body).kind >= ConsensusFork.Capella:

nfuzz/libnfuzz.nim

@@ -8,6 +8,8 @@
 # Required for deserialisation of ValidatorSig in Attestation due to
 # https://github.com/nim-lang/Nim/issues/11225
 
+{.push raises: [].}
+
 import
   stew/ptrops, chronicles,
   ../beacon_chain/networking/network_metadata,
@@ -41,7 +43,7 @@ type
   FuzzCrashError = object of CatchableError
 
 # TODO: change ptr uint to ptr csize_t when available in newer Nim version.
-proc copyState(state: phase0.BeaconState, xoutput: ptr byte,
+func copyState(state: phase0.BeaconState, xoutput: ptr byte,
     xoutput_size: ptr uint): bool {.raises: [FuzzCrashError].} =
   var resultState =
     try:
@@ -132,15 +134,20 @@ proc nfuzz_block(input: openArray[byte], xoutput: ptr byte,
     state_transition(
       getRuntimeConfig(some "mainnet"), data, data.beaconBlock, flags, noRollback).isOk
 
-proc nfuzz_block_header(input: openArray[byte], xoutput: ptr byte,
+func nfuzz_block_header(input: openArray[byte], xoutput: ptr byte,
     xoutput_size: ptr uint, disable_bls: bool): bool {.exportc, raises: [FuzzCrashError].} =
   decodeAndProcess(BlockHeaderInput):
     process_block_header(data.state, data.beaconBlock.message, flags, cache).isOk
 
+from ".."/beacon_chain/bloomfilter import constructBloomFilter
+
 proc nfuzz_deposit(input: openArray[byte], xoutput: ptr byte,
     xoutput_size: ptr uint, disable_bls: bool): bool {.exportc, raises: [FuzzCrashError].} =
   decodeAndProcess(DepositInput):
-    process_deposit(getRuntimeConfig(some "mainnet"), data.state, data.deposit, flags).isOk
+    process_deposit(
+      getRuntimeConfig(some "mainnet"), data.state,
+      constructBloomFilter(data.state.validators.asSeq)[], data.deposit,
+      flags).isOk
 
 proc nfuzz_proposer_slashing(input: openArray[byte], xoutput: ptr byte,
     xoutput_size: ptr uint, disable_bls: bool): bool {.exportc, raises: [FuzzCrashError].} =
@@ -156,7 +163,7 @@ proc nfuzz_voluntary_exit(input: openArray[byte], xoutput: ptr byte,
 # However, list_size needs to be known also outside this proc to allocate xoutput.
 # TODO: rework to copy immediatly in an uint8 openArray, considering we have to
 # go over the list anyhow?
-proc nfuzz_shuffle(input_seed: ptr byte, xoutput: var openArray[uint64]): bool
+func nfuzz_shuffle(input_seed: ptr byte, xoutput: var openArray[uint64]): bool
     {.exportc, raises: [].} =
   var seed: Eth2Digest
   # Should be OK as max 2 bytes are passed by the framework.

tests/consensus_spec/altair/test_fixture_operations.nim

@@ -110,11 +110,15 @@ suite baseDescription & "Block Header " & preset():
     runTest[altair.BeaconBlock, typeof applyBlockHeader](
       OpBlockHeaderDir, suiteName, "Block Header", "block", applyBlockHeader, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var altair.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/consensus_spec/bellatrix/test_fixture_operations.nim

@@ -116,11 +116,15 @@ suite baseDescription & "Block Header " & preset():
       OpBlockHeaderDir, suiteName, "Block Header", "block",
       applyBlockHeader, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var bellatrix.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/consensus_spec/capella/test_fixture_operations.nim

@@ -133,11 +133,15 @@ suite baseDescription & "BLS to execution change " & preset():
       OpBlsToExecutionChangeDir, suiteName, "BLS to execution change", "address_change",
       applyBlsToExecutionChange, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var capella.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/consensus_spec/deneb/test_fixture_operations.nim

@@ -136,11 +136,15 @@ suite baseDescription & "BLS to execution change " & preset():
       OpBlsToExecutionChangeDir, suiteName, "BLS to execution change", "address_change",
       applyBlsToExecutionChange, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var deneb.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/consensus_spec/phase0/test_fixture_operations.nim

@@ -103,11 +103,15 @@ suite baseDescription & "Block Header " & preset():
       OpBlockHeaderDir, suiteName, "Block Header", "block",
       applyBlockHeader, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var phase0.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/teststateutil.nim

@@ -10,9 +10,11 @@
 import
   chronicles,
   ./mocking/mock_deposits,
-  ./helpers/math_helpers,
   ../beacon_chain/spec/[
-    forks, helpers, state_transition, state_transition_block]
+    forks, state_transition, state_transition_block]
+
+from "."/helpers/math_helpers import round_multiple_down
+from ".."/beacon_chain/bloomfilter import constructBloomFilter
 
 proc valid_deposit(state: var ForkyHashedBeaconState) =
   const deposit_amount = MAX_EFFECTIVE_BALANCE
@@ -29,7 +31,9 @@ proc valid_deposit(state: var ForkyHashedBeaconState) =
                       state.data.balances.item(validator_index)
                     else:
                       0
-  doAssert process_deposit(defaultRuntimeConfig, state.data, deposit, {}).isOk
+  doAssert process_deposit(
+    defaultRuntimeConfig, state.data,
+    constructBloomFilter(state.data.validators.asSeq)[], deposit, {}).isOk
   doAssert state.data.validators.len == pre_val_count + 1
   doAssert state.data.balances.len == pre_val_count + 1
   doAssert state.data.balances.item(validator_index) == pre_balance + deposit.data.amount