Add per-epoch benchmarks, optimise function.

Cargo.toml

@@ -4,6 +4,7 @@ members = [
 	"eth2/block_proposer",
 	"eth2/fork_choice",
 	"eth2/state_processing",
+	"eth2/state_processing/benching_utils",
 	"eth2/types",
 	"eth2/utils/bls",
 	"eth2/utils/boolean-bitfield",

eth2/state_processing/Cargo.toml

@@ -11,9 +11,11 @@ harness = false
 [dev-dependencies]
 criterion = "0.2"
 env_logger = "0.6.0"
+benching_utils = { path = "./benching_utils" }
 
 [dependencies]
 bls = { path = "../utils/bls" }
+fnv = "1.0"
 hashing = { path = "../utils/hashing" }
 int_to_bytes = { path = "../utils/int_to_bytes" }
 integer-sqrt = "0.1"

eth2/state_processing/benches/benches.rs

@@ -1,60 +1,291 @@
 use criterion::Criterion;
 use criterion::{black_box, criterion_group, criterion_main, Benchmark};
+use state_processing::{
+    per_epoch_processing,
+    per_epoch_processing::{
+        calculate_active_validator_indices, calculate_attester_sets, clean_attestations,
+        process_crosslinks, process_eth1_data, process_justification,
+        process_rewards_and_penalities, process_validator_registry, update_active_tree_index_roots,
+        update_latest_slashed_balances,
+    },
+};
 // use env_logger::{Builder, Env};
-use state_processing::SlotProcessable;
+use benching_utils::BeaconStateBencher;
-use types::beacon_state::BeaconStateBuilder;
+use types::{validator_registry::get_active_validator_indices, *};
-use types::*;
 
 fn epoch_processing(c: &mut Criterion) {
     // Builder::from_env(Env::default().default_filter_or("debug")).init();
+    //
+    let spec = ChainSpec::foundation();
 
-    let mut builder = BeaconStateBuilder::new(16_384);
+    let validator_count = 16_384;
 
-    builder.build_fast().unwrap();
+    let mut builder = BeaconStateBencher::new(validator_count, &spec);
-    builder.teleport_to_end_of_epoch(builder.spec.genesis_epoch + 4);
+    builder.teleport_to_end_of_epoch(spec.genesis_epoch + 4, &spec);
-
+    builder.insert_attestations(&spec);
-    let mut state = builder.cloned_state();
+    let mut state = builder.build();
 
     // Build all the caches so the following state does _not_ include the cache-building time.
     state
-        .build_epoch_cache(RelativeEpoch::Previous, &builder.spec)
+        .build_epoch_cache(RelativeEpoch::Previous, &spec)
         .unwrap();
     state
-        .build_epoch_cache(RelativeEpoch::Current, &builder.spec)
+        .build_epoch_cache(RelativeEpoch::Current, &spec)
-        .unwrap();
-    state
-        .build_epoch_cache(RelativeEpoch::Next, &builder.spec)
         .unwrap();
+    state.build_epoch_cache(RelativeEpoch::Next, &spec).unwrap();
 
-    let cached_state = state.clone();
+    // Assert that the state has the maximum possible attestations.
+    let committees_per_epoch = spec.get_epoch_committee_count(validator_count);
+    let committees_per_slot = committees_per_epoch / spec.slots_per_epoch;
+    let previous_epoch_attestations = committees_per_epoch;
+    let current_epoch_attestations =
+        committees_per_slot * (spec.slots_per_epoch - spec.min_attestation_inclusion_delay);
+    assert_eq!(
+        state.latest_attestations.len() as u64,
+        previous_epoch_attestations + current_epoch_attestations
+    );
 
-    // Drop all the caches so the following state includes the cache-building time.
+    // Assert that each attestation in the state has full participation.
-    state.drop_cache(RelativeEpoch::Previous);
+    let committee_size = validator_count / committees_per_epoch as usize;
-    state.drop_cache(RelativeEpoch::Current);
+    for a in &state.latest_attestations {
-    state.drop_cache(RelativeEpoch::Next);
+        assert_eq!(a.aggregation_bitfield.num_set_bits(), committee_size);
+    }
 
-    let cacheless_state = state;
+    // Assert that we will run the first arm of process_rewards_and_penalities
+    let epochs_since_finality = state.next_epoch(&spec) - state.finalized_epoch;
+    assert!(epochs_since_finality <= 4);
 
-    let spec_a = builder.spec.clone();
+    bench_epoch_processing(c, &state, &spec, "16k_validators");
-    let spec_b = builder.spec.clone();
+}
 
+fn bench_epoch_processing(c: &mut Criterion, state: &BeaconState, spec: &ChainSpec, desc: &str) {
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
     c.bench(
-        "epoch processing",
+        &format!("epoch_process_with_caches_{}", desc),
-        Benchmark::new("with pre-built caches", move |b| {
+        Benchmark::new("full run", move |b| {
             b.iter_with_setup(
-                || cached_state.clone(),
+                || state_clone.clone(),
-                |mut state| black_box(state.per_slot_processing(Hash256::zero(), &spec_a).unwrap()),
+                |mut state| black_box(per_epoch_processing(&mut state, &spec_clone).unwrap()),
             )
         })
         .sample_size(10),
     );
 
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
     c.bench(
-        "epoch processing",
+        &format!("epoch_process_with_caches_{}", desc),
-        Benchmark::new("without pre-built caches", move |b| {
+        Benchmark::new("calculate_active_validator_indices", move |b| {
             b.iter_with_setup(
-                || cacheless_state.clone(),
+                || state_clone.clone(),
-                |mut state| black_box(state.per_slot_processing(Hash256::zero(), &spec_b).unwrap()),
+                |mut state| black_box(calculate_active_validator_indices(&mut state, &spec_clone)),
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    let active_validator_indices = calculate_active_validator_indices(&state, &spec);
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("calculate_current_total_balance", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |state| {
+                    black_box(state.get_total_balance(&active_validator_indices[..], &spec_clone))
+                },
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("calculate_previous_total_balance", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |state| {
+                    black_box(state.get_total_balance(
+                        &get_active_validator_indices(
+                            &state.validator_registry,
+                            state.previous_epoch(&spec_clone),
+                        )[..],
+                        &spec_clone,
+                    ))
+                },
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("process_eth1_data", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| black_box(process_eth1_data(&mut state, &spec_clone)),
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("calculate_attester_sets", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| black_box(calculate_attester_sets(&mut state, &spec_clone).unwrap()),
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    let previous_epoch = state.previous_epoch(&spec);
+    let attesters = calculate_attester_sets(&state, &spec).unwrap();
+    let active_validator_indices = calculate_active_validator_indices(&state, &spec);
+    let current_total_balance = state.get_total_balance(&active_validator_indices[..], &spec);
+    let previous_total_balance = state.get_total_balance(
+        &get_active_validator_indices(&state.validator_registry, previous_epoch)[..],
+        &spec,
+    );
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("process_justification", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| {
+                    black_box(process_justification(
+                        &mut state,
+                        current_total_balance,
+                        previous_total_balance,
+                        attesters.previous_epoch_boundary.balance,
+                        attesters.current_epoch_boundary.balance,
+                        &spec_clone,
+                    ))
+                },
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("process_crosslinks", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| black_box(process_crosslinks(&mut state, &spec_clone).unwrap()),
+            )
+        })
+        .sample_size(10),
+    );
+
+    let mut state_clone = state.clone();
+    let spec_clone = spec.clone();
+    let previous_epoch = state.previous_epoch(&spec);
+    let attesters = calculate_attester_sets(&state, &spec).unwrap();
+    let active_validator_indices = calculate_active_validator_indices(&state, &spec);
+    let previous_total_balance = state.get_total_balance(
+        &get_active_validator_indices(&state.validator_registry, previous_epoch)[..],
+        &spec,
+    );
+    let winning_root_for_shards = process_crosslinks(&mut state_clone, &spec).unwrap();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("process_rewards_and_penalties", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| {
+                    black_box(
+                        process_rewards_and_penalities(
+                            &mut state,
+                            &active_validator_indices,
+                            &attesters,
+                            previous_total_balance,
+                            &winning_root_for_shards,
+                            &spec_clone,
+                        )
+                        .unwrap(),
+                    )
+                },
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("process_ejections", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| black_box(state.process_ejections(&spec_clone)),
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("process_validator_registry", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| black_box(process_validator_registry(&mut state, &spec_clone)),
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("update_active_tree_index_roots", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| {
+                    black_box(update_active_tree_index_roots(&mut state, &spec_clone).unwrap())
+                },
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("update_latest_slashed_balances", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| black_box(update_latest_slashed_balances(&mut state, &spec_clone)),
+            )
+        })
+        .sample_size(10),
+    );
+
+    let state_clone = state.clone();
+    let spec_clone = spec.clone();
+    c.bench(
+        &format!("epoch_process_with_caches_{}", desc),
+        Benchmark::new("clean_attestations", move |b| {
+            b.iter_with_setup(
+                || state_clone.clone(),
+                |mut state| black_box(clean_attestations(&mut state, &spec_clone)),
             )
         })
         .sample_size(10),

eth2/state_processing/benching_utils/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "benching_utils"
+version = "0.1.0"
+authors = ["Paul Hauner <paul@paulhauner.com>"]
+edition = "2018"
+
+[dependencies]
+bls = { path = "../../utils/bls" }
+hashing = { path = "../../utils/hashing" }
+int_to_bytes = { path = "../../utils/int_to_bytes" }
+integer-sqrt = "0.1"
+log = "0.4"
+merkle_proof = { path = "../../utils/merkle_proof" }
+ssz = { path = "../../utils/ssz" }
+ssz_derive = { path = "../../utils/ssz_derive" }
+types = { path = "../../types" }
+rayon = "1.0"

eth2/state_processing/benching_utils/src/lib.rs

@@ -0,0 +1,195 @@
+use bls::get_withdrawal_credentials;
+use int_to_bytes::int_to_bytes48;
+use rayon::prelude::*;
+use types::beacon_state::BeaconStateBuilder;
+use types::*;
+
+pub struct BeaconStateBencher {
+    state: BeaconState,
+}
+
+impl BeaconStateBencher {
+    pub fn new(validator_count: usize, spec: &ChainSpec) -> Self {
+        let keypairs: Vec<Keypair> = (0..validator_count)
+            .collect::<Vec<usize>>()
+            .par_iter()
+            .map(|&i| {
+                let secret = int_to_bytes48(i as u64 + 1);
+                let sk = SecretKey::from_bytes(&secret).unwrap();
+                let pk = PublicKey::from_secret_key(&sk);
+                Keypair { sk, pk }
+            })
+            .collect();
+
+        let validators = keypairs
+            .iter()
+            .map(|keypair| {
+                let withdrawal_credentials = Hash256::from_slice(&get_withdrawal_credentials(
+                    &keypair.pk,
+                    spec.bls_withdrawal_prefix_byte,
+                ));
+
+                Validator {
+                    pubkey: keypair.pk.clone(),
+                    withdrawal_credentials,
+                    activation_epoch: spec.far_future_epoch,
+                    exit_epoch: spec.far_future_epoch,
+                    withdrawable_epoch: spec.far_future_epoch,
+                    initiated_exit: false,
+                    slashed: false,
+                }
+            })
+            .collect();
+
+        let mut state_builder = BeaconStateBuilder::new(
+            0,
+            Eth1Data {
+                deposit_root: Hash256::zero(),
+                block_hash: Hash256::zero(),
+            },
+            spec,
+        );
+
+        let balances = vec![32_000_000_000; validator_count];
+
+        state_builder.import_existing_validators(
+            validators,
+            balances,
+            validator_count as u64,
+            spec,
+        );
+
+        Self {
+            state: state_builder.build(spec).unwrap(),
+        }
+    }
+
+    pub fn build(self) -> BeaconState {
+        self.state
+    }
+
+    /// Sets the `BeaconState` to be in the last slot of the given epoch.
+    ///
+    /// Sets all justification/finalization parameters to be be as "perfect" as possible (i.e.,
+    /// highest justified and finalized slots, full justification bitfield, etc).
+    pub fn teleport_to_end_of_epoch(&mut self, epoch: Epoch, spec: &ChainSpec) {
+        let state = &mut self.state;
+
+        let slot = epoch.end_slot(spec.slots_per_epoch);
+
+        state.slot = slot;
+        state.validator_registry_update_epoch = epoch - 1;
+
+        state.previous_shuffling_epoch = epoch - 1;
+        state.current_shuffling_epoch = epoch;
+
+        state.previous_shuffling_seed = Hash256::from_low_u64_le(0);
+        state.current_shuffling_seed = Hash256::from_low_u64_le(1);
+
+        state.previous_justified_epoch = epoch - 2;
+        state.justified_epoch = epoch - 1;
+        state.justification_bitfield = u64::max_value();
+        state.finalized_epoch = epoch - 1;
+    }
+
+    /// Creates a full set of attestations for the `BeaconState`. Each attestation has full
+    /// participation from its committee and references the expected beacon_block hashes.
+    ///
+    /// These attestations should be fully conducive to justification and finalization.
+    pub fn insert_attestations(&mut self, spec: &ChainSpec) {
+        let state = &mut self.state;
+
+        state
+            .build_epoch_cache(RelativeEpoch::Previous, spec)
+            .unwrap();
+        state
+            .build_epoch_cache(RelativeEpoch::Current, spec)
+            .unwrap();
+
+        let current_epoch = state.current_epoch(spec);
+        let previous_epoch = state.previous_epoch(spec);
+
+        let first_slot = previous_epoch.start_slot(spec.slots_per_epoch).as_u64();
+        let last_slot = current_epoch.end_slot(spec.slots_per_epoch).as_u64()
+            - spec.min_attestation_inclusion_delay;
+        let last_slot = std::cmp::min(state.slot.as_u64(), last_slot);
+
+        for slot in first_slot..last_slot + 1 {
+            let slot = Slot::from(slot);
+
+            let committees = state
+                .get_crosslink_committees_at_slot(slot, spec)
+                .unwrap()
+                .clone();
+
+            for (committee, shard) in committees {
+                state
+                    .latest_attestations
+                    .push(committee_to_pending_attestation(
+                        state, &committee, shard, slot, spec,
+                    ))
+            }
+        }
+    }
+}
+
+fn committee_to_pending_attestation(
+    state: &BeaconState,
+    committee: &[usize],
+    shard: u64,
+    slot: Slot,
+    spec: &ChainSpec,
+) -> PendingAttestation {
+    let current_epoch = state.current_epoch(spec);
+    let previous_epoch = state.previous_epoch(spec);
+
+    let mut aggregation_bitfield = Bitfield::new();
+    let mut custody_bitfield = Bitfield::new();
+
+    for (i, _) in committee.iter().enumerate() {
+        aggregation_bitfield.set(i, true);
+        custody_bitfield.set(i, true);
+    }
+
+    let is_previous_epoch =
+        state.slot.epoch(spec.slots_per_epoch) != slot.epoch(spec.slots_per_epoch);
+
+    let justified_epoch = if is_previous_epoch {
+        state.previous_justified_epoch
+    } else {
+        state.justified_epoch
+    };
+
+    let epoch_boundary_root = if is_previous_epoch {
+        *state
+            .get_block_root(previous_epoch.start_slot(spec.slots_per_epoch), spec)
+            .unwrap()
+    } else {
+        *state
+            .get_block_root(current_epoch.start_slot(spec.slots_per_epoch), spec)
+            .unwrap()
+    };
+
+    let justified_block_root = *state
+        .get_block_root(justified_epoch.start_slot(spec.slots_per_epoch), spec)
+        .unwrap();
+
+    PendingAttestation {
+        aggregation_bitfield,
+        data: AttestationData {
+            slot,
+            shard,
+            beacon_block_root: *state.get_block_root(slot, spec).unwrap(),
+            epoch_boundary_root,
+            crosslink_data_root: Hash256::zero(),
+            latest_crosslink: Crosslink {
+                epoch: slot.epoch(spec.slots_per_epoch),
+                crosslink_data_root: Hash256::zero(),
+            },
+            justified_epoch,
+            justified_block_root,
+        },
+        custody_bitfield,
+        inclusion_slot: slot + spec.min_attestation_inclusion_delay,
+    }
+}

eth2/state_processing/src/per_epoch_processing.rs

@@ -1,11 +1,12 @@
 use attester_sets::AttesterSets;
 use errors::EpochProcessingError as Error;
+use fnv::FnvHashSet;
 use inclusion_distance::{inclusion_distance, inclusion_slot};
 use integer_sqrt::IntegerSquareRoot;
 use log::debug;
 use rayon::prelude::*;
 use ssz::TreeHash;
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::iter::FromIterator;
 use types::{validator_registry::get_active_validator_indices, *};
 use winning_root::{winning_root, WinningRoot};
@@ -17,9 +18,7 @@ pub mod tests;
 pub mod winning_root;
 
 pub fn per_epoch_processing(state: &mut BeaconState, spec: &ChainSpec) -> Result<(), Error> {
-    let current_epoch = state.current_epoch(spec);
     let previous_epoch = state.previous_epoch(spec);
-    let next_epoch = state.next_epoch(spec);
 
     debug!(
         "Starting per-epoch processing on epoch {}...",
@@ -31,14 +30,12 @@ pub fn per_epoch_processing(state: &mut BeaconState, spec: &ChainSpec) -> Result
     state.build_epoch_cache(RelativeEpoch::Current, spec)?;
     state.build_epoch_cache(RelativeEpoch::Next, spec)?;
 
-    let attesters = AttesterSets::new(&state, spec)?;
+    let attesters = calculate_attester_sets(&state, spec)?;
 
-    let active_validator_indices = get_active_validator_indices(
+    let active_validator_indices = calculate_active_validator_indices(&state, spec);
-        &state.validator_registry,
-        state.slot.epoch(spec.slots_per_epoch),
-    );
 
     let current_total_balance = state.get_total_balance(&active_validator_indices[..], spec);
+
     let previous_total_balance = state.get_total_balance(
         &get_active_validator_indices(&state.validator_registry, previous_epoch)[..],
         spec,
@@ -59,11 +56,9 @@ pub fn per_epoch_processing(state: &mut BeaconState, spec: &ChainSpec) -> Result
     let winning_root_for_shards = process_crosslinks(state, spec)?;
 
     // Rewards and Penalities
-    let active_validator_indices_hashset: HashSet<usize> =
-        HashSet::from_iter(active_validator_indices.iter().cloned());
     process_rewards_and_penalities(
         state,
-        active_validator_indices_hashset,
+        &active_validator_indices,
         &attesters,
         previous_total_balance,
         &winning_root_for_shards,
@@ -77,27 +72,9 @@ pub fn per_epoch_processing(state: &mut BeaconState, spec: &ChainSpec) -> Result
     process_validator_registry(state, spec)?;
 
     // Final updates
-    let active_tree_root = get_active_validator_indices(
+    update_active_tree_index_roots(state, spec)?;
-        &state.validator_registry,
+    update_latest_slashed_balances(state, spec);
-        next_epoch + Epoch::from(spec.activation_exit_delay),
+    clean_attestations(state, spec);
-    )
-    .hash_tree_root();
-    state.latest_active_index_roots[(next_epoch.as_usize()
-        + spec.activation_exit_delay as usize)
-        % spec.latest_active_index_roots_length] = Hash256::from_slice(&active_tree_root[..]);
-
-    state.latest_slashed_balances[next_epoch.as_usize() % spec.latest_slashed_exit_length] =
-        state.latest_slashed_balances[current_epoch.as_usize() % spec.latest_slashed_exit_length];
-    state.latest_randao_mixes[next_epoch.as_usize() % spec.latest_randao_mixes_length] = state
-        .get_randao_mix(current_epoch, spec)
-        .and_then(|x| Some(*x))
-        .ok_or_else(|| Error::NoRandaoSeed)?;
-    state.latest_attestations = state
-        .latest_attestations
-        .iter()
-        .filter(|a| a.data.slot.epoch(spec.slots_per_epoch) >= current_epoch)
-        .cloned()
-        .collect();
 
     // Rotate the epoch caches to suit the epoch transition.
     state.advance_caches();
@@ -107,8 +84,22 @@ pub fn per_epoch_processing(state: &mut BeaconState, spec: &ChainSpec) -> Result
     Ok(())
 }
 
+pub fn calculate_active_validator_indices(state: &BeaconState, spec: &ChainSpec) -> Vec<usize> {
+    get_active_validator_indices(
+        &state.validator_registry,
+        state.slot.epoch(spec.slots_per_epoch),
+    )
+}
+
+pub fn calculate_attester_sets(
+    state: &BeaconState,
+    spec: &ChainSpec,
+) -> Result<AttesterSets, BeaconStateError> {
+    AttesterSets::new(&state, spec)
+}
+
 /// Spec v0.4.0
-fn process_eth1_data(state: &mut BeaconState, spec: &ChainSpec) {
+pub fn process_eth1_data(state: &mut BeaconState, spec: &ChainSpec) {
     let next_epoch = state.next_epoch(spec);
     let voting_period = spec.epochs_per_eth1_voting_period;
 
@@ -123,7 +114,7 @@ fn process_eth1_data(state: &mut BeaconState, spec: &ChainSpec) {
 }
 
 /// Spec v0.4.0
-fn process_justification(
+pub fn process_justification(
     state: &mut BeaconState,
     current_total_balance: u64,
     previous_total_balance: u64,
@@ -201,7 +192,7 @@ fn process_justification(
 
 pub type WinningRootHashSet = HashMap<u64, WinningRoot>;
 
-fn process_crosslinks(
+pub fn process_crosslinks(
     state: &mut BeaconState,
     spec: &ChainSpec,
 ) -> Result<WinningRootHashSet, Error> {
@@ -260,9 +251,9 @@ fn process_crosslinks(
 }
 
 /// Spec v0.4.0
-fn process_rewards_and_penalities(
+pub fn process_rewards_and_penalities(
     state: &mut BeaconState,
-    active_validator_indices: HashSet<usize>,
+    active_validator_indices: &[usize],
     attesters: &AttesterSets,
     previous_total_balance: u64,
     winning_root_for_shards: &WinningRootHashSet,
@@ -270,6 +261,9 @@ fn process_rewards_and_penalities(
 ) -> Result<(), Error> {
     let next_epoch = state.next_epoch(spec);
 
+    let active_validator_indices: FnvHashSet<usize> =
+        FnvHashSet::from_iter(active_validator_indices.iter().cloned());
+
     let previous_epoch_attestations: Vec<&PendingAttestation> = state
         .latest_attestations
         .par_iter()
@@ -281,95 +275,126 @@ fn process_rewards_and_penalities(
     if base_reward_quotient == 0 {
         return Err(Error::BaseRewardQuotientIsZero);
     }
+    if previous_total_balance == 0 {
+        return Err(Error::PreviousTotalBalanceIsZero);
+    }
 
     // Justification and finalization
 
     let epochs_since_finality = next_epoch - state.finalized_epoch;
 
     if epochs_since_finality <= 4 {
-        for index in 0..state.validator_balances.len() {
+        state.validator_balances = state
-            let base_reward = state.base_reward(index, base_reward_quotient, spec);
+            .validator_balances
+            .par_iter()
+            .enumerate()
+            .map(|(index, &balance)| {
+                let mut balance = balance;
+                let base_reward = state.base_reward(index, base_reward_quotient, spec);
 
-            // Expected FFG source
+                // Expected FFG source
-            if attesters.previous_epoch.indices.contains(&index) {
+                if attesters.previous_epoch.indices.contains(&index) {
-                safe_add_assign!(
+                    safe_add_assign!(
-                    state.validator_balances[index],
+                        balance,
-                    base_reward * attesters.previous_epoch.balance / previous_total_balance
+                        base_reward * attesters.previous_epoch.balance / previous_total_balance
-                );
-            } else if active_validator_indices.contains(&index) {
-                safe_sub_assign!(state.validator_balances[index], base_reward);
-            }
-
-            // Expected FFG target
-            if attesters.previous_epoch_boundary.indices.contains(&index) {
-                safe_add_assign!(
-                    state.validator_balances[index],
-                    base_reward * attesters.previous_epoch_boundary.balance
-                        / previous_total_balance
-                );
-            } else if active_validator_indices.contains(&index) {
-                safe_sub_assign!(state.validator_balances[index], base_reward);
-            }
-
-            // Expected beacon chain head
-            if attesters.previous_epoch_head.indices.contains(&index) {
-                safe_add_assign!(
-                    state.validator_balances[index],
-                    base_reward * attesters.previous_epoch_head.balance / previous_total_balance
-                );
-            } else if active_validator_indices.contains(&index) {
-                safe_sub_assign!(state.validator_balances[index], base_reward);
-            }
-        }
-
-        // Inclusion distance
-        for &index in &attesters.previous_epoch.indices {
-            let base_reward = state.base_reward(index, base_reward_quotient, spec);
-            let inclusion_distance =
-                inclusion_distance(state, &previous_epoch_attestations, index, spec)?;
-
-            safe_add_assign!(
-                state.validator_balances[index],
-                base_reward * spec.min_attestation_inclusion_delay / inclusion_distance
-            )
-        }
-    } else {
-        for index in 0..state.validator_balances.len() {
-            let inactivity_penalty =
-                state.inactivity_penalty(index, epochs_since_finality, base_reward_quotient, spec);
-
-            if active_validator_indices.contains(&index) {
-                if !attesters.previous_epoch.indices.contains(&index) {
-                    safe_sub_assign!(state.validator_balances[index], inactivity_penalty);
-                }
-                if !attesters.previous_epoch_boundary.indices.contains(&index) {
-                    safe_sub_assign!(state.validator_balances[index], inactivity_penalty);
-                }
-                if !attesters.previous_epoch_head.indices.contains(&index) {
-                    safe_sub_assign!(state.validator_balances[index], inactivity_penalty);
-                }
-
-                if state.validator_registry[index].slashed {
-                    let base_reward = state.base_reward(index, base_reward_quotient, spec);
-                    safe_sub_assign!(
-                        state.validator_balances[index],
-                        2 * inactivity_penalty + base_reward
                     );
+                } else if active_validator_indices.contains(&index) {
+                    safe_sub_assign!(balance, base_reward);
                 }
-            }
-        }
 
-        for &index in &attesters.previous_epoch.indices {
+                // Expected FFG target
-            let base_reward = state.base_reward(index, base_reward_quotient, spec);
+                if attesters.previous_epoch_boundary.indices.contains(&index) {
-            let inclusion_distance =
+                    safe_add_assign!(
-                inclusion_distance(state, &previous_epoch_attestations, index, spec)?;
+                        balance,
+                        base_reward * attesters.previous_epoch_boundary.balance
+                            / previous_total_balance
+                    );
+                } else if active_validator_indices.contains(&index) {
+                    safe_sub_assign!(balance, base_reward);
+                }
 
-            safe_sub_assign!(
+                // Expected beacon chain head
-                state.validator_balances[index],
+                if attesters.previous_epoch_head.indices.contains(&index) {
-                base_reward
+                    safe_add_assign!(
-                    - base_reward * spec.min_attestation_inclusion_delay / inclusion_distance
+                        balance,
-            );
+                        base_reward * attesters.previous_epoch_head.balance
-        }
+                            / previous_total_balance
+                    );
+                } else if active_validator_indices.contains(&index) {
+                    safe_sub_assign!(balance, base_reward);
+                };
+
+                if attesters.previous_epoch.indices.contains(&index) {
+                    let base_reward = state.base_reward(index, base_reward_quotient, spec);
+                    let inclusion_distance =
+                        inclusion_distance(state, &previous_epoch_attestations, index, spec);
+
+                    if let Ok(inclusion_distance) = inclusion_distance {
+                        if inclusion_distance > 0 {
+                            safe_add_assign!(
+                                balance,
+                                base_reward * spec.min_attestation_inclusion_delay
+                                    / inclusion_distance
+                            )
+                        }
+                    }
+                }
+
+                balance
+            })
+            .collect();
+    } else {
+        state.validator_balances = state
+            .validator_balances
+            .par_iter()
+            .enumerate()
+            .map(|(index, &balance)| {
+                let mut balance = balance;
+
+                let inactivity_penalty = state.inactivity_penalty(
+                    index,
+                    epochs_since_finality,
+                    base_reward_quotient,
+                    spec,
+                );
+
+                if active_validator_indices.contains(&index) {
+                    if !attesters.previous_epoch.indices.contains(&index) {
+                        safe_sub_assign!(balance, inactivity_penalty);
+                    }
+                    if !attesters.previous_epoch_boundary.indices.contains(&index) {
+                        safe_sub_assign!(balance, inactivity_penalty);
+                    }
+                    if !attesters.previous_epoch_head.indices.contains(&index) {
+                        safe_sub_assign!(balance, inactivity_penalty);
+                    }
+
+                    if state.validator_registry[index].slashed {
+                        let base_reward = state.base_reward(index, base_reward_quotient, spec);
+                        safe_sub_assign!(balance, 2 * inactivity_penalty + base_reward);
+                    }
+                }
+
+                if attesters.previous_epoch.indices.contains(&index) {
+                    let base_reward = state.base_reward(index, base_reward_quotient, spec);
+                    let inclusion_distance =
+                        inclusion_distance(state, &previous_epoch_attestations, index, spec);
+
+                    if let Ok(inclusion_distance) = inclusion_distance {
+                        if inclusion_distance > 0 {
+                            safe_sub_assign!(
+                                balance,
+                                base_reward
+                                    - base_reward * spec.min_attestation_inclusion_delay
+                                        / inclusion_distance
+                            );
+                        }
+                    }
+                }
+
+                balance
+            })
+            .collect();
     }
 
     // Attestation inclusion
@@ -413,8 +438,8 @@ fn process_rewards_and_penalities(
             if let Some(winning_root) = winning_root_for_shards.get(&shard) {
                 // Hash set de-dedups and (hopefully) offers a speed improvement from faster
                 // lookups.
-                let attesting_validator_indices: HashSet<usize> =
+                let attesting_validator_indices: FnvHashSet<usize> =
-                    HashSet::from_iter(winning_root.attesting_validator_indices.iter().cloned());
+                    FnvHashSet::from_iter(winning_root.attesting_validator_indices.iter().cloned());
 
                 for &index in &crosslink_committee {
                     let base_reward = state.base_reward(index, base_reward_quotient, spec);
@@ -444,7 +469,7 @@ fn process_rewards_and_penalities(
 }
 
 // Spec v0.4.0
-fn process_validator_registry(state: &mut BeaconState, spec: &ChainSpec) -> Result<(), Error> {
+pub fn process_validator_registry(state: &mut BeaconState, spec: &ChainSpec) -> Result<(), Error> {
     let current_epoch = state.current_epoch(spec);
     let next_epoch = state.next_epoch(spec);
 
@@ -489,3 +514,44 @@ fn process_validator_registry(state: &mut BeaconState, spec: &ChainSpec) -> Resu
 
     Ok(())
 }
+
+// Spec v0.4.0
+pub fn update_active_tree_index_roots(
+    state: &mut BeaconState,
+    spec: &ChainSpec,
+) -> Result<(), Error> {
+    let next_epoch = state.next_epoch(spec);
+
+    let active_tree_root = get_active_validator_indices(
+        &state.validator_registry,
+        next_epoch + Epoch::from(spec.activation_exit_delay),
+    )
+    .hash_tree_root();
+
+    state.latest_active_index_roots[(next_epoch.as_usize()
+        + spec.activation_exit_delay as usize)
+        % spec.latest_active_index_roots_length] = Hash256::from_slice(&active_tree_root[..]);
+
+    Ok(())
+}
+
+// Spec v0.4.0
+pub fn update_latest_slashed_balances(state: &mut BeaconState, spec: &ChainSpec) {
+    let current_epoch = state.current_epoch(spec);
+    let next_epoch = state.next_epoch(spec);
+
+    state.latest_slashed_balances[next_epoch.as_usize() % spec.latest_slashed_exit_length] =
+        state.latest_slashed_balances[current_epoch.as_usize() % spec.latest_slashed_exit_length];
+}
+
+// Spec v0.4.0
+pub fn clean_attestations(state: &mut BeaconState, spec: &ChainSpec) {
+    let current_epoch = state.current_epoch(spec);
+
+    state.latest_attestations = state
+        .latest_attestations
+        .iter()
+        .filter(|a| a.data.slot.epoch(spec.slots_per_epoch) >= current_epoch)
+        .cloned()
+        .collect();
+}

eth2/state_processing/src/per_epoch_processing/attester_sets.rs

@@ -1,9 +1,9 @@
-use std::collections::HashSet;
+use fnv::FnvHashSet;
 use types::*;
 
 #[derive(Default)]
 pub struct Attesters {
-    pub indices: HashSet<usize>,
+    pub indices: FnvHashSet<usize>,
     pub balance: u64,
 }
 

eth2/state_processing/src/per_epoch_processing/errors.rs

@@ -6,6 +6,8 @@ pub enum EpochProcessingError {
     NoBlockRoots,
     BaseRewardQuotientIsZero,
     NoRandaoSeed,
+    PreviousTotalBalanceIsZero,
+    InclusionDistanceZero,
     BeaconStateError(BeaconStateError),
     InclusionError(InclusionError),
 }

eth2/state_processing/src/per_epoch_processing/tests.rs

@@ -1,21 +1,18 @@
 #![cfg(test)]
 use crate::per_epoch_processing;
+use benching_utils::BeaconStateBencher;
 use env_logger::{Builder, Env};
-use types::beacon_state::BeaconStateBuilder;
 use types::*;
 
 #[test]
 fn runs_without_error() {
     Builder::from_env(Env::default().default_filter_or("error")).init();
 
-    let mut builder = BeaconStateBuilder::new(8);
+    let spec = ChainSpec::few_validators();
-    builder.spec = ChainSpec::few_validators();
 
-    builder.build().unwrap();
+    let mut builder = BeaconStateBencher::new(8, &spec);
-    builder.teleport_to_end_of_epoch(builder.spec.genesis_epoch + 4);
+    builder.teleport_to_end_of_epoch(spec.genesis_epoch + 4, &spec);
+    let mut state = builder.build();
 
-    let mut state = builder.cloned_state();
+    per_epoch_processing(&mut state, &spec).unwrap();
-
-    let spec = &builder.spec;
-    per_epoch_processing(&mut state, spec).unwrap();
 }

eth2/types/src/beacon_state/helpers.rs

@@ -11,7 +11,7 @@ pub fn verify_bitfield_length(bitfield: &Bitfield, committee_size: usize) -> boo
     }
 
     for i in committee_size..(bitfield.num_bytes() * 8) {
-        if bitfield.get(i).expect("Impossible due to previous check.") {
+        if bitfield.get(i).unwrap_or(false) {
             return false;
         }
     }