Backport Core PRs to beta (#7891)
* update back-references more aggressively after answering from cache (#7578) * Add new EF ropstens nodes. (#7824) * Add new EF ropstens nodes. * Fix tests * Add a timeout for light client sync requests (#7848) * Add a timeout for light client sync requests * Adjusting timeout to number of headers * Flush keyfiles. Resolves #7632 (#7868) * Fix wallet import (#7873) * rpc: generate new account id for imported wallets * ethstore: handle duplicate wallet filenames * ethstore: simplify deduplication of wallet file names * ethstore: do not dedup wallet filenames on update * ethstore: fix minor grumbles * [WASM] mem_cmp added to the Wasm runtime (#7539) * mem_cmp added to the Wasm runtime * schedule.wasm.mem_copy to schedule.wasm.mem_cmp for mem_cmp * [Wasm] memcmp fix and test added (#7590) * [Wasm] memcmp fix and test added * [Wasm] use reqrep_test! macro for memcmp test * wasmi interpreter (#7796) * adjust storage update evm-style (#7812) * disable internal memory (#7842)
This commit is contained in:
committed by
Afri Schoedon
parent
af70a681d5
commit
b60511e3d2
@@ -35,6 +35,7 @@
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::mem;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Instant, Duration};
|
||||
|
||||
use ethcore::encoded;
|
||||
use light::client::{AsLightClient, LightChainClient};
|
||||
@@ -58,6 +59,13 @@ mod sync_round;
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
// Base number of milliseconds for the header request timeout.
|
||||
const REQ_TIMEOUT_MILLISECS_BASE: u64 = 7000;
|
||||
// Additional number of milliseconds for each requested header.
|
||||
// If we request N headers, then the timeout will be:
|
||||
// REQ_TIMEOUT_MILLISECS_BASE + N * REQ_TIMEOUT_MILLISECS_PER_HEADER
|
||||
const REQ_TIMEOUT_MILLISECS_PER_HEADER: u64 = 10;
|
||||
|
||||
/// Peer chain info.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
struct ChainInfo {
|
||||
@@ -225,12 +233,18 @@ pub struct LightSync<L: AsLightClient> {
|
||||
start_block_number: u64,
|
||||
best_seen: Mutex<Option<ChainInfo>>, // best seen block on the network.
|
||||
peers: RwLock<HashMap<PeerId, Mutex<Peer>>>, // peers which are relevant to synchronization.
|
||||
pending_reqs: Mutex<HashSet<ReqId>>, // requests from this handler.
|
||||
pending_reqs: Mutex<HashMap<ReqId, PendingReq>>, // requests from this handler
|
||||
client: Arc<L>,
|
||||
rng: Mutex<OsRng>,
|
||||
state: Mutex<SyncState>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct PendingReq {
|
||||
started: Instant,
|
||||
timeout: Duration,
|
||||
}
|
||||
|
||||
impl<L: AsLightClient + Send + Sync> Handler for LightSync<L> {
|
||||
fn on_connect(
|
||||
&self,
|
||||
@@ -353,7 +367,7 @@ impl<L: AsLightClient + Send + Sync> Handler for LightSync<L> {
|
||||
return
|
||||
}
|
||||
|
||||
if !self.pending_reqs.lock().remove(&req_id) {
|
||||
if self.pending_reqs.lock().remove(&req_id).is_none() {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -412,7 +426,7 @@ impl<L: AsLightClient> LightSync<L> {
|
||||
*state = SyncState::AncestorSearch(AncestorSearch::begin(chain_info.best_block_number));
|
||||
}
|
||||
|
||||
// handles request dispatch, block import, and state machine transitions.
|
||||
// handles request dispatch, block import, state machine transitions, and timeouts.
|
||||
fn maintain_sync(&self, ctx: &BasicContext) {
|
||||
const DRAIN_AMOUNT: usize = 128;
|
||||
|
||||
@@ -502,6 +516,32 @@ impl<L: AsLightClient> LightSync<L> {
|
||||
}
|
||||
}
|
||||
|
||||
// handle requests timeouts
|
||||
{
|
||||
let mut pending_reqs = self.pending_reqs.lock();
|
||||
let mut unfulfilled = Vec::new();
|
||||
for (req_id, info) in pending_reqs.iter() {
|
||||
if info.started.elapsed() >= info.timeout {
|
||||
debug!(target: "sync", "{} timed out", req_id);
|
||||
unfulfilled.push(req_id.clone());
|
||||
}
|
||||
}
|
||||
|
||||
if !unfulfilled.is_empty() {
|
||||
for unfulfilled in unfulfilled.iter() {
|
||||
pending_reqs.remove(unfulfilled);
|
||||
}
|
||||
drop(pending_reqs);
|
||||
|
||||
*state = match mem::replace(&mut *state, SyncState::Idle) {
|
||||
SyncState::Idle => SyncState::Idle,
|
||||
SyncState::AncestorSearch(search) =>
|
||||
SyncState::AncestorSearch(search.requests_abandoned(&unfulfilled)),
|
||||
SyncState::Rounds(round) => SyncState::Rounds(round.requests_abandoned(&unfulfilled)),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
// allow dispatching of requests.
|
||||
{
|
||||
let peers = self.peers.read();
|
||||
@@ -535,7 +575,12 @@ impl<L: AsLightClient> LightSync<L> {
|
||||
if requested_from.contains(peer) { continue }
|
||||
match ctx.request_from(*peer, request.clone()) {
|
||||
Ok(id) => {
|
||||
self.pending_reqs.lock().insert(id.clone());
|
||||
let timeout_ms = REQ_TIMEOUT_MILLISECS_BASE +
|
||||
req.max * REQ_TIMEOUT_MILLISECS_PER_HEADER;
|
||||
self.pending_reqs.lock().insert(id.clone(), PendingReq {
|
||||
started: Instant::now(),
|
||||
timeout: Duration::from_millis(timeout_ms),
|
||||
});
|
||||
requested_from.insert(peer.clone());
|
||||
|
||||
return Some(id)
|
||||
@@ -571,7 +616,7 @@ impl<L: AsLightClient> LightSync<L> {
|
||||
start_block_number: client.as_light_client().chain_info().best_block_number,
|
||||
best_seen: Mutex::new(None),
|
||||
peers: RwLock::new(HashMap::new()),
|
||||
pending_reqs: Mutex::new(HashSet::new()),
|
||||
pending_reqs: Mutex::new(HashMap::new()),
|
||||
client: client,
|
||||
rng: Mutex::new(OsRng::new()?),
|
||||
state: Mutex::new(SyncState::Idle),
|
||||
|
||||
@@ -179,7 +179,7 @@ impl Fetcher {
|
||||
};
|
||||
|
||||
trace!(target: "sync", "Received response for subchain ({} -> {})",
|
||||
request.subchain_parent.0 + 1, request.subchain_end.0);
|
||||
request.subchain_parent.0, request.subchain_end.0);
|
||||
|
||||
let headers = ctx.data();
|
||||
|
||||
@@ -241,6 +241,8 @@ impl Fetcher {
|
||||
}
|
||||
|
||||
fn requests_abandoned(mut self, abandoned: &[ReqId]) -> SyncRound {
|
||||
trace!(target: "sync", "Abandonned requests {:?}", abandoned);
|
||||
|
||||
for abandoned in abandoned {
|
||||
match self.pending.remove(abandoned) {
|
||||
None => {},
|
||||
@@ -258,12 +260,14 @@ impl Fetcher {
|
||||
while let Some(pending_req) = self.requests.pop() {
|
||||
match dispatcher(pending_req.headers_request.clone()) {
|
||||
Some(req_id) => {
|
||||
trace!(target: "sync", "Assigned request for subchain ({} -> {})",
|
||||
pending_req.subchain_parent.0 + 1, pending_req.subchain_end.0);
|
||||
trace!(target: "sync", "Assigned request {} for subchain ({} -> {})",
|
||||
req_id, pending_req.subchain_parent.0, pending_req.subchain_end.0);
|
||||
|
||||
self.pending.insert(req_id, pending_req);
|
||||
}
|
||||
None => {
|
||||
trace!(target: "sync", "Failed to assign request for subchain ({} -> {})",
|
||||
pending_req.subchain_parent.0, pending_req.subchain_end.0);
|
||||
self.requests.push(pending_req);
|
||||
break;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user