overhaul to flag
This commit is contained in:
parent
83f5cc6aa6
commit
a285fbab6d
@ -324,6 +324,7 @@ impl<V> Client<V> where V: Verifier {
|
|||||||
invalid: invalid_blocks,
|
invalid: invalid_blocks,
|
||||||
enacted: enacted,
|
enacted: enacted,
|
||||||
retracted: retracted,
|
retracted: retracted,
|
||||||
|
is_last: self.queue_info().is_empty()
|
||||||
})).unwrap();
|
})).unwrap();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -334,12 +335,6 @@ impl<V> Client<V> where V: Verifier {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
{
|
|
||||||
if self.queue_info().is_empty() {
|
|
||||||
io.send(NetworkIoMessage::User(SyncMessage::BlockQueueEmpty)).expect("error sending message to sync module");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
imported
|
imported
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -35,13 +35,13 @@ pub enum SyncMessage {
|
|||||||
retracted: Vec<H256>,
|
retracted: Vec<H256>,
|
||||||
/// Hashes of blocks that are now included in cannonical chain
|
/// Hashes of blocks that are now included in cannonical chain
|
||||||
enacted: Vec<H256>,
|
enacted: Vec<H256>,
|
||||||
|
/// Set when blockqueue is empty
|
||||||
|
is_last: bool,
|
||||||
},
|
},
|
||||||
/// Best Block Hash in chain has been changed
|
/// Best Block Hash in chain has been changed
|
||||||
NewChainHead,
|
NewChainHead,
|
||||||
/// A block is ready
|
/// A block is ready
|
||||||
BlockVerified,
|
BlockVerified,
|
||||||
/// blocks queue is empty
|
|
||||||
BlockQueueEmpty,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// IO Message type used for Network service
|
/// IO Message type used for Network service
|
||||||
|
@ -151,7 +151,7 @@ impl<C, S, A, M, EM> Eth for EthClient<C, S, A, M, EM>
|
|||||||
Params::None => {
|
Params::None => {
|
||||||
let status = take_weak!(self.sync).status();
|
let status = take_weak!(self.sync).status();
|
||||||
let res = match status.state {
|
let res = match status.state {
|
||||||
SyncState::NotSynced | SyncState::Idle | SyncState::FullySynced => SyncStatus::None,
|
SyncState::NotSynced | SyncState::Idle => SyncStatus::None,
|
||||||
SyncState::Waiting | SyncState::Blocks | SyncState::NewBlocks => SyncStatus::Info(SyncInfo {
|
SyncState::Waiting | SyncState::Blocks | SyncState::NewBlocks => SyncStatus::Info(SyncInfo {
|
||||||
starting_block: U256::from(status.start_block_number),
|
starting_block: U256::from(status.start_block_number),
|
||||||
current_block: U256::from(take_weak!(self.client).chain_info().best_block_number),
|
current_block: U256::from(take_weak!(self.client).chain_info().best_block_number),
|
||||||
|
@ -116,8 +116,6 @@ pub enum SyncState {
|
|||||||
Blocks,
|
Blocks,
|
||||||
/// Downloading blocks learned from NewHashes packet
|
/// Downloading blocks learned from NewHashes packet
|
||||||
NewBlocks,
|
NewBlocks,
|
||||||
/// Once has an event that client finished processing new blocks
|
|
||||||
FullySynced,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Syncing status and statistics
|
/// Syncing status and statistics
|
||||||
@ -217,6 +215,8 @@ pub struct ChainSync {
|
|||||||
network_id: U256,
|
network_id: U256,
|
||||||
/// Miner
|
/// Miner
|
||||||
miner: Arc<Miner>,
|
miner: Arc<Miner>,
|
||||||
|
/// Fully-synced flag
|
||||||
|
is_fully_synced: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
type RlpResponseResult = Result<Option<(PacketId, RlpStream)>, PacketDecodeError>;
|
type RlpResponseResult = Result<Option<(PacketId, RlpStream)>, PacketDecodeError>;
|
||||||
@ -243,6 +243,7 @@ impl ChainSync {
|
|||||||
max_download_ahead_blocks: max(MAX_HEADERS_TO_REQUEST, config.max_download_ahead_blocks),
|
max_download_ahead_blocks: max(MAX_HEADERS_TO_REQUEST, config.max_download_ahead_blocks),
|
||||||
network_id: config.network_id,
|
network_id: config.network_id,
|
||||||
miner: miner,
|
miner: miner,
|
||||||
|
is_fully_synced: true,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -629,7 +630,7 @@ impl ChainSync {
|
|||||||
|
|
||||||
fn can_sync(&self) -> bool {
|
fn can_sync(&self) -> bool {
|
||||||
match self.state {
|
match self.state {
|
||||||
SyncState::Idle | SyncState::NotSynced | SyncState::FullySynced => true,
|
SyncState::Idle | SyncState::NotSynced => true,
|
||||||
_ => false
|
_ => false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -947,7 +948,7 @@ impl ChainSync {
|
|||||||
/// Called when peer sends us new transactions
|
/// Called when peer sends us new transactions
|
||||||
fn on_peer_transactions(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
|
fn on_peer_transactions(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> {
|
||||||
// accepting transactions once only fully synced
|
// accepting transactions once only fully synced
|
||||||
if self.state != SyncState::FullySynced {
|
if !self.is_fully_synced {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1292,8 +1293,10 @@ impl ChainSync {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// called when block is imported to chain, updates transactions queue and propagates the blocks
|
/// called when block is imported to chain, updates transactions queue and propagates the blocks
|
||||||
pub fn chain_new_blocks(&mut self, io: &mut SyncIo, imported: &[H256], invalid: &[H256], enacted: &[H256], retracted: &[H256]) {
|
pub fn chain_new_blocks(&mut self, io: &mut SyncIo, imported: &[H256], invalid: &[H256], enacted: &[H256], retracted: &[H256], is_last: bool) {
|
||||||
if self.state == SyncState::FullySynced {
|
// Set the state in which it can accept transactions from the net
|
||||||
|
self.is_fully_synced = is_last;
|
||||||
|
if self.is_fully_synced {
|
||||||
// Notify miner
|
// Notify miner
|
||||||
self.miner.chain_new_blocks(io.chain(), imported, invalid, enacted, retracted);
|
self.miner.chain_new_blocks(io.chain(), imported, invalid, enacted, retracted);
|
||||||
}
|
}
|
||||||
@ -1303,21 +1306,7 @@ impl ChainSync {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn chain_new_head(&mut self, io: &mut SyncIo) {
|
pub fn chain_new_head(&mut self, io: &mut SyncIo) {
|
||||||
if self.state == SyncState::FullySynced {
|
self.miner.prepare_sealing(io.chain());
|
||||||
self.miner.prepare_sealing(io.chain());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// called once has nothing to download and client reports that all that downloaded is imported
|
|
||||||
pub fn set_fully_synced(&mut self) {
|
|
||||||
self.state = SyncState::FullySynced;
|
|
||||||
}
|
|
||||||
|
|
||||||
// handles event from client about empty blow_queue
|
|
||||||
pub fn client_block_queue_empty(&mut self) {
|
|
||||||
if self.state == SyncState::Idle {
|
|
||||||
self.set_fully_synced();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1654,7 +1643,6 @@ mod tests {
|
|||||||
client.add_blocks(1, EachBlockWith::UncleAndTransaction);
|
client.add_blocks(1, EachBlockWith::UncleAndTransaction);
|
||||||
client.add_blocks(1, EachBlockWith::Transaction);
|
client.add_blocks(1, EachBlockWith::Transaction);
|
||||||
let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
|
let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
|
||||||
sync.state = SyncState::FullySynced;
|
|
||||||
|
|
||||||
let good_blocks = vec![client.block_hash_delta_minus(2)];
|
let good_blocks = vec![client.block_hash_delta_minus(2)];
|
||||||
let retracted_blocks = vec![client.block_hash_delta_minus(1)];
|
let retracted_blocks = vec![client.block_hash_delta_minus(1)];
|
||||||
@ -1663,10 +1651,10 @@ mod tests {
|
|||||||
let mut io = TestIo::new(&mut client, &mut queue, None);
|
let mut io = TestIo::new(&mut client, &mut queue, None);
|
||||||
|
|
||||||
// when
|
// when
|
||||||
sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks);
|
sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks, true);
|
||||||
assert_eq!(sync.miner.status().transaction_queue_future, 0);
|
assert_eq!(sync.miner.status().transaction_queue_future, 0);
|
||||||
assert_eq!(sync.miner.status().transaction_queue_pending, 1);
|
assert_eq!(sync.miner.status().transaction_queue_pending, 1);
|
||||||
sync.chain_new_blocks(&mut io, &good_blocks, &[], &[], &retracted_blocks);
|
sync.chain_new_blocks(&mut io, &good_blocks, &[], &[], &retracted_blocks, true);
|
||||||
|
|
||||||
// then
|
// then
|
||||||
let status = sync.miner.status();
|
let status = sync.miner.status();
|
||||||
@ -1682,7 +1670,6 @@ mod tests {
|
|||||||
client.add_blocks(1, EachBlockWith::UncleAndTransaction);
|
client.add_blocks(1, EachBlockWith::UncleAndTransaction);
|
||||||
client.add_blocks(1, EachBlockWith::Transaction);
|
client.add_blocks(1, EachBlockWith::Transaction);
|
||||||
let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
|
let mut sync = dummy_sync_with_peer(client.block_hash_delta_minus(5));
|
||||||
sync.state = SyncState::Idle;
|
|
||||||
|
|
||||||
let good_blocks = vec![client.block_hash_delta_minus(2)];
|
let good_blocks = vec![client.block_hash_delta_minus(2)];
|
||||||
let retracted_blocks = vec![client.block_hash_delta_minus(1)];
|
let retracted_blocks = vec![client.block_hash_delta_minus(1)];
|
||||||
@ -1691,10 +1678,10 @@ mod tests {
|
|||||||
let mut io = TestIo::new(&mut client, &mut queue, None);
|
let mut io = TestIo::new(&mut client, &mut queue, None);
|
||||||
|
|
||||||
// when
|
// when
|
||||||
sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks);
|
sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks, false);
|
||||||
assert_eq!(sync.miner.status().transaction_queue_future, 0);
|
assert_eq!(sync.miner.status().transaction_queue_future, 0);
|
||||||
assert_eq!(sync.miner.status().transaction_queue_pending, 0);
|
assert_eq!(sync.miner.status().transaction_queue_pending, 0);
|
||||||
sync.chain_new_blocks(&mut io, &good_blocks, &[], &[], &retracted_blocks);
|
sync.chain_new_blocks(&mut io, &good_blocks, &[], &[], &retracted_blocks, false);
|
||||||
|
|
||||||
// then
|
// then
|
||||||
let status = sync.miner.status();
|
let status = sync.miner.status();
|
||||||
|
@ -166,17 +166,14 @@ impl NetworkProtocolHandler<SyncMessage> for EthSync {
|
|||||||
|
|
||||||
fn message(&self, io: &NetworkContext<SyncMessage>, message: &SyncMessage) {
|
fn message(&self, io: &NetworkContext<SyncMessage>, message: &SyncMessage) {
|
||||||
match *message {
|
match *message {
|
||||||
SyncMessage::NewChainBlocks { ref imported, ref invalid, ref enacted, ref retracted } => {
|
SyncMessage::NewChainBlocks { ref imported, ref invalid, ref enacted, ref retracted, is_last } => {
|
||||||
let mut sync_io = NetSyncIo::new(io, self.chain.deref());
|
let mut sync_io = NetSyncIo::new(io, self.chain.deref());
|
||||||
self.sync.write().unwrap().chain_new_blocks(&mut sync_io, imported, invalid, enacted, retracted);
|
self.sync.write().unwrap().chain_new_blocks(&mut sync_io, imported, invalid, enacted, retracted, is_last);
|
||||||
},
|
},
|
||||||
SyncMessage::NewChainHead => {
|
SyncMessage::NewChainHead => {
|
||||||
let mut sync_io = NetSyncIo::new(io, self.chain.deref());
|
let mut sync_io = NetSyncIo::new(io, self.chain.deref());
|
||||||
self.sync.write().unwrap().chain_new_head(&mut sync_io);
|
self.sync.write().unwrap().chain_new_head(&mut sync_io);
|
||||||
},
|
},
|
||||||
SyncMessage::BlockQueueEmpty => {
|
|
||||||
self.sync.write().unwrap().client_block_queue_empty();
|
|
||||||
}
|
|
||||||
_ => {/* Ignore other messages */},
|
_ => {/* Ignore other messages */},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -168,6 +168,6 @@ impl TestNet {
|
|||||||
|
|
||||||
pub fn trigger_chain_new_blocks(&mut self, peer_id: usize) {
|
pub fn trigger_chain_new_blocks(&mut self, peer_id: usize) {
|
||||||
let mut peer = self.peer_mut(peer_id);
|
let mut peer = self.peer_mut(peer_id);
|
||||||
peer.sync.chain_new_blocks(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None), &[], &[], &[], &[]);
|
peer.sync.chain_new_blocks(&mut TestIo::new(&mut peer.chain, &mut peer.queue, None), &[], &[], &[], &[], true);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user