Cache the fork block header after snapshot restoration (#2391)

This commit is contained in:
Arkadiy Paronyan 2016-09-29 12:34:40 +02:00 committed by GitHub
parent 4d115987cb
commit 0e55b6c6c9
4 changed files with 48 additions and 14 deletions

View File

@ -15,9 +15,10 @@
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
use std::sync::Arc; use std::sync::Arc;
use std::collections::HashMap;
use network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId, use network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId,
NetworkConfiguration as BasicNetworkConfiguration, NonReservedPeerMode, NetworkError}; NetworkConfiguration as BasicNetworkConfiguration, NonReservedPeerMode, NetworkError};
use util::{U256, H256, Secret, Populatable}; use util::{U256, H256, Secret, Populatable, Bytes};
use io::{TimerToken}; use io::{TimerToken};
use ethcore::client::{BlockChainClient, ChainNotify}; use ethcore::client::{BlockChainClient, ChainNotify};
use ethcore::header::BlockNumber; use ethcore::header::BlockNumber;
@ -78,7 +79,11 @@ impl EthSync {
let service = try!(NetworkService::new(try!(network_config.into_basic()))); let service = try!(NetworkService::new(try!(network_config.into_basic())));
let sync = Arc::new(EthSync{ let sync = Arc::new(EthSync{
network: service, network: service,
handler: Arc::new(SyncProtocolHandler { sync: RwLock::new(chain_sync), chain: chain }), handler: Arc::new(SyncProtocolHandler {
sync: RwLock::new(chain_sync),
chain: chain,
overlay: RwLock::new(HashMap::new()),
}),
}); });
Ok(sync) Ok(sync)
@ -99,6 +104,8 @@ struct SyncProtocolHandler {
chain: Arc<BlockChainClient>, chain: Arc<BlockChainClient>,
/// Sync strategy /// Sync strategy
sync: RwLock<ChainSync>, sync: RwLock<ChainSync>,
/// Chain overlay used to cache data such as fork block.
overlay: RwLock<HashMap<BlockNumber, Bytes>>,
} }
impl NetworkProtocolHandler for SyncProtocolHandler { impl NetworkProtocolHandler for SyncProtocolHandler {
@ -107,20 +114,20 @@ impl NetworkProtocolHandler for SyncProtocolHandler {
} }
fn read(&self, io: &NetworkContext, peer: &PeerId, packet_id: u8, data: &[u8]) { fn read(&self, io: &NetworkContext, peer: &PeerId, packet_id: u8, data: &[u8]) {
ChainSync::dispatch_packet(&self.sync, &mut NetSyncIo::new(io, &*self.chain), *peer, packet_id, data); ChainSync::dispatch_packet(&self.sync, &mut NetSyncIo::new(io, &*self.chain, &self.overlay), *peer, packet_id, data);
} }
fn connected(&self, io: &NetworkContext, peer: &PeerId) { fn connected(&self, io: &NetworkContext, peer: &PeerId) {
self.sync.write().on_peer_connected(&mut NetSyncIo::new(io, &*self.chain), *peer); self.sync.write().on_peer_connected(&mut NetSyncIo::new(io, &*self.chain, &self.overlay), *peer);
} }
fn disconnected(&self, io: &NetworkContext, peer: &PeerId) { fn disconnected(&self, io: &NetworkContext, peer: &PeerId) {
self.sync.write().on_peer_aborting(&mut NetSyncIo::new(io, &*self.chain), *peer); self.sync.write().on_peer_aborting(&mut NetSyncIo::new(io, &*self.chain, &self.overlay), *peer);
} }
fn timeout(&self, io: &NetworkContext, _timer: TimerToken) { fn timeout(&self, io: &NetworkContext, _timer: TimerToken) {
self.sync.write().maintain_peers(&mut NetSyncIo::new(io, &*self.chain)); self.sync.write().maintain_peers(&mut NetSyncIo::new(io, &*self.chain, &self.overlay));
self.sync.write().maintain_sync(&mut NetSyncIo::new(io, &*self.chain)); self.sync.write().maintain_sync(&mut NetSyncIo::new(io, &*self.chain, &self.overlay));
} }
} }
@ -134,7 +141,7 @@ impl ChainNotify for EthSync {
_duration: u64) _duration: u64)
{ {
self.network.with_context(ETH_PROTOCOL, |context| { self.network.with_context(ETH_PROTOCOL, |context| {
let mut sync_io = NetSyncIo::new(context, &*self.handler.chain); let mut sync_io = NetSyncIo::new(context, &*self.handler.chain, &self.handler.overlay);
self.handler.sync.write().chain_new_blocks( self.handler.sync.write().chain_new_blocks(
&mut sync_io, &mut sync_io,
&imported, &imported,
@ -203,7 +210,7 @@ impl ManageNetwork for EthSync {
fn stop_network(&self) { fn stop_network(&self) {
self.network.with_context(ETH_PROTOCOL, |context| { self.network.with_context(ETH_PROTOCOL, |context| {
let mut sync_io = NetSyncIo::new(context, &*self.handler.chain); let mut sync_io = NetSyncIo::new(context, &*self.handler.chain, &self.handler.overlay);
self.handler.sync.write().abort(&mut sync_io); self.handler.sync.write().abort(&mut sync_io);
}); });
self.stop(); self.stop();

View File

@ -449,7 +449,9 @@ impl ChainSync {
let confirmed = match self.peers.get_mut(&peer_id) { let confirmed = match self.peers.get_mut(&peer_id) {
Some(ref mut peer) if peer.asking == PeerAsking::ForkHeader => { Some(ref mut peer) if peer.asking == PeerAsking::ForkHeader => {
let item_count = r.item_count(); let item_count = r.item_count();
if item_count == 0 || (item_count == 1 && try!(r.at(0)).as_raw().sha3() == self.fork_block.unwrap().1) { let (fork_number, fork_hash) = self.fork_block.expect("ForkHeader request is sent only fork block is Some; qed").clone();
let header = try!(r.at(0)).as_raw();
if item_count == 0 || (item_count == 1 && header.sha3() == fork_hash) {
peer.asking = PeerAsking::Nothing; peer.asking = PeerAsking::Nothing;
if item_count == 0 { if item_count == 0 {
trace!(target: "sync", "{}: Chain is too short to confirm the block", peer_id); trace!(target: "sync", "{}: Chain is too short to confirm the block", peer_id);
@ -457,6 +459,9 @@ impl ChainSync {
} else { } else {
trace!(target: "sync", "{}: Confirmed peer", peer_id); trace!(target: "sync", "{}: Confirmed peer", peer_id);
peer.confirmation = ForkConfirmation::Confirmed; peer.confirmation = ForkConfirmation::Confirmed;
if !io.chain_overlay().read().contains_key(&fork_number) {
io.chain_overlay().write().insert(fork_number, header.to_vec());
}
} }
true true
} else { } else {
@ -1135,8 +1140,13 @@ impl ChainSync {
let mut count = 0; let mut count = 0;
let mut data = Bytes::new(); let mut data = Bytes::new();
let inc = (skip + 1) as BlockNumber; let inc = (skip + 1) as BlockNumber;
let overlay = io.chain_overlay().read();
while number <= last && count < max_count { while number <= last && count < max_count {
if let Some(mut hdr) = io.chain().block_header(BlockID::Number(number)) { if let Some(hdr) = overlay.get(&number) {
trace!(target: "sync", "{}: Returning cached fork header", peer_id);
data.extend(hdr);
count += 1;
} else if let Some(mut hdr) = io.chain().block_header(BlockID::Number(number)) {
data.append(&mut hdr); data.append(&mut hdr);
count += 1; count += 1;
} }

View File

@ -14,8 +14,11 @@
// You should have received a copy of the GNU General Public License // You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
use std::collections::HashMap;
use ethcore::header::BlockNumber;
use network::{NetworkContext, PeerId, PacketId, NetworkError}; use network::{NetworkContext, PeerId, PacketId, NetworkError};
use ethcore::client::BlockChainClient; use ethcore::client::BlockChainClient;
use util::{RwLock, Bytes};
/// IO interface for the syning handler. /// IO interface for the syning handler.
/// Provides peer connection management and an interface to the blockchain client. /// Provides peer connection management and an interface to the blockchain client.
@ -41,20 +44,24 @@ pub trait SyncIo {
} }
/// Check if the session is expired /// Check if the session is expired
fn is_expired(&self) -> bool; fn is_expired(&self) -> bool;
/// Return sync overlay
fn chain_overlay(&self) -> &RwLock<HashMap<BlockNumber, Bytes>>;
} }
/// Wraps `NetworkContext` and the blockchain client /// Wraps `NetworkContext` and the blockchain client
pub struct NetSyncIo<'s, 'h> where 'h: 's { pub struct NetSyncIo<'s, 'h> where 'h: 's {
network: &'s NetworkContext<'h>, network: &'s NetworkContext<'h>,
chain: &'s BlockChainClient chain: &'s BlockChainClient,
chain_overlay: &'s RwLock<HashMap<BlockNumber, Bytes>>,
} }
impl<'s, 'h> NetSyncIo<'s, 'h> { impl<'s, 'h> NetSyncIo<'s, 'h> {
/// Creates a new instance from the `NetworkContext` and the blockchain client reference. /// Creates a new instance from the `NetworkContext` and the blockchain client reference.
pub fn new(network: &'s NetworkContext<'h>, chain: &'s BlockChainClient) -> NetSyncIo<'s, 'h> { pub fn new(network: &'s NetworkContext<'h>, chain: &'s BlockChainClient, chain_overlay: &'s RwLock<HashMap<BlockNumber, Bytes>>) -> NetSyncIo<'s, 'h> {
NetSyncIo { NetSyncIo {
network: network, network: network,
chain: chain, chain: chain,
chain_overlay: chain_overlay,
} }
} }
} }
@ -80,6 +87,10 @@ impl<'s, 'h> SyncIo for NetSyncIo<'s, 'h> {
self.chain self.chain
} }
fn chain_overlay(&self) -> &RwLock<HashMap<BlockNumber, Bytes>> {
self.chain_overlay
}
fn peer_info(&self, peer_id: PeerId) -> String { fn peer_info(&self, peer_id: PeerId) -> String {
self.network.peer_info(peer_id) self.network.peer_info(peer_id)
} }

View File

@ -26,6 +26,7 @@ pub struct TestIo<'p> {
pub chain: &'p mut TestBlockChainClient, pub chain: &'p mut TestBlockChainClient,
pub queue: &'p mut VecDeque<TestPacket>, pub queue: &'p mut VecDeque<TestPacket>,
pub sender: Option<PeerId>, pub sender: Option<PeerId>,
overlay: RwLock<HashMap<BlockNumber, Bytes>>,
} }
impl<'p> TestIo<'p> { impl<'p> TestIo<'p> {
@ -33,7 +34,8 @@ impl<'p> TestIo<'p> {
TestIo { TestIo {
chain: chain, chain: chain,
queue: queue, queue: queue,
sender: sender sender: sender,
overlay: RwLock::new(HashMap::new()),
} }
} }
} }
@ -70,6 +72,10 @@ impl<'p> SyncIo for TestIo<'p> {
fn chain(&self) -> &BlockChainClient { fn chain(&self) -> &BlockChainClient {
self.chain self.chain
} }
fn chain_overlay(&self) -> &RwLock<HashMap<BlockNumber, Bytes>> {
&self.overlay
}
} }
pub struct TestPacket { pub struct TestPacket {