cerc-io/lighthouse
11
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Fix clippy warnings (#813)

Browse Source 
* Clippy account manager

* Clippy account_manager

* Clippy beacon_node/beacon_chain

* Clippy beacon_node/client

* Clippy beacon_node/eth1

* Clippy beacon_node/eth2-libp2p

* Clippy beacon_node/genesis

* Clippy beacon_node/network

* Clippy beacon_node/rest_api

* Clippy beacon_node/src

* Clippy beacon_node/store

* Clippy eth2/lmd_ghost

* Clippy eth2/operation_pool

* Clippy eth2/state_processing

* Clippy eth2/types

* Clippy eth2/utils/bls

* Clippy eth2/utils/cahced_tree_hash

* Clippy eth2/utils/deposit_contract

* Clippy eth2/utils/eth2_interop_keypairs

* Clippy eth2/utils/eth2_testnet_config

* Clippy eth2/utils/lighthouse_metrics

* Clippy eth2/utils/ssz

* Clippy eth2/utils/ssz_types

* Clippy eth2/utils/tree_hash_derive

* Clippy lcli

* Clippy tests/beacon_chain_sim

* Clippy validator_client

* Cargo fmt
This commit is contained in:
pscott
2020-01-21 18:38:56 +11:00
committed by Age Manning
parent 1abb964652
commit 7396cd2cab
78 changed files with 387 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files
beacon_node/beacon_chain/src/beacon_chain.rs
+63 -60
View File
@@ -25,6 +25,7 @@ use state_processing::{
per_block_processing, per_slot_processing, BlockProcessingError, BlockSignatureStrategy,
};
use std::borrow::Cow;
use std::cmp::Ordering;
use std::fs;
use std::io::prelude::*;
use std::sync::Arc;
@@ -512,65 +513,67 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
pub fn state_at_slot(&self, slot: Slot) -> Result<BeaconState<T::EthSpec>, Error> {
let head_state = self.head()?.beacon_state;
if slot == head_state.slot {
Ok(head_state)
} else if slot > head_state.slot {
if slot > head_state.slot + T::EthSpec::slots_per_epoch() {
warn!(
self.log,
"Skipping more than an epoch";
"head_slot" => head_state.slot,
"request_slot" => slot
)
}
let start_slot = head_state.slot;
let task_start = Instant::now();
let max_task_runtime = Duration::from_millis(self.spec.milliseconds_per_slot);
let head_state_slot = head_state.slot;
let mut state = head_state;
while state.slot < slot {
// Do not allow and forward state skip that takes longer than the maximum task duration.
//
// This is a protection against nodes doing too much work when they're not synced
// to a chain.
if task_start + max_task_runtime < Instant::now() {
return Err(Error::StateSkipTooLarge {
start_slot,
requested_slot: slot,
max_task_runtime,
});
match slot.cmp(&head_state.slot) {
Ordering::Equal => Ok(head_state),
Ordering::Greater => {
if slot > head_state.slot + T::EthSpec::slots_per_epoch() {
warn!(
self.log,
"Skipping more than an epoch";
"head_slot" => head_state.slot,
"request_slot" => slot
)
}
// Note: supplying some `state_root` when it is known would be a cheap and easy
// optimization.
match per_slot_processing(&mut state, None, &self.spec) {
Ok(()) => (),
Err(e) => {
warn!(
self.log,
"Unable to load state at slot";
"error" => format!("{:?}", e),
"head_slot" => head_state_slot,
"requested_slot" => slot
);
return Err(Error::NoStateForSlot(slot));
}
};
}
Ok(state)
} else {
let state_root = self
.rev_iter_state_roots()?
.take_while(|(_root, current_slot)| *current_slot >= slot)
.find(|(_root, current_slot)| *current_slot == slot)
.map(|(root, _slot)| root)
.ok_or_else(|| Error::NoStateForSlot(slot))?;
let start_slot = head_state.slot;
let task_start = Instant::now();
let max_task_runtime = Duration::from_millis(self.spec.milliseconds_per_slot);
Ok(self
.get_state_caching(&state_root, Some(slot))?
.ok_or_else(|| Error::NoStateForSlot(slot))?)
let head_state_slot = head_state.slot;
let mut state = head_state;
while state.slot < slot {
// Do not allow and forward state skip that takes longer than the maximum task duration.
//
// This is a protection against nodes doing too much work when they're not synced
// to a chain.
if task_start + max_task_runtime < Instant::now() {
return Err(Error::StateSkipTooLarge {
start_slot,
requested_slot: slot,
max_task_runtime,
});
}
// Note: supplying some `state_root` when it is known would be a cheap and easy
// optimization.
match per_slot_processing(&mut state, None, &self.spec) {
Ok(()) => (),
Err(e) => {
warn!(
self.log,
"Unable to load state at slot";
"error" => format!("{:?}", e),
"head_slot" => head_state_slot,
"requested_slot" => slot
);
return Err(Error::NoStateForSlot(slot));
}
};
}
Ok(state)
}
Ordering::Less => {
let state_root = self
.rev_iter_state_roots()?
.take_while(|(_root, current_slot)| *current_slot >= slot)
.find(|(_root, current_slot)| *current_slot == slot)
.map(|(root, _slot)| root)
.ok_or_else(|| Error::NoStateForSlot(slot))?;
Ok(self
.get_state_caching(&state_root, Some(slot))?
.ok_or_else(|| Error::NoStateForSlot(slot))?)
}
}
}
@@ -638,7 +641,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
let head_state = &self.head()?.beacon_state;
let mut state = if epoch(slot) == epoch(head_state.slot) {
self.head()?.beacon_state.clone()
self.head()?.beacon_state
} else {
self.state_at_slot(slot)?
};
@@ -671,7 +674,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
let head_state = &self.head()?.beacon_state;
let mut state = if epoch == as_epoch(head_state.slot) {
self.head()?.beacon_state.clone()
self.head()?.beacon_state
} else {
self.state_at_slot(epoch.start_slot(T::EthSpec::slots_per_epoch()))?
};
@@ -1754,9 +1757,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
let mut dump = vec![];
let mut last_slot = CheckPoint {
beacon_block: self.head()?.beacon_block.clone(),
beacon_block: self.head()?.beacon_block,
beacon_block_root: self.head()?.beacon_block_root,
beacon_state: self.head()?.beacon_state.clone(),
beacon_state: self.head()?.beacon_state,
beacon_state_root: self.head()?.beacon_state_root,
};
beacon_node/beacon_chain/src/builder.rs
+5 -5
View File
@@ -448,7 +448,7 @@ where
let fork_choice = if let Some(persisted_beacon_chain) = &self.persisted_beacon_chain {
ForkChoice::from_ssz_container(
persisted_beacon_chain.fork_choice.clone(),
store.clone(),
store,
block_root_tree,
)
.map_err(|e| format!("Unable to decode fork choice from db: {:?}", e))?
@@ -462,7 +462,7 @@ where
.ok_or_else(|| "fork_choice_backend requires a genesis_block_root")?;
let backend = ThreadSafeReducedTree::new(
store.clone(),
store,
block_root_tree,
&finalized_checkpoint.beacon_block,
finalized_checkpoint.beacon_block_root,
@@ -626,7 +626,7 @@ mod test {
#[test]
fn recent_genesis() {
let validator_count = 8;
let genesis_time = 13371337;
let genesis_time = 13_371_337;
let log = get_logger();
let store = Arc::new(MemoryStore::open());
@@ -641,7 +641,7 @@ mod test {
let chain = BeaconChainBuilder::new(MinimalEthSpec)
.logger(log.clone())
.store(store.clone())
.store(store)
.store_migrator(NullMigrator)
.genesis_state(genesis_state)
.expect("should build state using recent genesis")
@@ -662,7 +662,7 @@ mod test {
assert_eq!(state.slot, Slot::new(0), "should start from genesis");
assert_eq!(
state.genesis_time, 13371337,
state.genesis_time, 13_371_337,
"should have the correct genesis time"
);
assert_eq!(
beacon_node/beacon_chain/src/eth1_chain.rs
+30 -30
View File
@@ -8,6 +8,7 @@ use rand::prelude::*;
use slog::{crit, debug, error, trace, Logger};
use ssz_derive::{Decode, Encode};
use state_processing::per_block_processing::get_new_eth1_data;
use std::cmp::Ordering;
use std::collections::HashMap;
use std::iter::DoubleEndedIterator;
use std::iter::FromIterator;
@@ -343,8 +344,7 @@ impl<T: EthSpec, S: Store<T>> Eth1ChainBackend<T, S> for CachingEth1Backend<T, S
.iter()
.rev()
.skip_while(|eth1_block| eth1_block.timestamp > voting_period_start_seconds)
.skip(eth1_follow_distance as usize)
.next()
.nth(eth1_follow_distance as usize)
.map(|block| {
trace!(
self.log,
@@ -392,21 +392,21 @@ impl<T: EthSpec, S: Store<T>> Eth1ChainBackend<T, S> for CachingEth1Backend<T, S
state.eth1_data.deposit_count
};
if deposit_index > deposit_count {
Err(Error::DepositIndexTooHigh)
} else if deposit_index == deposit_count {
Ok(vec![])
} else {
let next = deposit_index;
let last = std::cmp::min(deposit_count, next + T::MaxDeposits::to_u64());
match deposit_index.cmp(&deposit_count) {
Ordering::Greater => Err(Error::DepositIndexTooHigh),
Ordering::Equal => Ok(vec![]),
Ordering::Less => {
let next = deposit_index;
let last = std::cmp::min(deposit_count, next + T::MaxDeposits::to_u64());
self.core
.deposits()
.read()
.cache
.get_deposits(next, last, deposit_count, DEPOSIT_TREE_DEPTH)
.map_err(|e| Error::BackendError(format!("Failed to get deposits: {:?}", e)))
.map(|(_deposit_root, deposits)| deposits)
self.core
.deposits()
.read()
.cache
.get_deposits(next, last, deposit_count, DEPOSIT_TREE_DEPTH)
.map_err(|e| Error::BackendError(format!("Failed to get deposits: {:?}", e)))
.map(|(_deposit_root, deposits)| deposits)
}
}
}
@@ -775,7 +775,7 @@ mod test {
let deposits_for_inclusion = eth1_chain
.deposits_for_block_inclusion(&state, &random_eth1_data(), spec)
.expect(&format!("should find deposit for {}", i));
.unwrap_or_else(|_| panic!("should find deposit for {}", i));
let expected_len =
std::cmp::min(i - initial_deposit_index, max_deposits as usize);
@@ -853,7 +853,7 @@ mod test {
state.slot = Slot::new(period * 1_000 + period / 2);
// Add 50% of the votes so a lookup is required.
for _ in 0..period / 2 + 1 {
for _ in 0..=period / 2 {
state
.eth1_data_votes
.push(random_eth1_data())
@@ -932,7 +932,7 @@ mod test {
state.slot = Slot::new(period / 2);
// Add 50% of the votes so a lookup is required.
for _ in 0..period / 2 + 1 {
for _ in 0..=period / 2 {
state
.eth1_data_votes
.push(random_eth1_data())
@@ -1090,7 +1090,7 @@ mod test {
eth1_block.number,
*new_eth1_data
.get(&eth1_block.clone().eth1_data().unwrap())
.expect(&format!(
.unwrap_or_else(|| panic!(
"new_eth1_data should have expected block #{}",
eth1_block.number
))
@@ -1135,8 +1135,8 @@ mod test {
let votes = collect_valid_votes(
&state,
HashMap::from_iter(new_eth1_data.clone().into_iter()),
HashMap::from_iter(all_eth1_data.clone().into_iter()),
HashMap::from_iter(new_eth1_data.into_iter()),
HashMap::from_iter(all_eth1_data.into_iter()),
);
assert_eq!(
votes.len(),
@@ -1164,7 +1164,7 @@ mod test {
let votes = collect_valid_votes(
&state,
HashMap::from_iter(new_eth1_data.clone().into_iter()),
HashMap::from_iter(all_eth1_data.clone().into_iter()),
HashMap::from_iter(all_eth1_data.into_iter()),
);
assert_votes!(
votes,
@@ -1196,8 +1196,8 @@ mod test {
let votes = collect_valid_votes(
&state,
HashMap::from_iter(new_eth1_data.clone().into_iter()),
HashMap::from_iter(all_eth1_data.clone().into_iter()),
HashMap::from_iter(new_eth1_data.into_iter()),
HashMap::from_iter(all_eth1_data.into_iter()),
);
assert_votes!(
votes,
@@ -1230,12 +1230,12 @@ mod test {
.expect("should have some eth1 data")
.clone();
state.eth1_data_votes = vec![non_new_eth1_data.0.clone()].into();
state.eth1_data_votes = vec![non_new_eth1_data.0].into();
let votes = collect_valid_votes(
&state,
HashMap::from_iter(new_eth1_data.clone().into_iter()),
HashMap::from_iter(all_eth1_data.clone().into_iter()),
HashMap::from_iter(new_eth1_data.into_iter()),
HashMap::from_iter(all_eth1_data.into_iter()),
);
assert_votes!(
@@ -1268,8 +1268,8 @@ mod test {
let votes = collect_valid_votes(
&state,
HashMap::from_iter(new_eth1_data.clone().into_iter()),
HashMap::from_iter(all_eth1_data.clone().into_iter()),
HashMap::from_iter(new_eth1_data.into_iter()),
HashMap::from_iter(all_eth1_data.into_iter()),
);
assert_votes!(
beacon_node/beacon_chain/src/fork_choice.rs
+1 -1
View File
@@ -294,7 +294,7 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
/// Returns a `SszForkChoice` which contains the current state of `Self`.
pub fn as_ssz_container(&self) -> SszForkChoice {
SszForkChoice {
genesis_block_root: self.genesis_block_root.clone(),
genesis_block_root: self.genesis_block_root,
justified_checkpoint: self.justified_checkpoint.read().clone(),
best_justified_checkpoint: self.best_justified_checkpoint.read().clone(),
backend_bytes: self.backend.as_bytes(),
beacon_node/beacon_chain/src/head_tracker.rs
+2 -2
View File
@@ -59,10 +59,10 @@ impl HeadTracker {
let slots_len = ssz_container.slots.len();
if roots_len != slots_len {
return Err(Error::MismatchingLengths {
Err(Error::MismatchingLengths {
roots_len,
slots_len,
});
})
} else {
let map = HashMap::from_iter(
ssz_container
beacon_node/beacon_chain/src/test_utils.rs
+1 -2
View File
@@ -173,7 +173,7 @@ impl<E: EthSpec> BeaconChainHarness<DiskHarnessType<E>> {
let chain = BeaconChainBuilder::new(eth_spec_instance)
.logger(log.clone())
.custom_spec(spec.clone())
.custom_spec(spec)
.store(store.clone())
.store_migrator(<BlockingMigrator<_> as Migrate<_, E>>::new(store))
.resume_from_db(Eth1Config::default())
@@ -236,7 +236,6 @@ where
self.chain
.state_at_slot(state_slot)
.expect("should find state for slot")
.clone()
};
// Determine the first slot where a block should be built.
beacon_node/client/src/builder.rs
+9 -9
View File
@@ -151,7 +151,7 @@ where
let builder = BeaconChainBuilder::new(eth_spec_instance)
.logger(context.log.clone())
.store(store.clone())
.store(store)
.store_migrator(store_migrator)
.custom_spec(spec.clone());
@@ -223,7 +223,7 @@ where
Box::new(future)
}
ClientGenesis::RemoteNode { server, .. } => {
let future = Bootstrapper::connect(server.to_string(), &context.log)
let future = Bootstrapper::connect(server, &context.log)
.map_err(|e| {
format!("Failed to initialize bootstrap client: {}", e)
})
@@ -306,14 +306,14 @@ where
.ok_or_else(|| "http_server requires a libp2p network sender")?;
let network_info = rest_api::NetworkInfo {
network_service: network.clone(),
network_chan: network_send.clone(),
network_service: network,
network_chan: network_send,
};
let (exit_signal, listening_addr) = rest_api::start_server(
&client_config.rest_api,
&context.executor,
beacon_chain.clone(),
beacon_chain,
network_info,
client_config
.create_db_path()
@@ -529,7 +529,7 @@ where
spec,
context.log,
)
.map_err(|e| format!("Unable to open database: {:?}", e).to_string())?;
.map_err(|e| format!("Unable to open database: {:?}", e))?;
self.store = Some(Arc::new(store));
Ok(self)
}
@@ -557,8 +557,8 @@ where
{
/// Specifies that the `Client` should use a `DiskStore` database.
pub fn simple_disk_store(mut self, path: &Path) -> Result<Self, String> {
let store = SimpleDiskStore::open(path)
.map_err(|e| format!("Unable to open database: {:?}", e).to_string())?;
let store =
SimpleDiskStore::open(path).map_err(|e| format!("Unable to open database: {:?}", e))?;
self.store = Some(Arc::new(store));
Ok(self)
}
@@ -660,7 +660,7 @@ where
.ok_or_else(|| "caching_eth1_backend requires a store".to_string())?;
let backend = if let Some(eth1_service_from_genesis) = self.eth1_service {
eth1_service_from_genesis.update_config(config.clone())?;
eth1_service_from_genesis.update_config(config)?;
// This cache is not useful because it's first (earliest) block likely the block that
// triggered genesis.
beacon_node/client/src/notifier.rs
+6 -5
View File
@@ -17,7 +17,7 @@ pub const WARN_PEER_COUNT: usize = 1;
const SECS_PER_MINUTE: f64 = 60.0;
const SECS_PER_HOUR: f64 = 3600.0;
const SECS_PER_DAY: f64 = 86400.0; // non-leap
const SECS_PER_WEEK: f64 = 604800.0; // non-leap
const SECS_PER_WEEK: f64 = 604_800.0; // non-leap
const DAYS_PER_WEEK: f64 = 7.0;
const HOURS_PER_DAY: f64 = 24.0;
const MINUTES_PER_HOUR: f64 = 60.0;
@@ -166,13 +166,14 @@ pub fn spawn_notifier<T: BeaconChainTypes>(
.then(move |result| {
match result {
Ok(()) => Ok(()),
Err(e) => Ok(error!(
Err(e) => {
error!(
log_3,
"Notifier failed to notify";
"error" => format!("{:?}", e)
))
}
});
);
Ok(())
} } });
let (exit_signal, exit) = exit_future::signal();
context
beacon_node/eth1/src/block_cache.rs
+1 -1
View File
@@ -224,7 +224,7 @@ mod tests {
);
}
let mut cache_2 = cache.clone();
let mut cache_2 = cache;
cache_2.truncate(17);
assert_eq!(
cache_2.blocks.len(),
beacon_node/eth1/src/deposit_cache.rs
+24 -23
View File
@@ -1,6 +1,7 @@
use crate::DepositLog;
use eth2_hashing::hash;
use ssz_derive::{Decode, Encode};
use std::cmp::Ordering;
use tree_hash::TreeHash;
use types::{Deposit, Hash256, DEPOSIT_TREE_DEPTH};
@@ -196,24 +197,26 @@ impl DepositCache {
/// - If a log with index `log.index - 1` is not already present in `self` (ignored when empty).
/// - If a log with `log.index` is already known, but the given `log` is distinct to it.
pub fn insert_log(&mut self, log: DepositLog) -> Result<(), Error> {
if log.index == self.logs.len() as u64 {
let deposit = Hash256::from_slice(&log.deposit_data.tree_hash_root());
self.leaves.push(deposit);
self.logs.push(log);
self.deposit_tree.push_leaf(deposit)?;
self.deposit_roots.push(self.deposit_tree.root());
Ok(())
} else if log.index < self.logs.len() as u64 {
if self.logs[log.index as usize] == log {
match log.index.cmp(&(self.logs.len() as u64)) {
Ordering::Equal => {
let deposit = Hash256::from_slice(&log.deposit_data.tree_hash_root());
self.leaves.push(deposit);
self.logs.push(log);
self.deposit_tree.push_leaf(deposit)?;
self.deposit_roots.push(self.deposit_tree.root());
Ok(())
} else {
Err(Error::DuplicateDistinctLog(log.index))
}
} else {
Err(Error::NonConsecutive {
Ordering::Less => {
if self.logs[log.index as usize] == log {
Ok(())
} else {
Err(Error::DuplicateDistinctLog(log.index))
}
}
Ordering::Greater => Err(Error::NonConsecutive {
log_index: log.index,
expected: self.logs.len(),
})
}),
}
}
@@ -312,14 +315,12 @@ impl DepositCache {
.logs
.binary_search_by(|deposit| deposit.block_number.cmp(&block_number));
match index {
Ok(index) => return self.logs.get(index).map(|x| x.index + 1),
Err(next) => {
return Some(
self.logs
.get(next.saturating_sub(1))
.map_or(0, |x| x.index + 1),
)
}
Ok(index) => self.logs.get(index).map(|x| x.index + 1),
Err(next) => Some(
self.logs
.get(next.saturating_sub(1))
.map_or(0, |x| x.index + 1),
),
}
}
@@ -329,7 +330,7 @@ impl DepositCache {
/// and queries the `deposit_roots` map to get the corresponding `deposit_root`.
pub fn get_deposit_root_from_cache(&self, block_number: u64) -> Option<Hash256> {
let index = self.get_deposit_count_from_cache(block_number)?;
Some(self.deposit_roots.get(index as usize)?.clone())
Some(*self.deposit_roots.get(index as usize)?)
}
}
beacon_node/eth1/src/http.rs
+2 -2
View File
@@ -136,7 +136,7 @@ pub fn get_deposit_count(
timeout,
)
.and_then(|result| match result {
None => Err(format!("Deposit root response was none")),
None => Err("Deposit root response was none".to_string()),
Some(bytes) => {
if bytes.is_empty() {
Ok(None)
@@ -173,7 +173,7 @@ pub fn get_deposit_root(
timeout,
)
.and_then(|result| match result {
None => Err(format!("Deposit root response was none")),
None => Err("Deposit root response was none".to_string()),
Some(bytes) => {
if bytes.is_empty() {
Ok(None)
beacon_node/eth1/src/service.rs
+1 -1
View File
@@ -613,7 +613,7 @@ impl Service {
Ok(BlockCacheUpdateOutcome::Success {
blocks_imported,
head_block_number: cache_4.clone().block_cache.read().highest_block_number(),
head_block_number: cache_4.block_cache.read().highest_block_number(),
})
})
}
beacon_node/eth1/tests/test.rs
+1 -1
View File
@@ -758,7 +758,7 @@ mod fast {
log,
);
let n = 10;
let deposits: Vec<_> = (0..n).into_iter().map(|_| random_deposit_data()).collect();
let deposits: Vec<_> = (0..n).map(|_| random_deposit_data()).collect();
for deposit in &deposits {
deposit_contract
.deposit(runtime, deposit.clone())
beacon_node/eth2-libp2p/src/behaviour.rs
+2 -2
View File
@@ -57,7 +57,7 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
net_conf: &NetworkConfig,
log: &slog::Logger,
) -> error::Result<Self> {
let local_peer_id = local_key.public().clone().into_peer_id();
let local_peer_id = local_key.public().into_peer_id();
let behaviour_log = log.new(o!());
let ping_config = PingConfig::new()
@@ -74,7 +74,7 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
Ok(Behaviour {
eth2_rpc: RPC::new(log.clone()),
gossipsub: Gossipsub::new(local_peer_id.clone(), net_conf.gs_config.clone()),
gossipsub: Gossipsub::new(local_peer_id, net_conf.gs_config.clone()),
discovery: Discovery::new(local_key, net_conf, log)?,
ping: Ping::new(ping_config),
identify,
beacon_node/eth2-libp2p/src/rpc/handler.rs
+2 -2
View File
@@ -145,7 +145,7 @@ where
// When terminating a stream, report the stream termination to the requesting user via
// an RPC error
let error = RPCErrorResponse::ServerError(ErrorMessage {
error_message: "Request timed out".as_bytes().to_vec(),
error_message: b"Request timed out".to_vec(),
});
// The stream termination type is irrelevant, this will terminate the
@@ -510,7 +510,7 @@ where
// notify the user
return Ok(Async::Ready(ProtocolsHandlerEvent::Custom(
RPCEvent::Error(
stream_id.get_ref().clone(),
*stream_id.get_ref(),
RPCError::Custom("Stream timed out".into()),
),
)));
beacon_node/eth2-libp2p/tests/common/mod.rs
+4 -5
View File
@@ -43,8 +43,7 @@ pub fn build_libp2p_instance(
) -> LibP2PService {
let config = build_config(port, boot_nodes, secret_key);
// launch libp2p service
let libp2p_service = LibP2PService::new(config.clone(), log.clone()).unwrap();
libp2p_service
LibP2PService::new(config, log.clone()).unwrap()
}
#[allow(dead_code)]
@@ -64,10 +63,10 @@ pub fn build_full_mesh(log: slog::Logger, n: usize, start_port: Option<u16>) ->
.map(|x| get_enr(&x).multiaddr()[1].clone())
.collect();
for i in 0..n {
for j in i..n {
for (i, node) in nodes.iter_mut().enumerate().take(n) {
for (j, multiaddr) in multiaddrs.iter().enumerate().skip(i) {
if i != j {
match libp2p::Swarm::dial_addr(&mut nodes[i].swarm, multiaddrs[j].clone()) {
match libp2p::Swarm::dial_addr(&mut node.swarm, multiaddr.clone()) {
Ok(()) => debug!(log, "Connected"),
Err(_) => error!(log, "Failed to connect"),
};
beacon_node/eth2-libp2p/tests/gossipsub_tests.rs
+28 -36
View File
@@ -62,7 +62,7 @@ fn test_gossipsub_forward() {
// Every node except the corner nodes are connected to 2 nodes.
if subscribed_count == (num_nodes * 2) - 2 {
node.swarm.publish(
&vec![Topic::new(topic.into_string())],
&[Topic::new(topic.into_string())],
pubsub_message.clone(),
);
}
@@ -96,45 +96,37 @@ fn test_gossipsub_full_mesh_publish() {
let mut received_count = 0;
tokio::run(futures::future::poll_fn(move || -> Result<_, ()> {
for node in nodes.iter_mut() {
loop {
match node.poll().unwrap() {
Async::Ready(Some(Libp2pEvent::PubsubMessage {
topics, message, ..
})) => {
assert_eq!(topics.len(), 1);
// Assert topic is the published topic
assert_eq!(
topics.first().unwrap(),
&TopicHash::from_raw(publishing_topic.clone())
);
// Assert message received is the correct one
assert_eq!(message, pubsub_message.clone());
received_count += 1;
if received_count == num_nodes - 1 {
return Ok(Async::Ready(()));
}
}
_ => break,
while let Async::Ready(Some(Libp2pEvent::PubsubMessage {
topics, message, ..
})) = node.poll().unwrap()
{
assert_eq!(topics.len(), 1);
// Assert topic is the published topic
assert_eq!(
topics.first().unwrap(),
&TopicHash::from_raw(publishing_topic.clone())
);
// Assert message received is the correct one
assert_eq!(message, pubsub_message.clone());
received_count += 1;
if received_count == num_nodes - 1 {
return Ok(Async::Ready(()));
}
}
}
loop {
match publishing_node.poll().unwrap() {
Async::Ready(Some(Libp2pEvent::PeerSubscribed(_, topic))) => {
// Received topics is one of subscribed eth2 topics
assert!(topic.clone().into_string().starts_with("/eth2/"));
// Publish on beacon block topic
if topic == TopicHash::from_raw("/eth2/beacon_block/ssz") {
subscribed_count += 1;
if subscribed_count == num_nodes - 1 {
publishing_node.swarm.publish(
&vec![Topic::new(topic.into_string())],
pubsub_message.clone(),
);
}
}
while let Async::Ready(Some(Libp2pEvent::PeerSubscribed(_, topic))) =
publishing_node.poll().unwrap()
{
// Received topics is one of subscribed eth2 topics
assert!(topic.clone().into_string().starts_with("/eth2/"));
// Publish on beacon block topic
if topic == TopicHash::from_raw("/eth2/beacon_block/ssz") {
subscribed_count += 1;
if subscribed_count == num_nodes - 1 {
publishing_node
.swarm
.publish(&[Topic::new(topic.into_string())], pubsub_message.clone());
}
_ => break,
}
}
Ok(Async::NotReady)
beacon_node/eth2-libp2p/tests/rpc_tests.rs
+21 -25
View File
@@ -3,6 +3,7 @@ use eth2_libp2p::rpc::methods::*;
use eth2_libp2p::rpc::*;
use eth2_libp2p::{Libp2pEvent, RPCEvent};
use slog::{warn, Level};
use std::sync::atomic::{AtomicBool, Ordering::Relaxed};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::prelude::*;
@@ -106,20 +107,19 @@ fn test_status_rpc() {
});
// execute the futures and check the result
let test_result = Arc::new(Mutex::new(false));
let test_result = Arc::new(AtomicBool::new(false));
let error_result = test_result.clone();
let thread_result = test_result.clone();
tokio::run(
sender_future
.select(receiver_future)
.timeout(Duration::from_millis(1000))
.map_err(move |_| *error_result.lock().unwrap() = false)
.map_err(move |_| error_result.store(false, Relaxed))
.map(move |result| {
*thread_result.lock().unwrap() = result.0;
()
thread_result.store(result.0, Relaxed);
}),
);
assert!(*test_result.lock().unwrap());
assert!(test_result.load(Relaxed));
}
#[test]
@@ -236,20 +236,19 @@ fn test_blocks_by_range_chunked_rpc() {
});
// execute the futures and check the result
let test_result = Arc::new(Mutex::new(false));
let test_result = Arc::new(AtomicBool::new(false));
let error_result = test_result.clone();
let thread_result = test_result.clone();
tokio::run(
sender_future
.select(receiver_future)
.timeout(Duration::from_millis(1000))
.map_err(move |_| *error_result.lock().unwrap() = false)
.map_err(move |_| error_result.store(false, Relaxed))
.map(move |result| {
*thread_result.lock().unwrap() = result.0;
()
thread_result.store(result.0, Relaxed);
}),
);
assert!(*test_result.lock().unwrap());
assert!(test_result.load(Relaxed));
}
#[test]
@@ -359,20 +358,19 @@ fn test_blocks_by_range_single_empty_rpc() {
});
// execute the futures and check the result
let test_result = Arc::new(Mutex::new(false));
let test_result = Arc::new(AtomicBool::new(false));
let error_result = test_result.clone();
let thread_result = test_result.clone();
tokio::run(
sender_future
.select(receiver_future)
.timeout(Duration::from_millis(1000))
.map_err(move |_| *error_result.lock().unwrap() = false)
.map_err(move |_| error_result.store(false, Relaxed))
.map(move |result| {
*thread_result.lock().unwrap() = result.0;
()
thread_result.store(result.0, Relaxed);
}),
);
assert!(*test_result.lock().unwrap());
assert!(test_result.load(Relaxed));
}
#[test]
@@ -486,20 +484,19 @@ fn test_blocks_by_root_chunked_rpc() {
});
// execute the futures and check the result
let test_result = Arc::new(Mutex::new(false));
let test_result = Arc::new(AtomicBool::new(false));
let error_result = test_result.clone();
let thread_result = test_result.clone();
tokio::run(
sender_future
.select(receiver_future)
.timeout(Duration::from_millis(1000))
.map_err(move |_| *error_result.lock().unwrap() = false)
.map_err(move |_| error_result.store(false, Relaxed))
.map(move |result| {
*thread_result.lock().unwrap() = result.0;
()
thread_result.store(result.0, Relaxed);
}),
);
assert!(*test_result.lock().unwrap());
assert!(test_result.load(Relaxed));
}
#[test]
@@ -558,18 +555,17 @@ fn test_goodbye_rpc() {
});
// execute the futures and check the result
let test_result = Arc::new(Mutex::new(false));
let test_result = Arc::new(AtomicBool::new(false));
let error_result = test_result.clone();
let thread_result = test_result.clone();
tokio::run(
sender_future
.select(receiver_future)
.timeout(Duration::from_millis(1000))
.map_err(move |_| *error_result.lock().unwrap() = false)
.map_err(move |_| error_result.store(false, Relaxed))
.map(move |result| {
*thread_result.lock().unwrap() = result.0;
()
thread_result.store(result.0, Relaxed);
}),
);
assert!(*test_result.lock().unwrap());
assert!(test_result.load(Relaxed));
}
beacon_node/genesis/src/common.rs
+1 -1
View File
@@ -19,7 +19,7 @@ pub fn genesis_deposits(
let depth = spec.deposit_contract_tree_depth as usize;
let mut tree = MerkleTree::create(&[], depth);
for (i, deposit_leaf) in deposit_root_leaves.iter().enumerate() {
if let Err(_) = tree.push_leaf(*deposit_leaf, depth) {
if tree.push_leaf(*deposit_leaf, depth).is_err() {
return Err(String::from("Failed to push leaf"));
}
beacon_node/genesis/tests/tests.rs
+1 -2
View File
@@ -59,7 +59,6 @@ fn basic() {
spec.min_genesis_active_validator_count = 8;
let deposits = (0..spec.min_genesis_active_validator_count + 2)
.into_iter()
.map(|i| {
deposit_contract.deposit_helper::<MinimalEthSpec>(
generate_deterministic_keypair(i as usize),
@@ -73,7 +72,7 @@ fn basic() {
})
.collect::<Vec<_>>();
let deposit_future = deposit_contract.deposit_multiple(deposits.clone());
let deposit_future = deposit_contract.deposit_multiple(deposits);
let wait_future =
service.wait_for_genesis_state::<MinimalEthSpec>(update_interval, spec.clone());
beacon_node/network/src/message_handler.rs
+1 -1
View File
@@ -229,7 +229,7 @@ impl<T: BeaconChainTypes> MessageHandler<T> {
.on_block_gossip(peer_id.clone(), block);
// TODO: Apply more sophisticated validation and decoding logic
if should_forward_on {
self.propagate_message(id, peer_id.clone());
self.propagate_message(id, peer_id);
}
}
Err(e) => {
beacon_node/network/src/message_processor.rs
+4 -4
View File
@@ -203,7 +203,7 @@ impl<T: BeaconChainTypes> MessageProcessor<T> {
);
self.network
.disconnect(peer_id.clone(), GoodbyeReason::IrrelevantNetwork);
.disconnect(peer_id, GoodbyeReason::IrrelevantNetwork);
} else if remote.head_slot
> self.chain.slot().unwrap_or_else(|_| Slot::from(0u64)) + FUTURE_SLOT_TOLERANCE
{
@@ -219,7 +219,7 @@ impl<T: BeaconChainTypes> MessageProcessor<T> {
"reason" => "different system clocks or genesis time"
);
self.network
.disconnect(peer_id.clone(), GoodbyeReason::IrrelevantNetwork);
.disconnect(peer_id, GoodbyeReason::IrrelevantNetwork);
} else if remote.finalized_epoch <= local.finalized_epoch
&& remote.finalized_root != Hash256::zero()
&& local.finalized_root != Hash256::zero()
@@ -239,7 +239,7 @@ impl<T: BeaconChainTypes> MessageProcessor<T> {
"reason" => "different finalized chain"
);
self.network
.disconnect(peer_id.clone(), GoodbyeReason::IrrelevantNetwork);
.disconnect(peer_id, GoodbyeReason::IrrelevantNetwork);
} else if remote.finalized_epoch < local.finalized_epoch {
// The node has a lower finalized epoch, their chain is not useful to us. There are two
// cases where a node can have a lower finalized epoch:
@@ -512,7 +512,7 @@ impl<T: BeaconChainTypes> MessageProcessor<T> {
// Inform the sync manager to find parents for this block
trace!(self.log, "Block with unknown parent received";
"peer_id" => format!("{:?}",peer_id));
self.send_to_sync(SyncMessage::UnknownBlock(peer_id, Box::new(block.clone())));
self.send_to_sync(SyncMessage::UnknownBlock(peer_id, Box::new(block)));
SHOULD_FORWARD_GOSSIP_BLOCK
}
BlockProcessingOutcome::FutureSlot {
beacon_node/network/src/service.rs
+1 -1
View File
@@ -263,7 +263,7 @@ fn network_service(
id,
source,
message,
topics: _,
..
} => {
message_handler_send
.try_send(HandlerMessage::PubsubMessage(id, source, message))
beacon_node/network/src/sync/network_context.rs
+3 -3
View File
@@ -66,7 +66,7 @@ impl SyncNetworkContext {
"count" => request.count,
"peer" => format!("{:?}", peer_id)
);
self.send_rpc_request(peer_id.clone(), RPCRequest::BlocksByRange(request))
self.send_rpc_request(peer_id, RPCRequest::BlocksByRange(request))
}
pub fn blocks_by_root_request(
@@ -81,7 +81,7 @@ impl SyncNetworkContext {
"count" => request.block_roots.len(),
"peer" => format!("{:?}", peer_id)
);
self.send_rpc_request(peer_id.clone(), RPCRequest::BlocksByRoot(request))
self.send_rpc_request(peer_id, RPCRequest::BlocksByRoot(request))
}
pub fn downvote_peer(&mut self, peer_id: PeerId) {
@@ -91,7 +91,7 @@ impl SyncNetworkContext {
"peer" => format!("{:?}", peer_id)
);
// TODO: Implement reputation
self.disconnect(peer_id.clone(), GoodbyeReason::Fault);
self.disconnect(peer_id, GoodbyeReason::Fault);
}
fn disconnect(&mut self, peer_id: PeerId, reason: GoodbyeReason) {
beacon_node/network/src/sync/range_sync/batch.rs
+8 -8
View File
@@ -62,16 +62,16 @@ impl<T: EthSpec> PendingBatches<T> {
let peer_request = batch.current_peer.clone();
self.peer_requests
.entry(peer_request)
.or_insert_with(|| HashSet::new())
.or_insert_with(HashSet::new)
.insert(request_id);
self.batches.insert(request_id, batch)
}
pub fn remove(&mut self, request_id: &RequestId) -> Option<Batch<T>> {
if let Some(batch) = self.batches.remove(request_id) {
pub fn remove(&mut self, request_id: RequestId) -> Option<Batch<T>> {
if let Some(batch) = self.batches.remove(&request_id) {
if let Entry::Occupied(mut entry) = self.peer_requests.entry(batch.current_peer.clone())
{
entry.get_mut().remove(request_id);
entry.get_mut().remove(&request_id);
if entry.get().is_empty() {
entry.remove();
@@ -85,8 +85,8 @@ impl<T: EthSpec> PendingBatches<T> {
/// Adds a block to the batches if the request id exists. Returns None if there is no batch
/// matching the request id.
pub fn add_block(&mut self, request_id: &RequestId, block: BeaconBlock<T>) -> Option<()> {
let batch = self.batches.get_mut(request_id)?;
pub fn add_block(&mut self, request_id: RequestId, block: BeaconBlock<T>) -> Option<()> {
let batch = self.batches.get_mut(&request_id)?;
batch.downloaded_blocks.push(block);
Some(())
}
@@ -101,7 +101,7 @@ impl<T: EthSpec> PendingBatches<T> {
pub fn remove_batch_by_peer(&mut self, peer_id: &PeerId) -> Option<Batch<T>> {
let request_ids = self.peer_requests.get(peer_id)?;
let request_id = request_ids.iter().next()?.clone();
self.remove(&request_id)
let request_id = *request_ids.iter().next()?;
self.remove(request_id)
}
}
beacon_node/network/src/sync/range_sync/chain.rs
+6 -6
View File
@@ -144,11 +144,11 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
) -> Option<ProcessingResult> {
if let Some(block) = beacon_block {
// This is not a stream termination, simply add the block to the request
self.pending_batches.add_block(&request_id, block.clone())?;
return Some(ProcessingResult::KeepChain);
self.pending_batches.add_block(request_id, block.clone())?;
Some(ProcessingResult::KeepChain)
} else {
// A stream termination has been sent. This batch has ended. Process a completed batch.
let batch = self.pending_batches.remove(&request_id)?;
let batch = self.pending_batches.remove(request_id)?;
Some(self.process_completed_batch(chain.clone(), network, batch, log))
}
}
@@ -433,7 +433,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
return true;
}
}
return false;
false
}
/// Returns a peer if there exists a peer which does not currently have a pending request.
@@ -500,10 +500,10 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
&mut self,
network: &mut SyncNetworkContext,
peer_id: &PeerId,
request_id: &RequestId,
request_id: RequestId,
log: &slog::Logger,
) -> Option<ProcessingResult> {
if let Some(batch) = self.pending_batches.remove(&request_id) {
if let Some(batch) = self.pending_batches.remove(request_id) {
warn!(log, "Batch failed. RPC Error"; "id" => batch.id, "retries" => batch.retries, "peer" => format!("{:?}", peer_id));
Some(self.failed_batch(network, batch, log))
beacon_node/network/src/sync/range_sync/range.rs
+22 -23
View File
@@ -188,7 +188,7 @@ impl<T: BeaconChainTypes> RangeSync<T> {
debug!(self.log, "Adding peer to the existing head chain peer pool"; "head_root" => format!("{}",remote.head_root), "head_slot" => remote.head_slot, "peer_id" => format!("{:?}", peer_id));
// add the peer to the head's pool
chain.add_peer(network, peer_id.clone(), &self.log);
chain.add_peer(network, peer_id, &self.log);
} else {
// There are no other head chains that match this peer's status, create a new one, and
let start_slot = std::cmp::min(local_info.head_slot, remote_finalized_slot);
@@ -305,29 +305,28 @@ impl<T: BeaconChainTypes> RangeSync<T> {
/// retries. In this case, we need to remove the chain and re-status all the peers.
fn remove_peer(&mut self, network: &mut SyncNetworkContext, peer_id: &PeerId) {
let log_ref = &self.log;
match self.chains.head_finalized_request(|chain| {
if chain.peer_pool.remove(peer_id) {
// this chain contained the peer
while let Some(batch) = chain.pending_batches.remove_batch_by_peer(peer_id) {
if let ProcessingResult::RemoveChain =
chain.failed_batch(network, batch, log_ref)
{
// a single batch failed, remove the chain
return Some(ProcessingResult::RemoveChain);
if let Some((index, ProcessingResult::RemoveChain)) =
self.chains.head_finalized_request(|chain| {
if chain.peer_pool.remove(peer_id) {
// this chain contained the peer
while let Some(batch) = chain.pending_batches.remove_batch_by_peer(peer_id) {
if let ProcessingResult::RemoveChain =
chain.failed_batch(network, batch, log_ref)
{
// a single batch failed, remove the chain
return Some(ProcessingResult::RemoveChain);
}
}
// peer removed from chain, no batch failed
Some(ProcessingResult::KeepChain)
} else {
None
}
// peer removed from chain, no batch failed
Some(ProcessingResult::KeepChain)
} else {
None
}
}) {
Some((index, ProcessingResult::RemoveChain)) => {
// the chain needed to be removed
debug!(self.log, "Chain being removed due to failed batch");
self.chains.remove_chain(network, index, &self.log);
}
_ => {} // chain didn't need to be removed, ignore
})
{
// the chain needed to be removed
debug!(self.log, "Chain being removed due to failed batch");
self.chains.remove_chain(network, index, &self.log);
}
}
@@ -344,7 +343,7 @@ impl<T: BeaconChainTypes> RangeSync<T> {
// check that this request is pending
let log_ref = &self.log;
match self.chains.head_finalized_request(|chain| {
chain.inject_error(network, &peer_id, &request_id, log_ref)
chain.inject_error(network, &peer_id, request_id, log_ref)
}) {
Some((_, ProcessingResult::KeepChain)) => {} // error handled chain persists
Some((index, ProcessingResult::RemoveChain)) => {
beacon_node/rest_api/src/helpers.rs
+2 -2
View File
@@ -145,7 +145,7 @@ pub fn state_at_slot<T: BeaconChainTypes>(
// I'm not sure if this `.clone()` will be optimized out. If not, it seems unnecessary.
Ok((
beacon_chain.head()?.beacon_state_root,
beacon_chain.head()?.beacon_state.clone(),
beacon_chain.head()?.beacon_state,
))
} else {
let root = state_root_at_slot(beacon_chain, slot)?;
@@ -209,7 +209,7 @@ pub fn state_root_at_slot<T: BeaconChainTypes>(
//
// Use `per_slot_processing` to advance the head state to the present slot,
// assuming that all slots do not contain a block (i.e., they are skipped slots).
let mut state = beacon_chain.head()?.beacon_state.clone();
let mut state = beacon_chain.head()?.beacon_state;
let spec = &T::EthSpec::default_spec();
for _ in state.slot.as_u64()..slot.as_u64() {
beacon_node/rest_api/src/lib.rs
+2
View File
@@ -54,6 +54,8 @@ pub struct NetworkInfo<T: BeaconChainTypes> {
pub network_chan: mpsc::UnboundedSender<NetworkMessage>,
}
// Allowing more than 7 arguments.
#[allow(clippy::too_many_arguments)]
pub fn start_server<T: BeaconChainTypes>(
config: &Config,
executor: &TaskExecutor,
beacon_node/rest_api/src/router.rs
+2
View File
@@ -19,6 +19,8 @@ where
Box::new(item.into_future())
}
// Allowing more than 7 arguments.
#[allow(clippy::too_many_arguments)]
pub fn route<T: BeaconChainTypes>(
req: Request<Body>,
beacon_chain: Arc<BeaconChain<T>>,
beacon_node/rest_api/tests/test.rs
+8 -14
View File
@@ -47,8 +47,7 @@ fn get_randao_reveal<T: BeaconChainTypes>(
.head()
.expect("should get head")
.beacon_state
.fork
.clone();
.fork;
let proposer_index = beacon_chain
.block_proposer(slot)
.expect("should get proposer index");
@@ -69,8 +68,7 @@ fn sign_block<T: BeaconChainTypes>(
.head()
.expect("should get head")
.beacon_state
.fork
.clone();
.fork;
let proposer_index = beacon_chain
.block_proposer(block.slot)
.expect("should get proposer index");
@@ -91,11 +89,7 @@ fn validator_produce_attestation() {
.client
.beacon_chain()
.expect("client should have beacon chain");
let state = beacon_chain
.head()
.expect("should get head")
.beacon_state
.clone();
let state = beacon_chain.head().expect("should get head").beacon_state;
let validator_index = 0;
let duties = state
@@ -169,7 +163,7 @@ fn validator_produce_attestation() {
remote_node
.http
.validator()
.publish_attestation(attestation.clone()),
.publish_attestation(attestation),
)
.expect("should publish attestation");
assert!(
@@ -344,7 +338,7 @@ fn validator_block_post() {
remote_node
.http
.validator()
.produce_block(slot, randao_reveal.clone()),
.produce_block(slot, randao_reveal),
)
.expect("should fetch block from http api");
@@ -360,12 +354,12 @@ fn validator_block_post() {
);
}
sign_block(beacon_chain.clone(), &mut block, spec);
sign_block(beacon_chain, &mut block, spec);
let block_root = block.canonical_root();
let publish_status = env
.runtime()
.block_on(remote_node.http.validator().publish_block(block.clone()))
.block_on(remote_node.http.validator().publish_block(block))
.expect("should publish block");
if cfg!(not(feature = "fake_crypto")) {
@@ -419,7 +413,7 @@ fn validator_block_get() {
.expect("client should have beacon chain");
let slot = Slot::new(1);
let randao_reveal = get_randao_reveal(beacon_chain.clone(), slot, spec);
let randao_reveal = get_randao_reveal(beacon_chain, slot, spec);
let block = env
.runtime()
beacon_node/src/config.rs
+1 -1
View File
@@ -268,7 +268,7 @@ pub fn get_configs<E: EthSpec>(
if eth2_config.spec_constants != client_config.spec_constants {
crit!(log, "Specification constants do not match.";
"client_config" => client_config.spec_constants.to_string(),
"eth2_config" => eth2_config.spec_constants.to_string()
"eth2_config" => eth2_config.spec_constants
);
return Err("Specification constant mismatch".into());
}
beacon_node/store/benches/benches.rs
+1 -2
View File
@@ -24,7 +24,6 @@ fn get_state<E: EthSpec>(validator_count: usize) -> BeaconState<E> {
}
state.validators = (0..validator_count)
.into_iter()
.collect::<Vec<_>>()
.par_iter()
.map(|&i| Validator {
@@ -77,7 +76,7 @@ fn all_benches(c: &mut Criterion) {
.sample_size(10),
);
let inner_state = state.clone();
let inner_state = state;
c.bench(
&format!("{}_validators", validator_count),
Benchmark::new("encode/beacon_state/committee_cache[0]", move |b| {
beacon_node/store/examples/ssz_encode_state_container.rs
-1
View File
@@ -27,7 +27,6 @@ fn get_state<E: EthSpec>(validator_count: usize) -> BeaconState<E> {
}
state.validators = (0..validator_count)
.into_iter()
.collect::<Vec<_>>()
.par_iter()
.map(|&i| Validator {
beacon_node/store/src/block_at_slot.rs
+5 -6
View File
@@ -1,5 +1,6 @@
use super::*;
use ssz::{Decode, DecodeError};
use std::cmp::Ordering;
fn get_block_bytes<T: Store<E>, E: EthSpec>(
store: &T,
@@ -45,12 +46,10 @@ fn get_at_preceeding_slot<T: Store<E>, E: EthSpec>(
if let Some(bytes) = get_block_bytes::<_, E>(store, root)? {
let this_slot = read_slot_from_block_bytes(&bytes)?;
if this_slot == slot {
break Ok(Some((root, bytes)));
} else if this_slot < slot {
break Ok(None);
} else {
root = read_parent_root_from_block_bytes(&bytes)?;
match this_slot.cmp(&slot) {
Ordering::Equal => break Ok(Some((root, bytes))),
Ordering::Less => break Ok(None),
Ordering::Greater => root = read_parent_root_from_block_bytes(&bytes)?,
}
} else {
break Ok(None);
beacon_node/store/src/block_root_tree.rs
+4 -5
View File
@@ -237,7 +237,7 @@ mod test {
.add_block_root(int_hash(i), int_hash(i - 1), Slot::new(i))
.expect("add_block_root ok");
let expected = (1..i + 1)
let expected = (1..=i)
.rev()
.map(|j| (int_hash(j), Slot::new(j)))
.collect::<Vec<_>>();
@@ -262,12 +262,12 @@ mod test {
.add_block_root(int_hash(i), int_hash(i - step_length), Slot::new(i))
.expect("add_block_root ok");
let sparse_expected = (1..i + 1)
let sparse_expected = (1..=i)
.rev()
.step_by(step_length as usize)
.map(|j| (int_hash(j), Slot::new(j)))
.collect_vec();
let every_slot_expected = (1..i + 1)
let every_slot_expected = (1..=i)
.rev()
.map(|j| {
let nearest = 1 + (j - 1) / step_length * step_length;
@@ -343,10 +343,9 @@ mod test {
// Check that advancing the finalized root onto one side completely removes the other
// side.
let fin_tree = tree.clone();
let fin_tree = tree;
let prune_point = num_blocks / 2;
let remaining_fork1_blocks = all_fork1_blocks
.clone()
.into_iter()
.take_while(|(_, slot)| *slot >= prune_point)
.collect_vec();
beacon_node/store/src/chunked_vector.rs
+3 -3
View File
@@ -185,7 +185,7 @@ pub trait Field<E: EthSpec>: Copy {
.values
.first()
.cloned()
.ok_or(ChunkError::MissingGenesisValue.into())
.ok_or_else(|| ChunkError::MissingGenesisValue.into())
}
/// Store the given `value` as the genesis value for this field, unless stored already.
@@ -685,7 +685,7 @@ mod test {
];
assert_eq!(
stitch(chunks.clone(), 2, 6, chunk_size, 12, 99).unwrap(),
stitch(chunks, 2, 6, chunk_size, 12, 99).unwrap(),
vec![99, 99, 2, 3, 4, 5, 99, 99, 99, 99, 99, 99]
);
}
@@ -707,7 +707,7 @@ mod test {
);
assert_eq!(
stitch(chunks.clone(), 2, 10, chunk_size, 8, default).unwrap(),
stitch(chunks, 2, 10, chunk_size, 8, default).unwrap(),
vec![v(8), v(9), v(2), v(3), v(4), v(5), v(6), v(7)]
);
}
beacon_node/store/src/forwards_iter.rs
+5 -5
View File
@@ -20,9 +20,9 @@ pub struct SimpleForwardsBlockRootsIterator {
/// Fusion of the above two approaches to forwards iteration. Fast and efficient.
pub enum HybridForwardsBlockRootsIterator<E: EthSpec> {
PreFinalization {
iter: FrozenForwardsBlockRootsIterator<E>,
iter: Box<FrozenForwardsBlockRootsIterator<E>>,
/// Data required by the `PostFinalization` iterator when we get to it.
continuation_data: Option<(BeaconState<E>, Hash256)>,
continuation_data: Box<Option<(BeaconState<E>, Hash256)>>,
},
PostFinalization {
iter: SimpleForwardsBlockRootsIterator,
@@ -99,13 +99,13 @@ impl<E: EthSpec> HybridForwardsBlockRootsIterator<E> {
if start_slot < latest_restore_point_slot {
PreFinalization {
iter: FrozenForwardsBlockRootsIterator::new(
iter: Box::new(FrozenForwardsBlockRootsIterator::new(
store,
start_slot,
latest_restore_point_slot,
spec,
),
continuation_data: Some((end_state, end_block_root)),
)),
continuation_data: Box::new(Some((end_state, end_block_root))),
}
} else {
PostFinalization {
beacon_node/store/src/hot_cold_store.rs
+5 -4
View File
@@ -145,14 +145,15 @@ impl<E: EthSpec> Store<E> for HotColdDB<E> {
let current_split_slot = store.get_split_slot();
if frozen_head.slot < current_split_slot {
Err(HotColdDBError::FreezeSlotError {
return Err(HotColdDBError::FreezeSlotError {
current_split_slot,
proposed_split_slot: frozen_head.slot,
})?;
}
.into());
}
if frozen_head.slot % E::slots_per_epoch() != 0 {
Err(HotColdDBError::FreezeSlotUnaligned(frozen_head.slot))?;
return Err(HotColdDBError::FreezeSlotUnaligned(frozen_head.slot).into());
}
// 1. Copy all of the states between the head and the split slot, from the hot DB
@@ -574,7 +575,7 @@ impl<E: EthSpec> HotColdDB<E> {
let key = Self::restore_point_key(restore_point_index);
RestorePointHash::db_get(&self.cold_db, &key)?
.map(|r| r.state_root)
.ok_or(HotColdDBError::MissingRestorePointHash(restore_point_index).into())
.ok_or_else(|| HotColdDBError::MissingRestorePointHash(restore_point_index).into())
}
/// Store the state root of a restore point.
beacon_node/store/src/iter.rs
+2 -2
View File
@@ -345,7 +345,7 @@ mod test {
state_b.state_roots[0] = state_a_root;
store.put_state(&state_a_root, &state_a).unwrap();
let iter = BlockRootsIterator::new(store.clone(), &state_b);
let iter = BlockRootsIterator::new(store, &state_b);
assert!(
iter.clone().any(|(_root, slot)| slot == 0),
@@ -394,7 +394,7 @@ mod test {
store.put_state(&state_a_root, &state_a).unwrap();
store.put_state(&state_b_root, &state_b).unwrap();
let iter = StateRootsIterator::new(store.clone(), &state_b);
let iter = StateRootsIterator::new(store, &state_b);
assert!(
iter.clone().any(|(_root, slot)| slot == 0),
beacon_node/store/src/memory_store.rs
+1 -5
View File
@@ -47,11 +47,7 @@ impl<E: EthSpec> Store<E> for MemoryStore<E> {
fn get_bytes(&self, col: &str, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
let column_key = Self::get_key_for_col(col, key);
Ok(self
.db
.read()
.get(&column_key)
.and_then(|val| Some(val.clone())))
Ok(self.db.read().get(&column_key).cloned())
}
/// Puts a key in the database.
beacon_node/store/src/migrate.rs
+3 -4
View File
@@ -60,13 +60,12 @@ impl<E: EthSpec, S: Store<E>> Migrate<S, E> for BlockingMigrator<S> {
}
}
type MpscSender<E> = mpsc::Sender<(Hash256, BeaconState<E>)>;
/// Migrator that runs a background thread to migrate state from the hot to the cold database.
pub struct BackgroundMigrator<E: EthSpec> {
db: Arc<DiskStore<E>>,
tx_thread: Mutex<(
mpsc::Sender<(Hash256, BeaconState<E>)>,
thread::JoinHandle<()>,
)>,
tx_thread: Mutex<(MpscSender<E>, thread::JoinHandle<()>)>,
}
impl<E: EthSpec> Migrate<DiskStore<E>, E> for BackgroundMigrator<E> {