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

Implement eth/64 (EIP-2364) and drop support for eth/62 (#11472)

Browse Source 
* sync: make code friendlier to future conditional parsing

* Implement eth/64 (EIP-2364) and drop support for eth/62
This commit is contained in:
Artem Vorotnikov
2020-02-19 16:42:52 +03:00
committed by GitHub
parent a49950e9c0
commit 06df521eff
18 changed files with 318 additions and 410 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,7 +15,7 @@
// along with Parity Ethereum. If not, see <http://www.gnu.org/licenses/>.
use std::sync::{Arc, mpsc, atomic};
use std::collections::{HashMap, BTreeMap};
use std::collections::{BTreeSet, HashMap, BTreeMap};
use std::io;
use std::ops::RangeInclusive;
use std::time::Duration;
@@ -27,10 +27,11 @@ use crate::sync_io::NetSyncIo;
use crate::light_sync::{self, SyncInfo};
use crate::private_tx::PrivateTxHandler;
use crate::chain::{
fork_filter::ForkFilterApi,
sync_packet::SyncPacket::{PrivateTransactionPacket, SignedPrivateTransactionPacket},
ChainSyncApi, SyncState, SyncStatus as EthSyncStatus, ETH_PROTOCOL_VERSION_62,
ETH_PROTOCOL_VERSION_63, PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_2,
PAR_PROTOCOL_VERSION_3, PAR_PROTOCOL_VERSION_4,
ChainSyncApi, SyncState, SyncStatus as EthSyncStatus,
ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_64,
PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_2, PAR_PROTOCOL_VERSION_3, PAR_PROTOCOL_VERSION_4,
};
use bytes::Bytes;
@@ -268,6 +269,8 @@ pub struct Params {
pub executor: Executor,
/// Blockchain client.
pub chain: Arc<dyn BlockChainClient>,
/// Forks.
pub forks: BTreeSet<BlockNumber>,
/// Snapshot service.
pub snapshot_service: Arc<dyn SnapshotService>,
/// Private tx service.
@@ -348,10 +351,13 @@ impl EthSync {
})
};
let fork_filter = ForkFilterApi::new(&*params.chain, params.forks);
let (priority_tasks_tx, priority_tasks_rx) = mpsc::channel();
let sync = ChainSyncApi::new(
params.config,
&*params.chain,
fork_filter,
params.private_tx_handler.as_ref().cloned(),
priority_tasks_rx,
);
@@ -605,7 +611,7 @@ impl ChainNotify for EthSync {
_ => {},
}
self.network.register_protocol(self.eth_handler.clone(), self.subprotocol_name, &[ETH_PROTOCOL_VERSION_62, ETH_PROTOCOL_VERSION_63])
self.network.register_protocol(self.eth_handler.clone(), self.subprotocol_name, &[ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_64])
.unwrap_or_else(|e| warn!("Error registering ethereum protocol: {:?}", e));
// register the warp sync subprotocol
self.network.register_protocol(self.eth_handler.clone(), WARP_SYNC_PROTOCOL_ID, &[PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_2, PAR_PROTOCOL_VERSION_3, PAR_PROTOCOL_VERSION_4])

143
ethcore/sync/src/chain/fork_filter.rs Normal file
View File

@@ -0,0 +1,143 @@
// Copyright 2015-2019 Parity Technologies (UK) Ltd.
// This file is part of Parity Ethereum.
// Parity Ethereum 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.
// Parity Ethereum 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 Parity Ethereum. If not, see <http://www.gnu.org/licenses/>.
//! This module contains a wrapper that connects this codebase with `ethereum-forkid` crate which provides `FORK_ID`
//! to support Ethereum network protocol, version 64 and above.
// Re-export ethereum-forkid crate contents here.
pub use ethereum_forkid::{BlockNumber, ForkId, RejectReason};
use client_traits::ChainInfo;
use ethereum_forkid::ForkFilter;
use parity_util_mem::MallocSizeOf;
/// Wrapper around fork filter that provides integration with `ForkFilter`.
#[derive(MallocSizeOf)]
pub struct ForkFilterApi {
inner: ForkFilter,
}
impl ForkFilterApi {
/// Create `ForkFilterApi` from `ChainInfo` and an `Iterator` over the hard forks.
pub fn new<C: ?Sized + ChainInfo, I: IntoIterator<Item = BlockNumber>>(client: &C, forks: I) -> Self {
let chain_info = client.chain_info();
Self {
inner: ForkFilter::new(chain_info.best_block_number, chain_info.genesis_hash, forks),
}
}
#[cfg(test)]
/// Dummy version of ForkFilterApi with no forks.
pub fn new_dummy<C: ?Sized + ChainInfo>(client: &C) -> Self {
let chain_info = client.chain_info();
Self {
inner: ForkFilter::new(chain_info.best_block_number, chain_info.genesis_hash, vec![]),
}
}
fn update_head<C: ?Sized + ChainInfo>(&mut self, client: &C) {
self.inner.set_head(client.chain_info().best_block_number);
}
/// Wrapper for `ForkFilter::current`
pub fn current<C: ?Sized + ChainInfo>(&mut self, client: &C) -> ForkId {
self.update_head(client);
self.inner.current()
}
/// Wrapper for `ForkFilter::is_compatible`
pub fn is_compatible<C: ?Sized + ChainInfo>(&mut self, client: &C, fork_id: ForkId) -> Result<(), RejectReason> {
self.update_head(client);
self.inner.is_compatible(fork_id)
}
}
#[cfg(test)]
mod tests {
use super::*;
use spec::Spec;
use ethcore::test_helpers::TestBlockChainClient;
fn test_spec<F: Fn() -> Spec>(spec_builder: F, forks: Vec<BlockNumber>) {
let spec = (spec_builder)();
let genesis_hash = spec.genesis_header().hash();
let spec_forks = spec.hard_forks.clone();
let client = TestBlockChainClient::new_with_spec(spec);
assert_eq!(
ForkFilterApi::new(&client, spec_forks).inner,
ForkFilter::new(0, genesis_hash, forks)
);
}
#[test]
fn ethereum_spec() {
test_spec(
|| spec::new_foundation(&String::new()),
vec![
1_150_000,
1_920_000,
2_463_000,
2_675_000,
4_370_000,
7_280_000,
9_069_000,
9_200_000,
],
)
}
#[test]
fn ropsten_spec() {
test_spec(
|| spec::new_ropsten(&String::new()),
vec![
10,
1_700_000,
4_230_000,
4_939_394,
6_485_846,
7_117_117,
],
)
}
#[test]
fn rinkeby_spec() {
test_spec(
|| spec::new_rinkeby(&String::new()),
vec![
1,
2,
3,
1_035_301,
3_660_663,
4_321_234,
5_435_345,
],
)
}
#[test]
fn goerli_spec() {
test_spec(
|| spec::new_goerli(&String::new()),
vec![
1_561_651,
],
)
}
}

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

@@ -20,7 +20,7 @@ use std::{mem, cmp};
use crate::{
snapshot_sync::ChunkType,
sync_io::SyncIo,
api::WARP_SYNC_PROTOCOL_ID,
api::{ETH_PROTOCOL, WARP_SYNC_PROTOCOL_ID},
block_sync::{BlockDownloaderImportError as DownloaderImportError, DownloadAction},
chain::{
sync_packet::{
@@ -32,7 +32,7 @@ use crate::{
}
},
BlockSet, ChainSync, ForkConfirmation, PacketDecodeError, PeerAsking, PeerInfo, SyncRequester,
SyncState, ETH_PROTOCOL_VERSION_62, ETH_PROTOCOL_VERSION_63, MAX_NEW_BLOCK_AGE, MAX_NEW_HASHES,
SyncState, ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_64, MAX_NEW_BLOCK_AGE, MAX_NEW_HASHES,
PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_3, PAR_PROTOCOL_VERSION_4,
}
};
@@ -562,17 +562,34 @@ impl SyncHandler {
/// Called by peer to report status
fn on_peer_status(sync: &mut ChainSync, io: &mut dyn SyncIo, peer_id: PeerId, r: &Rlp) -> Result<(), DownloaderImportError> {
let mut r = r.iter();
sync.handshaking_peers.remove(&peer_id);
let protocol_version: u8 = r.val_at(0)?;
let protocol_version: u8 = r.next().ok_or(rlp::DecoderError::RlpIsTooShort)?.as_val()?;
let eth_protocol_version = io.protocol_version(&ETH_PROTOCOL, peer_id);
let warp_protocol_version = io.protocol_version(&WARP_SYNC_PROTOCOL_ID, peer_id);
let warp_protocol = warp_protocol_version != 0;
let private_tx_protocol = warp_protocol_version >= PAR_PROTOCOL_VERSION_3.0;
let network_id = r.next().ok_or(rlp::DecoderError::RlpIsTooShort)?.as_val()?;
let difficulty = Some(r.next().ok_or(rlp::DecoderError::RlpIsTooShort)?.as_val()?);
let latest_hash = r.next().ok_or(rlp::DecoderError::RlpIsTooShort)?.as_val()?;
let genesis = r.next().ok_or(rlp::DecoderError::RlpIsTooShort)?.as_val()?;
let forkid_validation_error = {
if eth_protocol_version >= ETH_PROTOCOL_VERSION_64.0 {
let fork_id = r.next().ok_or(rlp::DecoderError::RlpIsTooShort)?.as_val()?;
sync.fork_filter.is_compatible(io.chain(), fork_id).err().map(|e| (fork_id, e))
} else {
None
}
};
let snapshot_hash = if warp_protocol { Some(r.next().ok_or(rlp::DecoderError::RlpIsTooShort)?.as_val()?) } else { None };
let snapshot_number = if warp_protocol { Some(r.next().ok_or(rlp::DecoderError::RlpIsTooShort)?.as_val()?) } else { None };
let private_tx_enabled = if private_tx_protocol { r.next().and_then(|v| v.as_val().ok()).unwrap_or(false) } else { false };
let peer = PeerInfo {
protocol_version,
network_id: r.val_at(1)?,
difficulty: Some(r.val_at(2)?),
latest_hash: r.val_at(3)?,
genesis: r.val_at(4)?,
network_id,
difficulty,
latest_hash,
genesis,
asking: PeerAsking::Nothing,
asking_blocks: Vec::new(),
asking_hash: None,
@@ -583,10 +600,10 @@ impl SyncHandler {
expired: false,
confirmation: if sync.fork_block.is_none() { ForkConfirmation::Confirmed } else { ForkConfirmation::Unconfirmed },
asking_snapshot_data: None,
snapshot_hash: if warp_protocol { Some(r.val_at(5)?) } else { None },
snapshot_number: if warp_protocol { Some(r.val_at(6)?) } else { None },
snapshot_hash,
snapshot_number,
block_set: None,
private_tx_enabled: if private_tx_protocol { r.val_at(7).unwrap_or(false) } else { false },
private_tx_enabled,
client_version: ClientVersion::from(io.peer_version(peer_id)),
};
@@ -627,10 +644,14 @@ impl SyncHandler {
trace!(target: "sync", "Peer {} network id mismatch (ours: {}, theirs: {})", peer_id, sync.network_id, peer.network_id);
return Err(DownloaderImportError::Invalid);
}
if let Some((fork_id, reason)) = forkid_validation_error {
trace!(target: "sync", "Peer {} incompatible fork id (fork id: {:#x}/{}, error: {:?})", peer_id, fork_id.hash.0, fork_id.next, reason);
return Err(DownloaderImportError::Invalid);
}
if false
|| (warp_protocol && (peer.protocol_version < PAR_PROTOCOL_VERSION_1.0 || peer.protocol_version > PAR_PROTOCOL_VERSION_4.0))
|| (!warp_protocol && (peer.protocol_version < ETH_PROTOCOL_VERSION_62.0 || peer.protocol_version > ETH_PROTOCOL_VERSION_63.0))
|| (!warp_protocol && (peer.protocol_version < ETH_PROTOCOL_VERSION_63.0 || peer.protocol_version > ETH_PROTOCOL_VERSION_64.0))
{
trace!(target: "sync", "Peer {} unsupported eth protocol ({})", peer_id, peer.protocol_version);
return Err(DownloaderImportError::Invalid);

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

@@ -92,6 +92,7 @@ mod propagator;
mod requester;
mod supplier;
pub mod fork_filter;
pub mod sync_packet;
use std::sync::{Arc, mpsc};
@@ -100,9 +101,10 @@ use std::cmp;
use std::time::{Duration, Instant};
use crate::{
EthProtocolInfo as PeerInfoDigest, PriorityTask, SyncConfig, WarpSync, WARP_SYNC_PROTOCOL_ID,
ETH_PROTOCOL, EthProtocolInfo as PeerInfoDigest, PriorityTask, SyncConfig, WarpSync, WARP_SYNC_PROTOCOL_ID,
api::{Notification, PRIORITY_TIMER_INTERVAL},
block_sync::{BlockDownloader, DownloadAction},
chain::fork_filter::ForkFilterApi,
sync_io::SyncIo,
snapshot_sync::Snapshot,
transactions_stats::{TransactionsStats, Stats as TransactionStats},
@@ -147,10 +149,10 @@ malloc_size_of_is_0!(PeerInfo);
pub type PacketDecodeError = DecoderError;
/// 63 version of Ethereum protocol.
/// Version 64 of the Ethereum protocol and number of packet IDs reserved by the protocol (packet count).
pub const ETH_PROTOCOL_VERSION_64: (u8, u8) = (64, 0x11);
/// Version 63 of the Ethereum protocol and number of packet IDs reserved by the protocol (packet count).
pub const ETH_PROTOCOL_VERSION_63: (u8, u8) = (63, 0x11);
/// 62 version of Ethereum protocol.
pub const ETH_PROTOCOL_VERSION_62: (u8, u8) = (62, 0x11);
/// 1 version of Parity protocol and the packet count.
pub const PAR_PROTOCOL_VERSION_1: (u8, u8) = (1, 0x15);
/// 2 version of Parity protocol (consensus messages added).
@@ -427,11 +429,12 @@ impl ChainSyncApi {
pub fn new(
config: SyncConfig,
chain: &dyn BlockChainClient,
fork_filter: ForkFilterApi,
private_tx_handler: Option<Arc<dyn PrivateTxHandler>>,
priority_tasks: mpsc::Receiver<PriorityTask>,
) -> Self {
ChainSyncApi {
sync: RwLock::new(ChainSync::new(config, chain, private_tx_handler)),
sync: RwLock::new(ChainSync::new(config, chain, fork_filter, private_tx_handler)),
priority_tasks: Mutex::new(priority_tasks),
}
}
@@ -660,6 +663,8 @@ pub struct ChainSync {
last_sent_block_number: BlockNumber,
/// Network ID
network_id: u64,
/// Fork filter
fork_filter: ForkFilterApi,
/// Optional fork block to check
fork_block: Option<(BlockNumber, H256)>,
/// Snapshot downloader.
@@ -688,6 +693,7 @@ impl ChainSync {
pub fn new(
config: SyncConfig,
chain: &dyn BlockChainClient,
fork_filter: ForkFilterApi,
private_tx_handler: Option<Arc<dyn PrivateTxHandler>>,
) -> Self {
let chain_info = chain.chain_info();
@@ -705,6 +711,7 @@ impl ChainSync {
old_blocks: None,
last_sent_block_number: 0,
network_id: config.network_id,
fork_filter,
fork_block: config.fork_block,
download_old_blocks: config.download_old_blocks,
snapshot: Snapshot::new(),
@@ -723,7 +730,7 @@ impl ChainSync {
let last_imported_number = self.new_blocks.last_imported_block_number();
SyncStatus {
state: self.state.clone(),
protocol_version: ETH_PROTOCOL_VERSION_63.0,
protocol_version: ETH_PROTOCOL_VERSION_64.0,
network_id: self.network_id,
start_block_number: self.starting_block,
last_imported_block_number: Some(last_imported_number),
@@ -1240,10 +1247,11 @@ impl ChainSync {
/// Send Status message
fn send_status(&mut self, io: &mut dyn SyncIo, peer: PeerId) -> Result<(), network::Error> {
let eth_protocol_version = io.protocol_version(&ETH_PROTOCOL, peer);
let warp_protocol_version = io.protocol_version(&WARP_SYNC_PROTOCOL_ID, peer);
let warp_protocol = warp_protocol_version != 0;
let private_tx_protocol = warp_protocol_version >= PAR_PROTOCOL_VERSION_3.0;
let protocol = if warp_protocol { warp_protocol_version } else { ETH_PROTOCOL_VERSION_63.0 };
let protocol = if warp_protocol { warp_protocol_version } else { eth_protocol_version };
trace!(target: "sync", "Sending status to {}, protocol version {}", peer, protocol);
let mut packet = RlpStream::new();
packet.begin_unbounded_list();
@@ -1253,6 +1261,9 @@ impl ChainSync {
packet.append(&chain.total_difficulty);
packet.append(&chain.best_block_hash);
packet.append(&chain.genesis_hash);
if eth_protocol_version >= ETH_PROTOCOL_VERSION_64.0 {
packet.append(&self.fork_filter.current(io.chain()));
}
if warp_protocol {
let manifest = io.snapshot_service().manifest();
let block_number = manifest.as_ref().map_or(0, |m| m.block_number);
@@ -1499,6 +1510,7 @@ pub mod tests {
use crate::{
api::SyncConfig,
chain::ForkFilterApi,
tests::{helpers::TestIo, snapshot::TestSnapshotService},
};
@@ -1594,8 +1606,7 @@ pub mod tests {
}
pub fn dummy_sync_with_peer(peer_latest_hash: H256, client: &dyn BlockChainClient) -> ChainSync {
let mut sync = ChainSync::new(SyncConfig::default(), client, None,);
let mut sync = ChainSync::new(SyncConfig::default(), client, ForkFilterApi::new_dummy(client), None,);
insert_dummy_peer(&mut sync, 0, peer_latest_hash);
sync
}

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

@@ -338,7 +338,10 @@ mod tests {
use crate::{
api::SyncConfig,
chain::{ChainSync, ForkConfirmation, PeerAsking, PeerInfo},
chain::{
fork_filter::ForkFilterApi,
ChainSync, ForkConfirmation, PeerAsking, PeerInfo
},
tests::{helpers::TestIo, snapshot::TestSnapshotService},
};
@@ -423,7 +426,7 @@ mod tests {
client.add_blocks(2, EachBlockWith::Uncle);
let queue = RwLock::new(VecDeque::new());
let block = client.block(BlockId::Latest).unwrap().into_inner();
let mut sync = ChainSync::new(SyncConfig::default(), &client, None);
let mut sync = ChainSync::new(SyncConfig::default(), &client, ForkFilterApi::new_dummy(&client), None);
sync.peers.insert(0,
PeerInfo {
// Messaging protocol
@@ -514,7 +517,7 @@ mod tests {
client.add_blocks(100, EachBlockWith::Uncle);
client.insert_transaction_to_queue();
// Sync with no peers
let mut sync = ChainSync::new(SyncConfig::default(), &client, None);
let mut sync = ChainSync::new(SyncConfig::default(), &client, ForkFilterApi::new_dummy(&client), None);
let queue = RwLock::new(VecDeque::new());
let ss = TestSnapshotService::new();
let mut io = TestIo::new(&mut client, &ss, &queue, None, None);
@@ -584,7 +587,7 @@ mod tests {
let mut client = TestBlockChainClient::new();
client.insert_transaction_with_gas_price_to_queue(U256::zero());
let block_hash = client.block_hash_delta_minus(1);
let mut sync = ChainSync::new(SyncConfig::default(), &client, None);
let mut sync = ChainSync::new(SyncConfig::default(), &client, ForkFilterApi::new_dummy(&client), None);
let queue = RwLock::new(VecDeque::new());
let ss = TestSnapshotService::new();
let mut io = TestIo::new(&mut client, &ss, &queue, None, None);
@@ -614,7 +617,7 @@ mod tests {
let tx1_hash = client.insert_transaction_to_queue();
let tx2_hash = client.insert_transaction_with_gas_price_to_queue(U256::zero());
let block_hash = client.block_hash_delta_minus(1);
let mut sync = ChainSync::new(SyncConfig::default(), &client, None);
let mut sync = ChainSync::new(SyncConfig::default(), &client, ForkFilterApi::new_dummy(&client), None);
let queue = RwLock::new(VecDeque::new());
let ss = TestSnapshotService::new();
let mut io = TestIo::new(&mut client, &ss, &queue, None, None);

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

@@ -20,11 +20,12 @@ use std::sync::Arc;
use crate::{
api::{SyncConfig, WARP_SYNC_PROTOCOL_ID},
chain::{
fork_filter::ForkFilterApi,
sync_packet::{
PacketInfo,
SyncPacket::{self, PrivateTransactionPacket, SignedPrivateTransactionPacket}
},
ChainSync, SyncSupplier, ETH_PROTOCOL_VERSION_63, PAR_PROTOCOL_VERSION_4
ChainSync, SyncSupplier, ETH_PROTOCOL_VERSION_64, PAR_PROTOCOL_VERSION_4
},
private_tx::SimplePrivateTxHandler,
sync_io::SyncIo,
@@ -156,7 +157,7 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
}
fn protocol_version(&self, protocol: &ProtocolId, _peer_id: PeerId) -> u8 {
if protocol == &WARP_SYNC_PROTOCOL_ID { PAR_PROTOCOL_VERSION_4.0 } else { ETH_PROTOCOL_VERSION_63.0 }
if protocol == &WARP_SYNC_PROTOCOL_ID { PAR_PROTOCOL_VERSION_4.0 } else { ETH_PROTOCOL_VERSION_64.0 }
}
fn is_expired(&self) -> bool {
@@ -380,7 +381,7 @@ impl TestNet<EthPeer<TestBlockChainClient>> {
let chain = TestBlockChainClient::new();
let ss = Arc::new(TestSnapshotService::new());
let private_tx_handler = Arc::new(SimplePrivateTxHandler::default());
let sync = ChainSync::new(config.clone(), &chain, Some(private_tx_handler.clone()));
let sync = ChainSync::new(config.clone(), &chain, ForkFilterApi::new_dummy(&chain), Some(private_tx_handler.clone()));
net.peers.push(Arc::new(EthPeer {
sync: RwLock::new(sync),
snapshot_service: ss,
@@ -431,10 +432,11 @@ impl TestNet<EthPeer<EthcoreClient>> {
miner.clone(),
channel.clone()
).unwrap();
let fork_filter = ForkFilterApi::new(&*client, spec.hard_forks.clone());
let private_tx_handler = Arc::new(SimplePrivateTxHandler::default());
let ss = Arc::new(TestSnapshotService::new());
let sync = ChainSync::new(config, &*client, Some(private_tx_handler.clone()));
let sync = ChainSync::new(config, &*client, fork_filter, Some(private_tx_handler.clone()));
let peer = Arc::new(EthPeer {
sync: RwLock::new(sync),
snapshot_service: ss,