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

Increase number of requested block bodies in chain sync (#10247)

Browse Source 
* Increase the number of block bodies requested during Sync.

* Increase the number of block bodies requested during Sync.

* Check if our peer is an older parity client with the bug
  of not handling large requests properly

* Add a ClientVersion struct and a ClientCapabilites trait

* Make ClientVersion its own module

* Refactor and extend use of ClientVersion

* Replace strings with ClientVersion in PeerInfo

* Group further functionality in ClientCapabilities

* Move parity client version data from tuple to its own struct.

* Implement accessor methods for ParityClientData and remove them
from ClientVersion.

* Minor fixes

* Make functions specific to parity return types specific to parity.

* Test for shorter ID strings

* Fix formatting and remove unneeded dependencies.

* Roll back Cargo.lock

* Commit last Cargo.lock

* Convert from string to ClientVersion

* * When checking if peer accepts service transactions just check
  if it's parity, remove version check.

* Remove dependency on semver in ethcore-sync

* Remove unnecessary String instantiation

* Rename peer_info to peer_version

* Update RPC test helpers

* Simplify From<String>

* Parse static version string only once

* Update RPC tests to new ClientVersion struct

* Document public members

* More robust parsing of ID string

* Minor changes.

* Update version in which large block bodies requests appear.

* Update ethcore/sync/src/block_sync.rs

Co-Authored-By: elferdo <elferdo@gmail.com>

* Update util/network/src/client_version.rs

Co-Authored-By: elferdo <elferdo@gmail.com>

* Update util/network/src/client_version.rs

Co-Authored-By: elferdo <elferdo@gmail.com>

* Update tests.

* Minor fixes.
This commit is contained in:
elferdo
2019-02-07 15:27:09 +01:00
committed by Afri Schoedon
parent d5c19f8719
commit b7e8621846
21 changed files with 607 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
ethcore/sync/src/api.rs
View File

@@ -24,6 +24,7 @@ use devp2p::NetworkService;
use network::{NetworkProtocolHandler, NetworkContext, PeerId, ProtocolId,
NetworkConfiguration as BasicNetworkConfiguration, NonReservedPeerMode, Error, ErrorKind,
ConnectionFilter};
use network::client_version::ClientVersion;
use types::pruning_info::PruningInfo;
use ethereum_types::{H256, H512, U256};
@@ -158,7 +159,7 @@ pub struct PeerInfo {
/// Public node id
pub id: Option<String>,
/// Node client ID
pub client_version: String,
pub client_version: ClientVersion,
/// Capabilities
pub capabilities: Vec<String>,
/// Remote endpoint address

21
ethcore/sync/src/block_sync.rs
View File

@@ -29,10 +29,13 @@ use ethcore::error::{ImportErrorKind, QueueErrorKind, BlockError, Error as Ethco
use sync_io::SyncIo;
use blocks::{BlockCollection, SyncBody, SyncHeader};
use chain::BlockSet;
use network::PeerId;
use network::client_version::ClientCapabilities;
const MAX_HEADERS_TO_REQUEST: usize = 128;
const MAX_BODIES_TO_REQUEST: usize = 32;
const MAX_RECEPITS_TO_REQUEST: usize = 128;
const MAX_BODIES_TO_REQUEST_LARGE: usize = 128;
const MAX_BODIES_TO_REQUEST_SMALL: usize = 32; // Size request for parity clients prior to 2.4.0
const MAX_RECEPITS_TO_REQUEST: usize = 256;
const SUBCHAIN_SIZE: u64 = 256;
const MAX_ROUND_PARENTS: usize = 16;
const MAX_PARALLEL_SUBCHAIN_DOWNLOAD: usize = 5;
@@ -464,12 +467,12 @@ impl BlockDownloader {
}
/// Find some headers or blocks to download for a peer.
pub fn request_blocks(&mut self, io: &mut SyncIo, num_active_peers: usize) -> Option<BlockRequest> {
pub fn request_blocks(&mut self, peer_id: PeerId, io: &mut SyncIo, num_active_peers: usize) -> Option<BlockRequest> {
match self.state {
State::Idle => {
self.start_sync_round(io);
if self.state == State::ChainHead {
return self.request_blocks(io, num_active_peers);
return self.request_blocks(peer_id, io, num_active_peers);
}
},
State::ChainHead => {
@@ -487,7 +490,15 @@ impl BlockDownloader {
},
State::Blocks => {
// check to see if we need to download any block bodies first
let needed_bodies = self.blocks.needed_bodies(MAX_BODIES_TO_REQUEST, false);
let client_version = io.peer_version(peer_id);
let number_of_bodies_to_request = if client_version.can_handle_large_requests() {
MAX_BODIES_TO_REQUEST_LARGE
} else {
MAX_BODIES_TO_REQUEST_SMALL
};
let needed_bodies = self.blocks.needed_bodies(number_of_bodies_to_request, false);
if !needed_bodies.is_empty() {
return Some(BlockRequest::Bodies {
hashes: needed_bodies,

12
ethcore/sync/src/chain/handler.rs
View File

@@ -23,6 +23,7 @@ use ethcore::verification::queue::kind::blocks::Unverified;
use ethereum_types::{H256, U256};
use hash::keccak;
use network::PeerId;
use network::client_version::ClientVersion;
use rlp::Rlp;
use snapshot::ChunkType;
use std::time::Instant;
@@ -107,7 +108,7 @@ impl SyncHandler {
/// Called by peer when it is disconnecting
pub fn on_peer_aborting(sync: &mut ChainSync, io: &mut SyncIo, peer_id: PeerId) {
trace!(target: "sync", "== Disconnecting {}: {}", peer_id, io.peer_info(peer_id));
trace!(target: "sync", "== Disconnecting {}: {}", peer_id, io.peer_version(peer_id));
sync.handshaking_peers.remove(&peer_id);
if sync.peers.contains_key(&peer_id) {
debug!(target: "sync", "Disconnected {}", peer_id);
@@ -133,7 +134,7 @@ impl SyncHandler {
/// Called when a new peer is connected
pub fn on_peer_connected(sync: &mut ChainSync, io: &mut SyncIo, peer: PeerId) {
trace!(target: "sync", "== Connected {}: {}", peer, io.peer_info(peer));
trace!(target: "sync", "== Connected {}: {}", peer, io.peer_version(peer));
if let Err(e) = sync.send_status(io, peer) {
debug!(target:"sync", "Error sending status request: {:?}", e);
io.disconnect_peer(peer);
@@ -579,6 +580,7 @@ impl SyncHandler {
snapshot_number: if warp_protocol { Some(r.val_at(6)?) } else { None },
block_set: None,
private_tx_enabled: if private_tx_protocol { r.val_at(7).unwrap_or(false) } else { false },
client_version: ClientVersion::from(io.peer_version(peer_id)),
};
trace!(target: "sync", "New peer {} (\
@@ -599,12 +601,12 @@ impl SyncHandler {
peer.private_tx_enabled
);
if io.is_expired() {
trace!(target: "sync", "Status packet from expired session {}:{}", peer_id, io.peer_info(peer_id));
trace!(target: "sync", "Status packet from expired session {}:{}", peer_id, io.peer_version(peer_id));
return Ok(());
}
if sync.peers.contains_key(&peer_id) {
debug!(target: "sync", "Unexpected status packet from {}:{}", peer_id, io.peer_info(peer_id));
debug!(target: "sync", "Unexpected status packet from {}:{}", peer_id, io.peer_version(peer_id));
return Ok(());
}
let chain_info = io.chain().chain_info();
@@ -633,7 +635,7 @@ impl SyncHandler {
// Don't activate peer immediatelly when searching for common block.
// Let the current sync round complete first.
sync.active_peers.insert(peer_id.clone());
debug!(target: "sync", "Connected {}:{}", peer_id, io.peer_info(peer_id));
debug!(target: "sync", "Connected {}:{}", peer_id, io.peer_version(peer_id));
if let Some((fork_block, _)) = sync.fork_block {
SyncRequester::request_fork_header(sync, io, peer_id, fork_block);

8
ethcore/sync/src/chain/mod.rs
View File

@@ -104,6 +104,7 @@ use parking_lot::{Mutex, RwLock, RwLockWriteGuard};
use bytes::Bytes;
use rlp::{RlpStream, DecoderError};
use network::{self, PeerId, PacketId};
use network::client_version::ClientVersion;
use ethcore::client::{BlockChainClient, BlockStatus, BlockId, BlockChainInfo, BlockQueueInfo};
use ethcore::snapshot::{RestorationStatus};
use sync_io::SyncIo;
@@ -342,6 +343,8 @@ pub struct PeerInfo {
snapshot_number: Option<BlockNumber>,
/// Block set requested
block_set: Option<BlockSet>,
/// Version of the software the peer is running
client_version: ClientVersion,
}
impl PeerInfo {
@@ -964,7 +967,7 @@ impl ChainSync {
if !have_latest && (higher_difficulty || force || self.state == SyncState::NewBlocks) {
// check if got new blocks to download
trace!(target: "sync", "Syncing with peer {}, force={}, td={:?}, our td={}, state={:?}", peer_id, force, peer_difficulty, syncing_difficulty, self.state);
if let Some(request) = self.new_blocks.request_blocks(io, num_active_peers) {
if let Some(request) = self.new_blocks.request_blocks(peer_id, io, num_active_peers) {
SyncRequester::request_blocks(self, io, peer_id, request, BlockSet::NewBlocks);
if self.state == SyncState::Idle {
self.state = SyncState::Blocks;
@@ -977,7 +980,7 @@ impl ChainSync {
let equal_or_higher_difficulty = peer_difficulty.map_or(false, |pd| pd >= syncing_difficulty);
if force || equal_or_higher_difficulty {
if let Some(request) = self.old_blocks.as_mut().and_then(|d| d.request_blocks(io, num_active_peers)) {
if let Some(request) = self.old_blocks.as_mut().and_then(|d| d.request_blocks(peer_id, io, num_active_peers)) {
SyncRequester::request_blocks(self, io, peer_id, request, BlockSet::OldBlocks);
return;
}
@@ -1459,6 +1462,7 @@ pub mod tests {
snapshot_hash: None,
asking_snapshot_data: None,
block_set: None,
client_version: ClientVersion::from(""),
});
}

41
ethcore/sync/src/chain/propagator.rs
View File

@@ -21,6 +21,7 @@ use bytes::Bytes;
use ethereum_types::H256;
use fastmap::H256FastSet;
use network::{PeerId, PacketId};
use network::client_version::ClientCapabilities;
use rand::Rng;
use rlp::{Encodable, RlpStream};
use sync_io::SyncIo;
@@ -41,28 +42,6 @@ use super::{
TRANSACTIONS_PACKET,
};
/// Checks if peer is able to process service transactions
fn accepts_service_transaction(client_id: &str) -> bool {
// Parity versions starting from this will accept service-transactions
const SERVICE_TRANSACTIONS_VERSION: (u32, u32) = (1u32, 6u32);
// Parity client string prefix
const LEGACY_CLIENT_ID_PREFIX: &'static str = "Parity/";
const PARITY_CLIENT_ID_PREFIX: &'static str = "Parity-Ethereum/";
const VERSION_PREFIX: &'static str = "/v";
let idx = client_id.rfind(VERSION_PREFIX).map(|idx| idx + VERSION_PREFIX.len()).unwrap_or(client_id.len());
let splitted = if client_id.starts_with(LEGACY_CLIENT_ID_PREFIX) || client_id.starts_with(PARITY_CLIENT_ID_PREFIX) {
client_id[idx..].split('.')
} else {
return false;
};
let ver: Vec<u32> = splitted
.take(2)
.filter_map(|s| s.parse().ok())
.collect();
ver.len() == 2 && (ver[0] > SERVICE_TRANSACTIONS_VERSION.0 || (ver[0] == SERVICE_TRANSACTIONS_VERSION.0 && ver[1] >= SERVICE_TRANSACTIONS_VERSION.1))
}
/// The Chain Sync Propagator: propagates data to peers
pub struct SyncPropagator;
@@ -146,7 +125,7 @@ impl SyncPropagator {
// most of times service_transactions will be empty
// => there's no need to merge packets
if !service_transactions.is_empty() {
let service_transactions_peers = SyncPropagator::select_peers_for_transactions(sync, |peer_id| accepts_service_transaction(&io.peer_info(*peer_id)));
let service_transactions_peers = SyncPropagator::select_peers_for_transactions(sync, |peer_id| io.peer_version(*peer_id).accepts_service_transaction());
let service_transactions_affected_peers = SyncPropagator::propagate_transactions_to_peers(
sync, io, service_transactions_peers, service_transactions, &mut should_continue
);
@@ -451,6 +430,7 @@ mod tests {
snapshot_hash: None,
asking_snapshot_data: None,
block_set: None,
client_version: ClientVersion::from(""),
});
let ss = TestSnapshotService::new();
let mut io = TestIo::new(&mut client, &ss, &queue, None);
@@ -598,20 +578,17 @@ mod tests {
io.peers_info.insert(1, "Geth".to_owned());
// and peer#2 is Parity, accepting service transactions
insert_dummy_peer(&mut sync, 2, block_hash);
io.peers_info.insert(2, "Parity-Ethereum/v2.6".to_owned());
// and peer#3 is Parity, discarding service transactions
io.peers_info.insert(2, "Parity-Ethereum/v2.6.0/linux/rustc".to_owned());
// and peer#3 is Parity, accepting service transactions
insert_dummy_peer(&mut sync, 3, block_hash);
io.peers_info.insert(3, "Parity/v1.5".to_owned());
// and peer#4 is Parity, accepting service transactions
insert_dummy_peer(&mut sync, 4, block_hash);
io.peers_info.insert(4, "Parity-Ethereum/ABCDEFGH/v2.7.3".to_owned());
io.peers_info.insert(3, "Parity-Ethereum/ABCDEFGH/v2.7.3/linux/rustc".to_owned());
// and new service transaction is propagated to peers
SyncPropagator::propagate_new_transactions(&mut sync, &mut io, || true);
// peer#2 && peer#4 are receiving service transaction
// peer#2 && peer#3 are receiving service transaction
assert!(io.packets.iter().any(|p| p.packet_id == 0x02 && p.recipient == 2)); // TRANSACTIONS_PACKET
assert!(io.packets.iter().any(|p| p.packet_id == 0x02 && p.recipient == 4)); // TRANSACTIONS_PACKET
assert!(io.packets.iter().any(|p| p.packet_id == 0x02 && p.recipient == 3)); // TRANSACTIONS_PACKET
assert_eq!(io.packets.len(), 2);
}
@@ -628,7 +605,7 @@ mod tests {
// when peer#1 is Parity, accepting service transactions
insert_dummy_peer(&mut sync, 1, block_hash);
io.peers_info.insert(1, "Parity-Ethereum/v2.6".to_owned());
io.peers_info.insert(1, "Parity-Ethereum/v2.6.0/linux/rustc".to_owned());
// and service + non-service transactions are propagated to peers
SyncPropagator::propagate_new_transactions(&mut sync, &mut io, || true);

1
ethcore/sync/src/chain/requester.rs
View File

@@ -62,6 +62,7 @@ impl SyncRequester {
for h in &hashes {
rlp.append(&h.clone());
}
SyncRequester::send_request(sync, io, peer_id, PeerAsking::BlockBodies, GET_BLOCK_BODIES_PACKET, rlp.out());
let peer = sync.peers.get_mut(&peer_id).expect("peer_id may originate either from on_packet, where it is already validated or from enumerating self.peers. qed");
peer.asking_blocks = hashes;

2
ethcore/sync/src/chain/supplier.rs
View File

@@ -91,7 +91,7 @@ impl SyncSupplier {
// Packets that require the peer to be confirmed
_ => {
if !sync.read().peers.contains_key(&peer) {
debug!(target:"sync", "Unexpected packet {} from unregistered peer: {}:{}", packet_id, peer, io.peer_info(peer));
debug!(target:"sync", "Unexpected packet {} from unregistered peer: {}:{}", packet_id, peer, io.peer_version(peer));
return;
}
debug!(target: "sync", "{} -> Dispatching packet: {}", peer, packet_id);

9
ethcore/sync/src/sync_io.rs
View File

@@ -16,6 +16,7 @@
use std::collections::HashMap;
use network::{NetworkContext, PeerId, PacketId, Error, SessionInfo, ProtocolId};
use network::client_version::ClientVersion;
use bytes::Bytes;
use ethcore::client::BlockChainClient;
use types::BlockNumber;
@@ -40,9 +41,9 @@ pub trait SyncIo {
fn chain(&self) -> &BlockChainClient;
/// Get the snapshot service.
fn snapshot_service(&self) -> &SnapshotService;
/// Returns peer identifier string
fn peer_info(&self, peer_id: PeerId) -> String {
peer_id.to_string()
/// Returns peer version identifier
fn peer_version(&self, peer_id: PeerId) -> ClientVersion {
ClientVersion::from(peer_id.to_string())
}
/// Returns information on p2p session
fn peer_session_info(&self, peer_id: PeerId) -> Option<SessionInfo>;
@@ -134,7 +135,7 @@ impl<'s> SyncIo for NetSyncIo<'s> {
self.network.protocol_version(*protocol, peer_id).unwrap_or(0)
}
fn peer_info(&self, peer_id: PeerId) -> String {
fn peer_version(&self, peer_id: PeerId) -> ClientVersion {
self.network.peer_client_version(peer_id)
}

9
ethcore/sync/src/tests/helpers.rs
View File

@@ -20,6 +20,7 @@ use ethereum_types::H256;
use parking_lot::{RwLock, Mutex};
use bytes::Bytes;
use network::{self, PeerId, ProtocolId, PacketId, SessionInfo};
use network::client_version::ClientVersion;
use tests::snapshot::*;
use ethcore::client::{TestBlockChainClient, BlockChainClient, Client as EthcoreClient,
ClientConfig, ChainNotify, NewBlocks, ChainMessageType, ClientIoMessage};
@@ -118,10 +119,12 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
&*self.chain
}
fn peer_info(&self, peer_id: PeerId) -> String {
self.peers_info.get(&peer_id)
fn peer_version(&self, peer_id: PeerId) -> ClientVersion {
let client_id = self.peers_info.get(&peer_id)
.cloned()
.unwrap_or_else(|| peer_id.to_string())
.unwrap_or_else(|| peer_id.to_string());
ClientVersion::from(client_id)
}
fn snapshot_service(&self) -> &SnapshotService {