Bulk signature verification (#507)

* Add basic block processing benches

* Start reviving state processing benches

* Fix old block builders

* Add optimization for faster pubkey add

* Tidy benches, add another

* Add extra block processing bench

* Start working on faster BLS scheme

* Add partially complete sig verify optimization

* Add .gitignore to state processing

* Add progress on faster signature verification

* Fix SignatureSet for fake_crypto

* Tidy attester slashings sig set

* Tidy bulk signature verifier

* Refactor signature sets to be cleaner

* Start threading SignatureStrategy through code

* Add (empty) test dir

* Move BenchingBlockBuilder

* Add initial block signature verification tests

* Add tests for bulk signature verification

* Start threading SignatureStrategy in block proc.

* Refactor per_block_processing errors

* Use sig set tuples instead of lists of two

* Remove dead code

* Thread VerifySignatures through per_block_processing

* Add bulk signature verification

* Introduce parallel bulk signature verification

* Expand state processing benches

* Fix additional compile errors

* Fix issue where par iter chunks is 0

* Update milagro_bls dep

* Remove debugs, code fragment in beacon chain

* Tidy, add comments to block sig verifier

* Fix various PR comments

* Add block_root option to per_block_processing

* Fix comment in block signature verifier

* Fix comments from PR review

* Remove old comment

* Fix comment

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -19,8 +19,7 @@ use state_processing::per_block_processing::{
     verify_attestation_for_state, VerifySignatures,
 };
 use state_processing::{
-    per_block_processing, per_block_processing_without_verifying_block_signature,
-    per_slot_processing, BlockProcessingError,
+    per_block_processing, per_slot_processing, BlockProcessingError, BlockSignatureStrategy,
 };
 use std::sync::Arc;
 use store::iter::{BlockRootsIterator, StateRootsIterator};
@@ -726,7 +725,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 finalized: finalized_epoch,
             })
         } else if let Err(e) =
-            verify_attestation_for_state(state, &attestation, &self.spec, VerifySignatures::True)
+            verify_attestation_for_state(state, &attestation, VerifySignatures::True, &self.spec)
         {
             warn!(
                 self.log,
@@ -896,7 +895,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         // Apply the received block to its parent state (which has been transitioned into this
         // slot).
-        match per_block_processing(&mut state, &block, &self.spec) {
+        match per_block_processing(
+            &mut state,
+            &block,
+            Some(block_root),
+            BlockSignatureStrategy::VerifyIndividual,
+            &self.spec,
+        ) {
             Err(BlockProcessingError::BeaconStateError(e)) => {
                 return Err(Error::BeaconStateError(e))
             }
@@ -1060,7 +1065,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             },
         };
 
-        per_block_processing_without_verifying_block_signature(&mut state, &block, &self.spec)?;
+        per_block_processing(
+            &mut state,
+            &block,
+            None,
+            BlockSignatureStrategy::NoVerification,
+            &self.spec,
+        )?;
 
         let state_root = state.canonical_root();
 

beacon_node/beacon_chain/src/errors.rs

@@ -1,7 +1,5 @@
 use crate::fork_choice::Error as ForkChoiceError;
-use state_processing::per_block_processing::errors::{
-    AttestationValidationError, IndexedAttestationValidationError,
-};
+use state_processing::per_block_processing::errors::AttestationValidationError;
 use state_processing::BlockProcessingError;
 use state_processing::SlotProcessingError;
 use types::*;
@@ -37,7 +35,6 @@ pub enum BeaconChainError {
         beacon_block_root: Hash256,
     },
     AttestationValidationError(AttestationValidationError),
-    IndexedAttestationValidationError(IndexedAttestationValidationError),
 }
 
 easy_from_to!(SlotProcessingError, BeaconChainError);
@@ -55,4 +52,3 @@ easy_from_to!(BlockProcessingError, BlockProductionError);
 easy_from_to!(BeaconStateError, BlockProductionError);
 easy_from_to!(SlotProcessingError, BlockProductionError);
 easy_from_to!(AttestationValidationError, BeaconChainError);
-easy_from_to!(IndexedAttestationValidationError, BeaconChainError);

beacon_node/beacon_chain/src/test_utils.rs

@@ -1,5 +1,6 @@
 use crate::{BeaconChain, BeaconChainTypes, BlockProcessingOutcome};
 use lmd_ghost::LmdGhost;
+use rayon::prelude::*;
 use sloggers::{null::NullLoggerBuilder, Build};
 use slot_clock::SlotClock;
 use slot_clock::TestingSlotClock;
@@ -164,7 +165,9 @@ where
         let mut state = {
             // Determine the slot for the first block (or skipped block).
             let state_slot = match block_strategy {
-                BlockStrategy::OnCanonicalHead => self.chain.read_slot_clock().unwrap() - 1,
+                BlockStrategy::OnCanonicalHead => {
+                    self.chain.read_slot_clock().expect("should know slot") - 1
+                }
                 BlockStrategy::ForkCanonicalChainAt { previous_slot, .. } => previous_slot,
             };
 
@@ -173,7 +176,9 @@ where
 
         // Determine the first slot where a block should be built.
         let mut slot = match block_strategy {
-            BlockStrategy::OnCanonicalHead => self.chain.read_slot_clock().unwrap(),
+            BlockStrategy::OnCanonicalHead => {
+                self.chain.read_slot_clock().expect("should know slot")
+            }
             BlockStrategy::ForkCanonicalChainAt { first_slot, .. } => first_slot,
         };
 
@@ -237,7 +242,9 @@ where
                 .expect("should be able to advance state to slot");
         }
 
-        state.build_all_caches(&self.spec).unwrap();
+        state
+            .build_all_caches(&self.spec)
+            .expect("should build caches");
 
         let proposer_index = match block_strategy {
             BlockStrategy::OnCanonicalHead => self
@@ -314,7 +321,7 @@ where
             AttestationStrategy::SomeValidators(vec) => vec.clone(),
         };
 
-        let mut vec = vec![];
+        let mut attestations = vec![];
 
         state
             .get_crosslink_committees_at_slot(state.slot)
@@ -323,55 +330,70 @@ where
             .for_each(|cc| {
                 let committee_size = cc.committee.len();
 
-                for (i, validator_index) in cc.committee.iter().enumerate() {
-                    // Note: searching this array is worst-case `O(n)`. A hashset could be a better
-                    // alternative.
-                    if attesting_validators.contains(validator_index) {
-                        let data = self
-                            .chain
-                            .produce_attestation_data_for_block(
-                                cc.shard,
-                                head_block_root,
-                                head_block_slot,
-                                state,
-                            )
-                            .expect("should produce attestation data");
+                let mut local_attestations: Vec<Attestation<E>> = cc
+                    .committee
+                    .par_iter()
+                    .enumerate()
+                    .filter_map(|(i, validator_index)| {
+                        // Note: searching this array is worst-case `O(n)`. A hashset could be a better
+                        // alternative.
+                        if attesting_validators.contains(validator_index) {
+                            let data = self
+                                .chain
+                                .produce_attestation_data_for_block(
+                                    cc.shard,
+                                    head_block_root,
+                                    head_block_slot,
+                                    state,
+                                )
+                                .expect("should produce attestation data");
 
-                        let mut aggregation_bits = BitList::with_capacity(committee_size).unwrap();
-                        aggregation_bits.set(i, true).unwrap();
-                        let custody_bits = BitList::with_capacity(committee_size).unwrap();
+                            let mut aggregation_bits = BitList::with_capacity(committee_size)
+                                .expect("should make aggregation bits");
+                            aggregation_bits
+                                .set(i, true)
+                                .expect("should be able to set aggregation bits");
+                            let custody_bits = BitList::with_capacity(committee_size)
+                                .expect("should make custody bits");
 
-                        let signature = {
-                            let message = AttestationDataAndCustodyBit {
-                                data: data.clone(),
-                                custody_bit: false,
-                            }
-                            .tree_hash_root();
+                            let signature = {
+                                let message = AttestationDataAndCustodyBit {
+                                    data: data.clone(),
+                                    custody_bit: false,
+                                }
+                                .tree_hash_root();
 
-                            let domain =
-                                spec.get_domain(data.target.epoch, Domain::Attestation, fork);
+                                let domain =
+                                    spec.get_domain(data.target.epoch, Domain::Attestation, fork);
 
-                            let mut agg_sig = AggregateSignature::new();
-                            agg_sig.add(&Signature::new(
-                                &message,
-                                domain,
-                                self.get_sk(*validator_index),
-                            ));
+                                let mut agg_sig = AggregateSignature::new();
+                                agg_sig.add(&Signature::new(
+                                    &message,
+                                    domain,
+                                    self.get_sk(*validator_index),
+                                ));
 
-                            agg_sig
-                        };
+                                agg_sig
+                            };
 
-                        vec.push(Attestation {
-                            aggregation_bits,
-                            data,
-                            custody_bits,
-                            signature,
-                        })
-                    }
-                }
+                            let attestation = Attestation {
+                                aggregation_bits,
+                                data,
+                                custody_bits,
+                                signature,
+                            };
+
+                            Some(attestation)
+                        } else {
+                            None
+                        }
+                    })
+                    .collect();
+
+                attestations.append(&mut local_attestations);
             });
 
-        vec
+        attestations
     }
 
     /// Creates two forks: