mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-07-11 11:45:44 +00:00
Networking and in-memory client (#38)
* Networking crate draft * Started work on syncing algo * Fixed query range * BlockCollection tests * In-mem client backend * Fixed tests * Renamed Transaction * Removed stray println * Docs
This commit is contained in:
Arkadiy Paronyan
committed by
Robert Habermeier
Robert Habermeier
parent
8e554129ec
commit
44499550d9
@@ -0,0 +1,372 @@
|
||||
// Copyright 2017 Parity Technologies (UK) Ltd.
|
||||
// This file is part of Polkadot.
|
||||
|
||||
// Polkadot is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Polkadot is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.?
|
||||
|
||||
use std::collections::HashMap;
|
||||
use io::SyncIo;
|
||||
use protocol::Protocol;
|
||||
use network::PeerId;
|
||||
use client::{ImportResult, BlockStatus, ClientInfo};
|
||||
use primitives::block::{HeaderHash, Number as BlockNumber, Header};
|
||||
use blocks::{self, BlockCollection};
|
||||
use message::{self, Message};
|
||||
use super::header_hash;
|
||||
|
||||
// Maximum parallel requests for a block.
|
||||
const MAX_BLOCK_DOWNLOAD: usize = 1;
|
||||
|
||||
struct PeerSync {
|
||||
pub common_hash: HeaderHash,
|
||||
pub common_number: BlockNumber,
|
||||
pub best_hash: HeaderHash,
|
||||
pub best_number: BlockNumber,
|
||||
pub state: PeerSyncState,
|
||||
}
|
||||
|
||||
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
|
||||
pub enum PeerSyncState {
|
||||
AncestorSearch(BlockNumber),
|
||||
Available,
|
||||
DownloadingNew(BlockNumber),
|
||||
DownloadingStale(HeaderHash),
|
||||
}
|
||||
|
||||
/// Relay chain sync strategy.
|
||||
pub struct ChainSync {
|
||||
genesis_hash: HeaderHash,
|
||||
peers: HashMap<PeerId, PeerSync>,
|
||||
blocks: BlockCollection,
|
||||
best_queued_number: BlockNumber,
|
||||
best_queued_hash: HeaderHash,
|
||||
required_block_attributes: Vec<message::BlockAttribute>,
|
||||
}
|
||||
|
||||
/// Syncing status and statistics
|
||||
#[derive(Clone, Copy)]
|
||||
pub struct Status;
|
||||
|
||||
impl ChainSync {
|
||||
pub fn new(info: &ClientInfo) -> ChainSync {
|
||||
ChainSync {
|
||||
genesis_hash: info.chain.genesis_hash,
|
||||
peers: HashMap::new(),
|
||||
blocks: BlockCollection::new(),
|
||||
best_queued_hash: info.best_queued_hash.unwrap_or(info.chain.best_hash),
|
||||
best_queued_number: info.best_queued_number.unwrap_or(info.chain.best_number),
|
||||
required_block_attributes: vec![message::BlockAttribute::Header, message::BlockAttribute::Body],
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns sync status
|
||||
pub fn status(&self) -> Status {
|
||||
Status
|
||||
}
|
||||
|
||||
pub fn new_peer(&mut self, io: &mut SyncIo, protocol: &Protocol, peer_id: PeerId) {
|
||||
if let Some(info) = protocol.peer_info(peer_id) {
|
||||
match (protocol.chain().block_status(&info.best_hash), info.best_number) {
|
||||
(Err(e), _) => {
|
||||
debug!(target:"sync", "Error reading blockchain: {:?}", e);
|
||||
io.disconnect_peer(peer_id);
|
||||
},
|
||||
(Ok(BlockStatus::KnownBad), _) => {
|
||||
debug!(target:"sync", "New peer with known bad best block {} ({}).", info.best_hash, info.best_number);
|
||||
io.disable_peer(peer_id);
|
||||
},
|
||||
(Ok(BlockStatus::Unknown), 0) => {
|
||||
debug!(target:"sync", "New peer with unkown genesis hash {} ({}).", info.best_hash, info.best_number);
|
||||
io.disable_peer(peer_id);
|
||||
},
|
||||
(Ok(BlockStatus::Unknown), _) => {
|
||||
debug!(target:"sync", "New peer with unkown best hash {} ({}), searching for common ancestor.", info.best_hash, info.best_number);
|
||||
self.peers.insert(peer_id, PeerSync {
|
||||
common_hash: self.genesis_hash,
|
||||
common_number: 0,
|
||||
best_hash: info.best_hash,
|
||||
best_number: info.best_number,
|
||||
state: PeerSyncState::AncestorSearch(info.best_number - 1),
|
||||
});
|
||||
Self::request_ancestry(io, protocol, peer_id, info.best_number - 1)
|
||||
},
|
||||
(Ok(BlockStatus::Queued), _) | (Ok(BlockStatus::InChain), _) => {
|
||||
debug!(target:"sync", "New peer with known best hash {} ({}).", info.best_hash, info.best_number);
|
||||
self.peers.insert(peer_id, PeerSync {
|
||||
common_hash: info.best_hash,
|
||||
common_number: info.best_number,
|
||||
best_hash: info.best_hash,
|
||||
best_number: info.best_number,
|
||||
state: PeerSyncState::Available,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn on_block_data(&mut self, io: &mut SyncIo, protocol: &Protocol, peer_id: PeerId, _request: message::BlockRequest, response: message::BlockResponse) {
|
||||
let count = response.blocks.len();
|
||||
let mut imported: usize = 0;
|
||||
let new_blocks = if let Some(ref mut peer) = self.peers.get_mut(&peer_id) {
|
||||
match peer.state {
|
||||
PeerSyncState::DownloadingNew(start_block) => {
|
||||
self.blocks.clear_peer_download(peer_id);
|
||||
peer.state = PeerSyncState::Available;
|
||||
|
||||
self.blocks.insert(start_block, response.blocks, peer_id);
|
||||
self.blocks.drain(self.best_queued_number + 1)
|
||||
},
|
||||
PeerSyncState::DownloadingStale(_) => {
|
||||
peer.state = PeerSyncState::Available;
|
||||
response.blocks.into_iter().map(|b| blocks::BlockData {
|
||||
origin: peer_id,
|
||||
block: b
|
||||
}).collect()
|
||||
},
|
||||
PeerSyncState::AncestorSearch(n) => {
|
||||
match response.blocks.get(0) {
|
||||
Some(ref block) => {
|
||||
match protocol.chain().block_hash(n) {
|
||||
Ok(Some(block_hash)) if block_hash == block.hash => {
|
||||
peer.common_hash = block.hash;
|
||||
peer.common_number = n;
|
||||
peer.state = PeerSyncState::Available;
|
||||
trace!(target:"sync", "Found common ancestor for peer {}: {} ({})", peer_id, block.hash, n);
|
||||
vec![]
|
||||
},
|
||||
Ok(_) if n > 0 => {
|
||||
let n = n - 1;
|
||||
peer.state = PeerSyncState::AncestorSearch(n);
|
||||
Self::request_ancestry(io, protocol, peer_id, n);
|
||||
return;
|
||||
},
|
||||
Ok(_) => { // genesis mismatch
|
||||
io.disable_peer(peer_id);
|
||||
return;
|
||||
},
|
||||
Err(e) => {
|
||||
debug!(target:"sync", "Error reading blockchain: {:?}", e);
|
||||
io.disconnect_peer(peer_id);
|
||||
return;
|
||||
}
|
||||
}
|
||||
},
|
||||
None => {
|
||||
trace!(target:"sync", "Invalid response when searching for ancestor from {}", peer_id);
|
||||
io.disconnect_peer(peer_id);
|
||||
return;
|
||||
}
|
||||
}
|
||||
},
|
||||
PeerSyncState::Available => Vec::new(),
|
||||
}
|
||||
} else {
|
||||
vec![]
|
||||
};
|
||||
|
||||
// Blocks in the response/drain should be in ascending order.
|
||||
for block in new_blocks {
|
||||
let origin = block.origin;
|
||||
let block = block.block;
|
||||
if let Some(header) = block.header {
|
||||
let number = header.number;
|
||||
let hash = header_hash(&header);
|
||||
let parent = header.parent_hash;
|
||||
let result = protocol.chain().import(header, block.body);
|
||||
match result {
|
||||
Ok(ImportResult::AlreadyInChain) => {
|
||||
trace!(target: "sync", "Block already in chain {}: {:?}", number, hash);
|
||||
self.block_imported(&hash, number);
|
||||
},
|
||||
Ok(ImportResult::AlreadyQueued) => {
|
||||
trace!(target: "sync", "Block already queued {}: {:?}", number, hash);
|
||||
self.block_imported(&hash, number);
|
||||
},
|
||||
Ok(ImportResult::Queued) => {
|
||||
trace!(target: "sync", "Block queued {}: {:?}", number, hash);
|
||||
self.block_imported(&hash, number);
|
||||
imported = imported + 1;
|
||||
},
|
||||
Ok(ImportResult::UnknownParent) => {
|
||||
debug!(target: "sync", "Block with unknown parent {}: {:?}, parent: {:?}", number, hash, parent);
|
||||
self.restart(io, protocol);
|
||||
return;
|
||||
},
|
||||
Ok(ImportResult::KnownBad) => {
|
||||
debug!(target: "sync", "Bad block {}: {:?}", number, hash);
|
||||
io.disable_peer(origin); //TODO: use persistent ID
|
||||
self.restart(io, protocol);
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
debug!(target: "sync", "Error importing block {}: {:?}: {:?}", number, hash, e);
|
||||
self.restart(io, protocol);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
trace!(target: "sync", "Imported {} of {}", imported, count);
|
||||
self.maintain_sync(io, protocol);
|
||||
}
|
||||
|
||||
fn maintain_sync(&mut self, io: &mut SyncIo, protocol: &Protocol) {
|
||||
let peers: Vec<PeerId> = self.peers.keys().map(|p| *p).collect();
|
||||
for peer in peers {
|
||||
self.download_new(io, protocol, peer);
|
||||
}
|
||||
}
|
||||
|
||||
fn block_imported(&mut self, hash: &HeaderHash, number: BlockNumber) {
|
||||
if number > self.best_queued_number {
|
||||
self.best_queued_number = number;
|
||||
self.best_queued_hash = *hash;
|
||||
}
|
||||
// Update common blocks
|
||||
for (_, peer) in self.peers.iter_mut() {
|
||||
if peer.best_number >= number {
|
||||
peer.common_number = number;
|
||||
peer.common_hash = *hash;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn update_chain_info(&mut self, best_header: &Header ) {
|
||||
let hash = header_hash(&best_header);
|
||||
self.block_imported(&hash, best_header.number)
|
||||
}
|
||||
|
||||
pub fn on_block_announce(&mut self, io: &mut SyncIo, protocol: &Protocol, peer_id: PeerId, header: &Header) {
|
||||
let hash = header_hash(&header);
|
||||
if let Some(ref mut peer) = self.peers.get_mut(&peer_id) {
|
||||
if header.number > peer.best_number {
|
||||
peer.best_number = header.number;
|
||||
peer.best_hash = hash;
|
||||
}
|
||||
} else {
|
||||
return;
|
||||
}
|
||||
|
||||
if !self.is_known_or_already_downloading(protocol, &hash) {
|
||||
let stale = header.number <= self.best_queued_number;
|
||||
if stale {
|
||||
if !self.is_known_or_already_downloading(protocol, &header.parent_hash) {
|
||||
trace!(target: "sync", "Ignoring unkown stale block announce from {}: {} {:?}", peer_id, hash, header);
|
||||
} else {
|
||||
trace!(target: "sync", "Downloading new stale block announced from {}: {} {:?}", peer_id, hash, header);
|
||||
self.download_stale(io, protocol, peer_id, &hash);
|
||||
}
|
||||
} else {
|
||||
trace!(target: "sync", "Downloading new block announced from {}: {} {:?}", peer_id, hash, header);
|
||||
self.download_new(io, protocol, peer_id);
|
||||
}
|
||||
} else {
|
||||
trace!(target: "sync", "Known block announce from {}: {}", peer_id, hash);
|
||||
}
|
||||
}
|
||||
|
||||
fn is_known_or_already_downloading(&self, protocol: &Protocol, hash: &HeaderHash) -> bool {
|
||||
self.peers.iter().any(|(_, p)| p.state == PeerSyncState::DownloadingStale(*hash))
|
||||
|| protocol.chain().block_status(hash).ok().map_or(false, |s| s != BlockStatus::Unknown)
|
||||
}
|
||||
|
||||
pub fn peer_disconnected(&mut self, io: &mut SyncIo, protocol: &Protocol, peer_id: PeerId) {
|
||||
self.blocks.clear_peer_download(peer_id);
|
||||
self.peers.remove(&peer_id);
|
||||
self.maintain_sync(io, protocol);
|
||||
}
|
||||
|
||||
pub fn restart(&mut self, io: &mut SyncIo, protocol: &Protocol) {
|
||||
self.blocks.clear();
|
||||
let ids: Vec<PeerId> = self.peers.keys().map(|p| *p).collect();
|
||||
for id in ids {
|
||||
self.new_peer(io, protocol, id);
|
||||
}
|
||||
match protocol.chain().info() {
|
||||
Ok(info) => {
|
||||
self.best_queued_hash = info.best_queued_hash.unwrap_or(info.chain.best_hash);
|
||||
self.best_queued_number = info.best_queued_number.unwrap_or(info.chain.best_number);
|
||||
},
|
||||
Err(e) => {
|
||||
debug!(target:"sync", "Error reading blockchain: {:?}", e);
|
||||
self.best_queued_hash = self.genesis_hash;
|
||||
self.best_queued_number = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn clear(&mut self) {
|
||||
self.blocks.clear();
|
||||
self.peers.clear();
|
||||
}
|
||||
|
||||
// Download old block.
|
||||
fn download_stale(&mut self, io: &mut SyncIo, protocol: &Protocol, peer_id: PeerId, hash: &HeaderHash) {
|
||||
if let Some(ref mut peer) = self.peers.get_mut(&peer_id) {
|
||||
match peer.state {
|
||||
PeerSyncState::Available => {
|
||||
let request = message::BlockRequest {
|
||||
id: 0,
|
||||
fields: self.required_block_attributes.clone(),
|
||||
from: message::FromBlock::Hash(*hash),
|
||||
to: None,
|
||||
direction: message::Direction::Ascending,
|
||||
max: Some(1),
|
||||
};
|
||||
peer.state = PeerSyncState::DownloadingStale(*hash);
|
||||
protocol.send_message(io, peer_id, Message::BlockRequest(request));
|
||||
},
|
||||
_ => (),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Issue a request for a peer to download new blocks, if any are available
|
||||
fn download_new(&mut self, io: &mut SyncIo, protocol: &Protocol, peer_id: PeerId) {
|
||||
if let Some(ref mut peer) = self.peers.get_mut(&peer_id) {
|
||||
match peer.state {
|
||||
PeerSyncState::Available => {
|
||||
if let Some(range) = self.blocks.needed_blocks(peer_id, MAX_BLOCK_DOWNLOAD, peer.common_number, peer.best_number) {
|
||||
let request = message::BlockRequest {
|
||||
id: 0,
|
||||
fields: self.required_block_attributes.clone(),
|
||||
from: message::FromBlock::Number(range.start),
|
||||
to: None,
|
||||
direction: message::Direction::Ascending,
|
||||
max: Some((range.end - range.start) as u32),
|
||||
};
|
||||
peer.state = PeerSyncState::DownloadingNew(range.start);
|
||||
protocol.send_message(io, peer_id, Message::BlockRequest(request));
|
||||
}
|
||||
},
|
||||
_ => (),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn request_ancestry(io: &mut SyncIo, protocol: &Protocol, peer_id: PeerId, block: BlockNumber) {
|
||||
let request = message::BlockRequest {
|
||||
id: 0,
|
||||
fields: vec![message::BlockAttribute::Header],
|
||||
from: message::FromBlock::Number(block),
|
||||
to: None,
|
||||
direction: message::Direction::Ascending,
|
||||
max: Some(1),
|
||||
};
|
||||
protocol.send_message(io, peer_id, Message::BlockRequest(request));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user