diff --git a/Cargo.lock b/Cargo.lock index 7eaf5b81f24..a7ebf333150 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1577,6 +1577,7 @@ version = "1.12.0" dependencies = [ "client-traits", "common-types", + "derive_more", "engine", "enum-primitive-derive", "env_logger 0.5.13", diff --git a/ethcore/client-traits/src/lib.rs b/ethcore/client-traits/src/lib.rs index ed0ad0dedc4..7692e9fc016 100644 --- a/ethcore/client-traits/src/lib.rs +++ b/ethcore/client-traits/src/lib.rs @@ -401,6 +401,9 @@ pub trait BlockChainClient: /// Schedule state-altering transaction to be executed on the next pending /// block with the given gas and nonce parameters. fn transact(&self, tx_request: TransactionRequest) -> Result<(), transaction::Error>; + + /// Returns true, if underlying import queue is processing possible fork at the moment + fn is_processing_fork(&self) -> bool; } /// The data required for a `Client` to create a transaction. diff --git a/ethcore/src/client/client.rs b/ethcore/src/client/client.rs index 17419c5eca2..043e04d808d 100644 --- a/ethcore/src/client/client.rs +++ b/ethcore/src/client/client.rs @@ -1764,6 +1764,11 @@ impl BlockChainClient for Client { } } + fn is_processing_fork(&self) -> bool { + let chain = self.chain.read(); + self.importer.block_queue.is_processing_fork(&chain.best_block_hash(), &chain) + } + fn block_total_difficulty(&self, id: BlockId) -> Option { let chain = self.chain.read(); diff --git a/ethcore/src/test_helpers/test_client.rs b/ethcore/src/test_helpers/test_client.rs index 0f4cde783b8..c0808f8dcc7 100644 --- a/ethcore/src/test_helpers/test_client.rs +++ b/ethcore/src/test_helpers/test_client.rs @@ -810,6 +810,8 @@ impl BlockChainClient for TestBlockChainClient { } } + fn is_processing_fork(&self) -> bool { false } + // works only if blocks are one after another 1 -> 2 -> 3 fn tree_route(&self, from: &H256, to: &H256) -> Option { Some(TreeRoute { diff --git a/ethcore/sync/Cargo.toml b/ethcore/sync/Cargo.toml index 61621ae281e..77c54534b78 100644 --- a/ethcore/sync/Cargo.toml +++ b/ethcore/sync/Cargo.toml @@ -13,6 +13,7 @@ bytes = { package = "parity-bytes", version = "0.1" } client-traits = { path = "../client-traits" } common-types = { path = "../types" } devp2p = { package = "ethcore-network-devp2p", path = "../../util/network-devp2p" } +derive_more = "0.99" enum-primitive-derive = "0.2" ethcore-io = { path = "../../util/io" } ethcore-private-tx = { path = "../private-tx" } diff --git a/ethcore/sync/src/api.rs b/ethcore/sync/src/api.rs index cd96c9eea58..7cfdc330f7a 100644 --- a/ethcore/sync/src/api.rs +++ b/ethcore/sync/src/api.rs @@ -466,6 +466,7 @@ const MAINTAIN_SYNC_TIMER: TimerToken = 1; const CONTINUE_SYNC_TIMER: TimerToken = 2; const TX_TIMER: TimerToken = 3; const PRIORITY_TIMER: TimerToken = 4; +const DELAYED_PROCESSING_TIMER: TimerToken = 5; pub(crate) const PRIORITY_TIMER_INTERVAL: Duration = Duration::from_millis(250); @@ -489,6 +490,7 @@ impl NetworkProtocolHandler for SyncProtocolHandler { io.register_timer(MAINTAIN_SYNC_TIMER, Duration::from_millis(1100)).expect("Error registering sync timer"); io.register_timer(CONTINUE_SYNC_TIMER, Duration::from_millis(2500)).expect("Error registering sync timer"); io.register_timer(TX_TIMER, Duration::from_millis(1300)).expect("Error registering transactions timer"); + io.register_timer(DELAYED_PROCESSING_TIMER, Duration::from_millis(2100)).expect("Error registering delayed processing timer"); io.register_timer(PRIORITY_TIMER, PRIORITY_TIMER_INTERVAL).expect("Error registering peers timer"); } @@ -539,6 +541,7 @@ impl NetworkProtocolHandler for SyncProtocolHandler { CONTINUE_SYNC_TIMER => self.sync.write().continue_sync(&mut io), TX_TIMER => self.sync.write().propagate_new_transactions(&mut io), PRIORITY_TIMER => self.sync.process_priority_queue(&mut io), + DELAYED_PROCESSING_TIMER => self.sync.process_delayed_requests(&mut io), _ => warn!("Unknown timer {} triggered.", timer), } } diff --git a/ethcore/sync/src/block_sync.rs b/ethcore/sync/src/block_sync.rs index 2a65e7ba265..a282149a16a 100644 --- a/ethcore/sync/src/block_sync.rs +++ b/ethcore/sync/src/block_sync.rs @@ -231,6 +231,19 @@ impl BlockDownloader { self.state = State::Blocks; } + /// Reset sync to the specified block + fn reset_to_block(&mut self, start_hash: &H256, start_number: BlockNumber) { + self.reset(); + self.last_imported_block = start_number; + self.last_imported_hash = start_hash.clone(); + self.last_round_start = start_number; + self.last_round_start_hash = start_hash.clone(); + self.imported_this_round = None; + self.round_parents = VecDeque::new(); + self.target_hash = None; + self.retract_step = 1; + } + /// Returns best imported block number. pub fn last_imported_block_number(&self) -> BlockNumber { self.last_imported_block @@ -439,22 +452,28 @@ impl BlockDownloader { trace_sync!(self, "Searching common header from the last round {} ({})", self.last_imported_block, self.last_imported_hash); } else { let best = io.chain().chain_info().best_block_number; + let best_hash = io.chain().chain_info().best_block_hash; let oldest_reorg = io.chain().pruning_info().earliest_state; if self.block_set == BlockSet::NewBlocks && best > start && start < oldest_reorg { debug_sync!(self, "Could not revert to previous ancient block, last: {} ({})", start, start_hash); - self.reset(); + self.reset_to_block(&best_hash, best); } else { let n = start - cmp::min(self.retract_step, start); - self.retract_step *= 2; - match io.chain().block_hash(BlockId::Number(n)) { - Some(h) => { - self.last_imported_block = n; - self.last_imported_hash = h; - trace_sync!(self, "Searching common header in the blockchain {} ({})", start, self.last_imported_hash); - } - None => { - debug_sync!(self, "Could not revert to previous block, last: {} ({})", start, self.last_imported_hash); - self.reset(); + if n == 0 { + debug_sync!(self, "Header not found, bottom line reached, resetting, last imported: {}", self.last_imported_hash); + self.reset_to_block(&best_hash, best); + } else { + self.retract_step *= 2; + match io.chain().block_hash(BlockId::Number(n)) { + Some(h) => { + self.last_imported_block = n; + self.last_imported_hash = h; + trace_sync!(self, "Searching common header in the blockchain {} ({})", start, self.last_imported_hash); + } + None => { + debug_sync!(self, "Could not revert to previous block, last: {} ({})", start, self.last_imported_hash); + self.reset_to_block(&best_hash, best); + } } } } diff --git a/ethcore/sync/src/chain/handler.rs b/ethcore/sync/src/chain/handler.rs index dae66140aab..3cd53743e7b 100644 --- a/ethcore/sync/src/chain/handler.rs +++ b/ethcore/sync/src/chain/handler.rs @@ -31,7 +31,7 @@ use crate::{ SnapshotDataPacket, SnapshotManifestPacket, StatusPacket, } }, - BlockSet, ChainSync, ForkConfirmation, PacketDecodeError, PeerAsking, PeerInfo, SyncRequester, + BlockSet, ChainSync, ForkConfirmation, PacketProcessError, PeerAsking, PeerInfo, SyncRequester, SyncState, ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_64, MAX_NEW_BLOCK_AGE, MAX_NEW_HASHES, PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_3, PAR_PROTOCOL_VERSION_4, } @@ -114,6 +114,7 @@ impl SyncHandler { debug!(target: "sync", "Disconnected {}", peer_id); sync.clear_peer_download(peer_id); sync.peers.remove(&peer_id); + sync.delayed_requests.retain(|(request_peer_id, _, _)| *request_peer_id != peer_id); sync.active_peers.remove(&peer_id); if sync.state == SyncState::SnapshotManifest { @@ -149,12 +150,6 @@ impl SyncHandler { trace!(target: "sync", "Ignoring new block from unconfirmed peer {}", peer_id); return Ok(()); } - let difficulty: U256 = r.val_at(1)?; - if let Some(ref mut peer) = sync.peers.get_mut(&peer_id) { - if peer.difficulty.map_or(true, |pd| difficulty > pd) { - peer.difficulty = Some(difficulty); - } - } let block = Unverified::from_rlp(r.at(0)?.as_raw().to_vec())?; let hash = block.header.hash(); let number = block.header.number(); @@ -162,10 +157,20 @@ impl SyncHandler { if number > sync.highest_block.unwrap_or(0) { sync.highest_block = Some(number); } + let parent_hash = block.header.parent_hash(); + let difficulty: U256 = r.val_at(1)?; + // Most probably the sent block is being imported by peer right now + // Use td and hash, that peer must have for now + let parent_td = difficulty.checked_sub(*block.header.difficulty()); + if let Some(ref mut peer) = sync.peers.get_mut(&peer_id) { + if peer.difficulty.map_or(true, |pd| parent_td.map_or(false, |td| td > pd)) { + peer.difficulty = parent_td; + } + } let mut unknown = false; if let Some(ref mut peer) = sync.peers.get_mut(&peer_id) { - peer.latest_hash = hash; + peer.latest_hash = *parent_hash; } let last_imported_number = sync.new_blocks.last_imported_block_number(); @@ -675,7 +680,7 @@ impl SyncHandler { } /// Called when peer sends us new transactions - pub fn on_peer_transactions(sync: &ChainSync, io: &mut dyn SyncIo, peer_id: PeerId, tx_rlp: Rlp) -> Result<(), PacketDecodeError> { + pub fn on_peer_transactions(sync: &ChainSync, io: &mut dyn SyncIo, peer_id: PeerId, tx_rlp: Rlp) -> Result<(), PacketProcessError> { // Accept transactions only when fully synced if !io.is_chain_queue_empty() || (sync.state != SyncState::Idle && sync.state != SyncState::NewBlocks) { trace!(target: "sync", "{} Ignoring transactions while syncing", peer_id); diff --git a/ethcore/sync/src/chain/mod.rs b/ethcore/sync/src/chain/mod.rs index 98627c4d1b5..d7bce672c79 100644 --- a/ethcore/sync/src/chain/mod.rs +++ b/ethcore/sync/src/chain/mod.rs @@ -113,13 +113,14 @@ use crate::{ use bytes::Bytes; use client_traits::BlockChainClient; +use derive_more::Display; use ethereum_types::{H256, U256}; use fastmap::{H256FastMap, H256FastSet}; use futures::sync::mpsc as futures_mpsc; use keccak_hash::keccak; use log::{error, trace, debug, warn}; use network::client_version::ClientVersion; -use network::{self, PeerId, PacketId}; +use network::{self, PeerId}; use parity_util_mem::{MallocSizeOfExt, malloc_size_of_is_0}; use parking_lot::{Mutex, RwLock, RwLockWriteGuard}; use rand::{Rng, seq::SliceRandom}; @@ -147,7 +148,23 @@ pub(crate) use self::supplier::SyncSupplier; malloc_size_of_is_0!(PeerInfo); -pub type PacketDecodeError = DecoderError; +/// Possible errors during packet's processing +#[derive(Debug, Display)] +pub enum PacketProcessError { + /// Error of RLP decoder + #[display(fmt = "Decoder Error: {}", _0)] + Decoder(DecoderError), + /// Underlying client is busy and cannot process the packet + /// The packet should be postponed for later response + #[display(fmt = "Underlying client is busy")] + ClientBusy, +} + +impl From for PacketProcessError { + fn from(err: DecoderError) -> Self { + PacketProcessError::Decoder(err).into() + } +} /// Version 64 of the Ethereum protocol and number of packet IDs reserved by the protocol (packet count). pub const ETH_PROTOCOL_VERSION_64: (u8, u8) = (64, 0x11); @@ -411,7 +428,7 @@ pub mod random { } } -pub type RlpResponseResult = Result, PacketDecodeError>; +pub type RlpResponseResult = Result, PacketProcessError>; pub type Peers = HashMap; /// Thread-safe wrapper for `ChainSync`. @@ -468,6 +485,17 @@ impl ChainSyncApi { SyncSupplier::dispatch_packet(&self.sync, io, peer, packet_id, data) } + /// Process the queue with requests, that were delayed with response. + pub fn process_delayed_requests(&self, io: &mut dyn SyncIo) { + let requests = self.sync.write().retrieve_delayed_requests(); + if !requests.is_empty() { + debug!(target: "sync", "Processing {} delayed requests", requests.len()); + for (peer_id, packet_id, packet_data) in requests { + SyncSupplier::dispatch_delayed_request(&self.sync, io, peer_id, packet_id, &packet_data); + } + } + } + /// Process a priority propagation queue. /// This task is run from a timer and should be time constrained. /// Hence we set up a deadline for the execution and cancel the task if the deadline is exceeded. @@ -672,6 +700,10 @@ pub struct ChainSync { /// Connected peers pending Status message. /// Value is request timestamp. handshaking_peers: HashMap, + /// Requests, that can not be processed at the moment + delayed_requests: Vec<(PeerId, u8, Vec)>, + /// Ids of delayed requests, used for lookup, id is composed from peer id and packet id + delayed_requests_ids: HashSet<(PeerId, u8)>, /// Sync start timestamp. Measured when first peer is connected sync_start_time: Option, /// Transactions propagation statistics @@ -707,6 +739,8 @@ impl ChainSync { peers: HashMap::new(), handshaking_peers: HashMap::new(), active_peers: HashSet::new(), + delayed_requests: Vec::new(), + delayed_requests_ids: HashSet::new(), new_blocks: BlockDownloader::new(BlockSet::NewBlocks, &chain_info.best_block_hash, chain_info.best_block_number), old_blocks: None, last_sent_block_number: 0, @@ -821,6 +855,22 @@ impl ChainSync { self.active_peers = self.peers.keys().cloned().collect(); } + /// Add a request for later processing + pub fn add_delayed_request(&mut self, peer: PeerId, packet_id: u8, data: &[u8]) { + // Ignore the request, if there is a request already in queue with the same id + if !self.delayed_requests_ids.contains(&(peer, packet_id)) { + self.delayed_requests_ids.insert((peer, packet_id)); + self.delayed_requests.push((peer, packet_id, data.to_vec())); + debug!(target: "sync", "Delayed request with packet id {} from peer {} added", packet_id, peer); + } + } + + /// Drain and return all delayed requests + pub fn retrieve_delayed_requests(&mut self) -> Vec<(PeerId, u8, Vec)> { + self.delayed_requests_ids.clear(); + self.delayed_requests.drain(..).collect() + } + /// Restart sync pub fn reset_and_continue(&mut self, io: &mut dyn SyncIo) { trace!(target: "sync", "Restarting"); @@ -1261,7 +1311,7 @@ impl ChainSync { packet.append(&chain.total_difficulty); packet.append(&chain.best_block_hash); packet.append(&chain.genesis_hash); - if eth_protocol_version >= ETH_PROTOCOL_VERSION_64.0 { + if eth_protocol_version >= ETH_PROTOCOL_VERSION_64.0 { packet.append(&self.fork_filter.current(io.chain())); } if warp_protocol { diff --git a/ethcore/sync/src/chain/supplier.rs b/ethcore/sync/src/chain/supplier.rs index a835dedf4e0..95da5910bc6 100644 --- a/ethcore/sync/src/chain/supplier.rs +++ b/ethcore/sync/src/chain/supplier.rs @@ -53,7 +53,7 @@ use super::{ ChainSync, SyncHandler, RlpResponseResult, - PacketDecodeError, + PacketProcessError, MAX_BODIES_TO_SEND, MAX_HEADERS_TO_SEND, MAX_NODE_DATA_TO_SEND, @@ -145,14 +145,47 @@ impl SyncSupplier { } }; - result.unwrap_or_else(|e| { - debug!(target:"sync", "{} -> Malformed packet {} : {}", peer, packet_id, e); - }) + match result { + Err(PacketProcessError::Decoder(e)) => debug!(target:"sync", "{} -> Malformed packet {} : {}", peer, packet_id, e), + Err(PacketProcessError::ClientBusy) => sync.write().add_delayed_request(peer, packet_id, data), + Ok(()) => {} + } + } + } + + /// Dispatch delayed request + /// The main difference with dispatch packet is the direct send of the responses to the peer + pub fn dispatch_delayed_request(sync: &RwLock, io: &mut dyn SyncIo, peer: PeerId, packet_id: u8, data: &[u8]) { + let rlp = Rlp::new(data); + + if let Some(id) = SyncPacket::from_u8(packet_id) { + let result = match id { + GetBlockHeadersPacket => SyncSupplier::send_rlp( + io, &rlp, peer, + SyncSupplier::return_block_headers, + |e| format!("Error sending block headers: {:?}", e)), + + _ => { + debug!(target:"sync", "Unexpected packet {} was dispatched for delayed processing", packet_id); + Ok(()) + } + }; + + match result { + Err(PacketProcessError::Decoder(e)) => debug!(target:"sync", "{} -> Malformed packet {} : {}", peer, packet_id, e), + Err(PacketProcessError::ClientBusy) => sync.write().add_delayed_request(peer, packet_id, data), + Ok(()) => {} + } } } /// Respond to GetBlockHeaders request fn return_block_headers(io: &dyn SyncIo, r: &Rlp, peer_id: PeerId) -> RlpResponseResult { + // Cannot return blocks, if forks processing is in progress, + // The request should be postponed for later processing + if io.chain().is_processing_fork() { + return Err(PacketProcessError::ClientBusy); + } let payload_soft_limit = io.payload_soft_limit(); // Packet layout: // [ block: { P , B_32 }, maxHeaders: P, skip: P, reverse: P in { 0 , 1 } ] @@ -175,11 +208,11 @@ impl SyncSupplier { trace!(target:"sync", "Returning single header: {:?}", hash); let mut rlp = RlpStream::new_list(1); rlp.append_raw(&hdr.into_inner(), 1); - return Ok(Some((BlockHeadersPacket.id(), rlp))); + return Ok(Some((BlockHeadersPacket, rlp))); } number } - None => return Ok(Some((BlockHeadersPacket.id(), RlpStream::new_list(0)))) //no such header, return nothing + None => return Ok(Some((BlockHeadersPacket, RlpStream::new_list(0)))) //no such header, return nothing } } else { let number = r.val_at::(0)?; @@ -229,7 +262,7 @@ impl SyncSupplier { let mut rlp = RlpStream::new_list(count as usize); rlp.append_raw(&data, count as usize); trace!(target: "sync", "{} -> GetBlockHeaders: returned {} entries", peer_id, count); - Ok(Some((BlockHeadersPacket.id(), rlp))) + Ok(Some((BlockHeadersPacket, rlp))) } /// Respond to GetBlockBodies request @@ -256,7 +289,7 @@ impl SyncSupplier { let mut rlp = RlpStream::new_list(added); rlp.append_raw(&data, added); trace!(target: "sync", "{} -> GetBlockBodies: returned {} entries", peer_id, added); - Ok(Some((BlockBodiesPacket.id(), rlp))) + Ok(Some((BlockBodiesPacket, rlp))) } /// Respond to GetNodeData request @@ -302,7 +335,7 @@ impl SyncSupplier { for d in data { rlp.append(&d); } - Ok(Some((NodeDataPacket.id(), rlp))) + Ok(Some((NodeDataPacket, rlp))) } fn return_receipts(io: &dyn SyncIo, rlp: &Rlp, peer_id: PeerId) -> RlpResponseResult { @@ -328,7 +361,7 @@ impl SyncSupplier { } let mut rlp_result = RlpStream::new_list(added_headers); rlp_result.append_raw(&data, added_headers); - Ok(Some((ReceiptsPacket.id(), rlp_result))) + Ok(Some((ReceiptsPacket, rlp_result))) } /// Respond to GetSnapshotManifest request @@ -351,7 +384,7 @@ impl SyncSupplier { RlpStream::new_list(0) } }; - Ok(Some((SnapshotManifestPacket.id(), rlp))) + Ok(Some((SnapshotManifestPacket, rlp))) } /// Respond to GetSnapshotData request @@ -370,7 +403,7 @@ impl SyncSupplier { RlpStream::new_list(0) } }; - Ok(Some((SnapshotDataPacket.id(), rlp))) + Ok(Some((SnapshotDataPacket, rlp))) } /// Respond to GetPrivateStatePacket @@ -387,13 +420,25 @@ impl SyncSupplier { Ok(bytes) => { let mut rlp = RlpStream::new_list(1); rlp.append(&bytes); - Ok(Some((PrivateStatePacket.id(), rlp))) + Ok(Some((PrivateStatePacket, rlp))) } } }) } - fn return_rlp(io: &mut dyn SyncIo, rlp: &Rlp, peer: PeerId, rlp_func: FRlp, error_func: FError) -> Result<(), PacketDecodeError> + fn return_rlp(io: &mut dyn SyncIo, rlp: &Rlp, peer: PeerId, rlp_func: FRlp, error_func: FError) -> Result<(), PacketProcessError> + where FRlp : Fn(&dyn SyncIo, &Rlp, PeerId) -> RlpResponseResult, + FError : FnOnce(network::Error) -> String + { + let response = rlp_func(io, rlp, peer); + if let Some((packet_id, rlp_stream)) = response? { + io.respond(packet_id.id(), rlp_stream.out()).unwrap_or_else( + |e| debug!(target: "sync", "{:?}", error_func(e))); + } + Ok(()) + } + + fn send_rlp(io: &mut dyn SyncIo, rlp: &Rlp, peer: PeerId, rlp_func: FRlp, error_func: FError) -> Result<(), PacketProcessError> where FRlp : Fn(&dyn SyncIo, &Rlp, PeerId) -> RlpResponseResult, FError : FnOnce(network::Error) -> String { @@ -401,7 +446,7 @@ impl SyncSupplier { match response { Err(e) => Err(e), Ok(Some((packet_id, rlp_stream))) => { - io.respond(packet_id, rlp_stream.out()).unwrap_or_else( + io.send(peer, packet_id, rlp_stream.out()).unwrap_or_else( |e| debug!(target: "sync", "{:?}", error_func(e))); Ok(()) } diff --git a/ethcore/verification/src/queue/mod.rs b/ethcore/verification/src/queue/mod.rs index 194311c1fae..fb9ba8e069e 100644 --- a/ethcore/verification/src/queue/mod.rs +++ b/ethcore/verification/src/queue/mod.rs @@ -28,6 +28,7 @@ use common_types::{ errors::{BlockError, EthcoreError as Error, ImportError}, verification::VerificationQueueInfo as QueueInfo, }; +use blockchain::BlockChain; use ethcore_io::*; use ethereum_types::{H256, U256}; use engine::Engine; @@ -42,6 +43,9 @@ pub mod kind; const MIN_MEM_LIMIT: usize = 16384; const MIN_QUEUE_LIMIT: usize = 512; +/// Empiric estimation of the minimal length of the processing queue, +/// That definitely doesn't contain forks inside. +const MAX_QUEUE_WITH_FORK: usize = 8; /// Type alias for block queue convenience. pub type BlockQueue = VerificationQueue; @@ -141,7 +145,7 @@ pub struct VerificationQueue { deleting: Arc, ready_signal: Arc>, empty: Arc, - processing: RwLock>, // hash to difficulty + processing: RwLock>, // item's hash to difficulty and parent item hash ticks_since_adjustment: AtomicUsize, max_queue_size: usize, max_mem_use: usize, @@ -490,7 +494,7 @@ impl VerificationQueue { match K::create(input, &*self.engine, self.verification.check_seal) { Ok(item) => { - if self.processing.write().insert(hash, item.difficulty()).is_some() { + if self.processing.write().insert(hash, (item.difficulty(), item.parent_hash())).is_some() { return Err((Error::Import(ImportError::AlreadyQueued), None)); } self.verification.sizes.unverified.fetch_add(item.malloc_size_of(), AtomicOrdering::SeqCst); @@ -538,7 +542,7 @@ impl VerificationQueue { bad.reserve(hashes.len()); for hash in hashes { bad.insert(hash.clone()); - if let Some(difficulty) = processing.remove(hash) { + if let Some((difficulty, _)) = processing.remove(hash) { let mut td = self.total_difficulty.write(); *td = *td - difficulty; } @@ -550,7 +554,7 @@ impl VerificationQueue { if bad.contains(&output.parent_hash()) { removed_size += output.malloc_size_of(); bad.insert(output.hash()); - if let Some(difficulty) = processing.remove(&output.hash()) { + if let Some((difficulty, _)) = processing.remove(&output.hash()) { let mut td = self.total_difficulty.write(); *td = *td - difficulty; } @@ -571,7 +575,7 @@ impl VerificationQueue { } let mut processing = self.processing.write(); for hash in hashes { - if let Some(difficulty) = processing.remove(hash) { + if let Some((difficulty, _)) = processing.remove(hash) { let mut td = self.total_difficulty.write(); *td = *td - difficulty; } @@ -604,6 +608,21 @@ impl VerificationQueue { && v.verified.load_len() == 0 } + /// Returns true if there are descendants of the current best block in the processing queue + pub fn is_processing_fork(&self, best_block_hash: &H256, chain: &BlockChain) -> bool { + let processing = self.processing.read(); + if processing.is_empty() || processing.len() > MAX_QUEUE_WITH_FORK { + // Assume, that long enough processing queue doesn't have fork blocks + return false; + } + for (_, item_parent_hash) in processing.values() { + if chain.tree_route(*best_block_hash, *item_parent_hash).map_or(true, |route| route.ancestor != *best_block_hash) { + return true; + } + } + false + } + /// Get queue status. pub fn queue_info(&self) -> QueueInfo { use std::mem::size_of;