grassrootseconomics/openethereum
5
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix not downloading old blocks (#8642)

Browse Source
This commit is contained in:
Nicolas Gotchac 2018-05-17 10:58:35 +02:00 committed by Afri Schoedon
parent cdbcfaa7de
commit d1934363e7
2 changed files with 40 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
ethcore/src/client/client.rs
View File

@ -17,7 +17,7 @@
use std::collections::{HashSet, BTreeMap, BTreeSet, VecDeque};
use std::fmt;
use std::str::FromStr;
use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering};
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use std::sync::{Arc, Weak};
use std::time::{Instant, Duration};
@ -210,6 +210,8 @@ pub struct Client {
queue_transactions: IoChannelQueue,
/// Ancient blocks import queue
queue_ancient_blocks: IoChannelQueue,
/// Hashes of pending ancient block wainting to be included
pending_ancient_blocks: RwLock<HashSet<H256>>,
/// Consensus messages import queue
queue_consensus_message: IoChannelQueue,
@ -433,6 +435,7 @@ impl Importer {
let hash = header.hash();
let _import_lock = self.import_lock.lock();
trace!(target: "client", "Trying to import old block #{}", header.number());
{
trace_time!("import_old_block");
// verify the block, passing the chain for updating the epoch verifier.
@ -761,6 +764,7 @@ impl Client {
notify: RwLock::new(Vec::new()),
queue_transactions: IoChannelQueue::new(MAX_TX_QUEUE_SIZE),
queue_ancient_blocks: IoChannelQueue::new(MAX_ANCIENT_BLOCKS_QUEUE_SIZE),
pending_ancient_blocks: RwLock::new(HashSet::new()),
queue_consensus_message: IoChannelQueue::new(usize::max_value()),
last_hashes: RwLock::new(VecDeque::new()),
factories: factories,
@ -2008,7 +2012,7 @@ impl BlockChainClient for Client {
impl IoClient for Client {
fn queue_transactions(&self, transactions: Vec<Bytes>, peer_id: usize) {
let len = transactions.len();
self.queue_transactions.queue(&mut self.io_channel.lock(), len, move |client| {
self.queue_transactions.queue(&mut self.io_channel.lock(), move |client| {
trace_time!("import_queued_transactions");
let txs: Vec<UnverifiedTransaction> = transactions
@ -2032,23 +2036,32 @@ impl IoClient for Client {
{
// check block order
if self.chain.read().is_known(&header.hash()) {
if self.chain.read().is_known(&hash) {
bail!(BlockImportErrorKind::Import(ImportErrorKind::AlreadyInChain));
}
let status = self.block_status(BlockId::Hash(*header.parent_hash()));
if status == BlockStatus::Unknown || status == BlockStatus::Pending {
bail!(BlockImportErrorKind::Block(BlockError::UnknownParent(*header.parent_hash())));
let parent_hash = *header.parent_hash();
let parent_pending = self.pending_ancient_blocks.read().contains(&parent_hash);
let status = self.block_status(BlockId::Hash(parent_hash));
if !parent_pending && (status == BlockStatus::Unknown || status == BlockStatus::Pending) {
bail!(BlockImportErrorKind::Block(BlockError::UnknownParent(parent_hash)));
}
}
match self.queue_ancient_blocks.queue(&mut self.io_channel.lock(), 1, move |client| {
client.importer.import_old_block(
self.pending_ancient_blocks.write().insert(hash);
trace!(target: "client", "Queuing old block #{}", header.number());
match self.queue_ancient_blocks.queue(&mut self.io_channel.lock(), move |client| {
let result = client.importer.import_old_block(
&header,
&block_bytes,
&receipts_bytes,
&**client.db.read(),
&*client.chain.read()
).map(|_| ()).unwrap_or_else(|e| {
);
client.pending_ancient_blocks.write().remove(&hash);
result.map(|_| ()).unwrap_or_else(|e| {
error!(target: "client", "Error importing ancient block: {}", e);
});
}) {
@ -2058,7 +2071,7 @@ impl IoClient for Client {
}
fn queue_consensus_message(&self, message: Bytes) {
match self.queue_consensus_message.queue(&mut self.io_channel.lock(), 1, move |client| {
match self.queue_consensus_message.queue(&mut self.io_channel.lock(), move |client| {
if let Err(e) = client.engine().handle_message(&message) {
debug!(target: "poa", "Invalid message received: {}", e);
}
@ -2471,35 +2484,38 @@ impl fmt::Display for QueueError {
/// Queue some items to be processed by IO client.
struct IoChannelQueue {
currently_queued: Arc<AtomicUsize>,
queue: Arc<Mutex<VecDeque<Box<Fn(&Client) + Send>>>>,
limit: usize,
}
impl IoChannelQueue {
pub fn new(limit: usize) -> Self {
IoChannelQueue {
currently_queued: Default::default(),
queue: Default::default(),
limit,
}
}
pub fn queue<F>(&self, channel: &mut IoChannel<ClientIoMessage>, count: usize, fun: F) -> Result<(), QueueError> where
F: Fn(&Client) + Send + Sync + 'static,
pub fn queue<F>(&self, channel: &mut IoChannel<ClientIoMessage>, fun: F) -> Result<(), QueueError>
where F: Fn(&Client) + Send + Sync + 'static
{
let queue_size = self.currently_queued.load(AtomicOrdering::Relaxed);
ensure!(queue_size < self.limit, QueueError::Full(self.limit));
{
let mut queue = self.queue.lock();
let queue_size = queue.len();
ensure!(queue_size < self.limit, QueueError::Full(self.limit));
let currently_queued = self.currently_queued.clone();
queue.push_back(Box::new(fun));
}
let queue = self.queue.clone();
let result = channel.send(ClientIoMessage::execute(move |client| {
currently_queued.fetch_sub(count, AtomicOrdering::SeqCst);
fun(client);
while let Some(fun) = queue.lock().pop_front() {
fun(client);
}
}));
match result {
Ok(_) => {
self.currently_queued.fetch_add(count, AtomicOrdering::SeqCst);
Ok(())
},
Ok(_) => Ok(()),
Err(e) => Err(QueueError::Channel(e)),
}
}

2
ethcore/sync/src/blocks.rs
View File

@ -266,7 +266,7 @@ impl BlockCollection {
}
}
/// Get a valid chain of blocks ordered in descending order and ready for importing into blockchain.
/// Get a valid chain of blocks ordered in ascending order and ready for importing into blockchain.
pub fn drain(&mut self) -> Vec<BlockAndReceipts> {
if self.blocks.is_empty() || self.head.is_none() {
return Vec::new();