Skip to content
This repository has been archived by the owner on Nov 6, 2020. It is now read-only.

Make specification of protocol in SyncRequester::send_request explicit #10295

Merged
merged 12 commits into from
Feb 8, 2019
Merged

Make specification of protocol in SyncRequester::send_request explicit #10295

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
27a6d98
* Make the specification of the protocol to which a packet_id belongs to
Feb 4, 2019
584ed4f
Adapt tests to new code.
Feb 4, 2019
aecfb1f
Strengthen tests to check if packet_id and protocol match when sending
Feb 6, 2019
0bef1ce
Early proposal for binding packet_id and protocol together
Feb 7, 2019
5d8bc34
Replace PacketId with SyncPacketId in chain sync.
Feb 7, 2019
afd1431
Rename module.
Feb 8, 2019
1d9ae78
Remove send_protocol from SyncIo and leave just send.
Feb 8, 2019
16afb11
Rename trait Packet to PacketInfo.
Feb 8, 2019
81022bf
Document.
Feb 8, 2019
9dbdd15
Revert "Early proposal for binding packet_id and protocol together"
Feb 8, 2019
349eae0
Merge branch 'master' into fhc-explicit-protocol-send-request
elferdo Feb 8, 2019
7b245ba
Fix conflict merge error.
Feb 8, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ethcore/sync/src/api.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -576,9 +576,9 @@ impl ChainNotify for EthSync {
match message_type {
ChainMessageType::Consensus(message) => self.eth_handler.sync.write().propagate_consensus_packet(&mut sync_io, message),
ChainMessageType::PrivateTransaction(transaction_hash, message) =>
self.eth_handler.sync.write().propagate_private_transaction(&mut sync_io, transaction_hash, PRIVATE_TRANSACTION_PACKET, message),
self.eth_handler.sync.write().propagate_private_transaction(&mut sync_io, transaction_hash, WARP_SYNC_PROTOCOL_ID, PRIVATE_TRANSACTION_PACKET, message),
ChainMessageType::SignedPrivateTransaction(transaction_hash, message) =>
self.eth_handler.sync.write().propagate_private_transaction(&mut sync_io, transaction_hash, SIGNED_PRIVATE_TRANSACTION_PACKET, message),
self.eth_handler.sync.write().propagate_private_transaction(&mut sync_io, transaction_hash, WARP_SYNC_PROTOCOL_ID, SIGNED_PRIVATE_TRANSACTION_PACKET, message),
}
});
}
Expand Down
12 changes: 6 additions & 6 deletions ethcore/sync/src/chain/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,15 +103,15 @@ use fastmap::{H256FastMap, H256FastSet};
use parking_lot::{Mutex, RwLock, RwLockWriteGuard};
use bytes::Bytes;
use rlp::{RlpStream, DecoderError};
use network::{self, PeerId, PacketId};
use network::{self, PeerId, PacketId, ProtocolId};
use ethcore::client::{BlockChainClient, BlockStatus, BlockId, BlockChainInfo, BlockQueueInfo};
use ethcore::snapshot::{RestorationStatus};
use sync_io::SyncIo;
use super::{WarpSync, SyncConfig};
use block_sync::{BlockDownloader, DownloadAction};
use rand::Rng;
use snapshot::{Snapshot};
use api::{EthProtocolInfo as PeerInfoDigest, WARP_SYNC_PROTOCOL_ID, PriorityTask};
use api::{EthProtocolInfo as PeerInfoDigest, ETH_PROTOCOL, WARP_SYNC_PROTOCOL_ID, PriorityTask};
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should rename WARP_SYNC_PROTOCOL_ID to WARP_SYNC_PROTOCOL (as there are already ETH_PROTOCOL and LIGHT_PROTOCOL)

use private_tx::PrivateTxHandler;
use transactions_stats::{TransactionsStats, Stats as TransactionStats};
use types::transaction::UnverifiedTransaction;
Expand Down Expand Up @@ -153,7 +153,7 @@ const MAX_TRANSACTION_PACKET_SIZE: usize = 5 * 1024 * 1024;
const SNAPSHOT_RESTORE_THRESHOLD: BlockNumber = 30000;
const SNAPSHOT_MIN_PEERS: usize = 3;

const STATUS_PACKET: u8 = 0x00;
pub const STATUS_PACKET: u8 = 0x00;
const NEW_BLOCK_HASHES_PACKET: u8 = 0x01;
const TRANSACTIONS_PACKET: u8 = 0x02;
pub const GET_BLOCK_HEADERS_PACKET: u8 = 0x03;
Expand Down Expand Up @@ -481,7 +481,7 @@ impl ChainSyncApi {
for peers in sync.get_peers(&chain_info, PeerState::SameBlock).chunks(10) {
check_deadline(deadline)?;
for peer in peers {
SyncPropagator::send_packet(io, *peer, NEW_BLOCK_PACKET, rlp.clone());
SyncPropagator::send_packet(io, ETH_PROTOCOL, *peer, NEW_BLOCK_PACKET, rlp.clone());
if let Some(ref mut peer) = sync.peers.get_mut(peer) {
peer.latest_hash = hash;
}
Expand Down Expand Up @@ -1328,8 +1328,8 @@ impl ChainSync {
}

/// Broadcast private transaction message to peers.
pub fn propagate_private_transaction(&mut self, io: &mut SyncIo, transaction_hash: H256, packet_id: PacketId, packet: Bytes) {
SyncPropagator::propagate_private_transaction(self, io, transaction_hash, packet_id, packet);
pub fn propagate_private_transaction(&mut self, io: &mut SyncIo, transaction_hash: H256, protocol: ProtocolId, packet_id: PacketId, packet: Bytes) {
elferdo marked this conversation as resolved.
Show resolved Hide resolved
SyncPropagator::propagate_private_transaction(self, io, transaction_hash, protocol, packet_id, packet);
}
}

Expand Down
23 changes: 12 additions & 11 deletions ethcore/sync/src/chain/propagator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,11 @@
use std::cmp;
use std::collections::HashSet;

use api::{ETH_PROTOCOL, WARP_SYNC_PROTOCOL_ID};
use bytes::Bytes;
use ethereum_types::H256;
use fastmap::H256FastSet;
use network::{PeerId, PacketId};
use network::{PeerId, PacketId,ProtocolId};
use rand::Rng;
use rlp::{Encodable, RlpStream};
use sync_io::SyncIo;
Expand Down Expand Up @@ -49,7 +50,7 @@ fn accepts_service_transaction(client_id: &str) -> bool {
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('.')
Expand All @@ -74,7 +75,7 @@ impl SyncPropagator {
let sent = peers.len();
let mut send_packet = |io: &mut SyncIo, rlp: Bytes| {
for peer_id in peers {
SyncPropagator::send_packet(io, *peer_id, NEW_BLOCK_PACKET, rlp.clone());
SyncPropagator::send_packet(io, ETH_PROTOCOL, *peer_id, NEW_BLOCK_PACKET, rlp.clone());
if let Some(ref mut peer) = sync.peers.get_mut(peer_id) {
peer.latest_hash = chain_info.best_block_hash.clone();
}
Expand Down Expand Up @@ -109,7 +110,7 @@ impl SyncPropagator {
if let Some(ref mut peer) = sync.peers.get_mut(peer_id) {
peer.latest_hash = best_block_hash;
}
SyncPropagator::send_packet(io, *peer_id, NEW_BLOCK_HASHES_PACKET, rlp.clone());
SyncPropagator::send_packet(io, ETH_PROTOCOL, *peer_id, NEW_BLOCK_HASHES_PACKET, rlp.clone());
}
sent
}
Expand Down Expand Up @@ -177,7 +178,7 @@ impl SyncPropagator {

let send_packet = |io: &mut SyncIo, peer_id: PeerId, sent: usize, rlp: Bytes| {
let size = rlp.len();
SyncPropagator::send_packet(io, peer_id, TRANSACTIONS_PACKET, rlp);
SyncPropagator::send_packet(io, ETH_PROTOCOL, peer_id, TRANSACTIONS_PACKET, rlp);
trace!(target: "sync", "{:02} <- Transactions ({} entries; {} bytes)", peer_id, sent, size);
};

Expand Down Expand Up @@ -296,7 +297,7 @@ impl SyncPropagator {
io.chain().chain_info().total_difficulty
);
for peer_id in &peers {
SyncPropagator::send_packet(io, *peer_id, NEW_BLOCK_PACKET, rlp.clone());
SyncPropagator::send_packet(io, ETH_PROTOCOL, *peer_id, NEW_BLOCK_PACKET, rlp.clone());
}
}
}
Expand All @@ -306,12 +307,12 @@ impl SyncPropagator {
let lucky_peers = ChainSync::select_random_peers(&sync.get_consensus_peers());
trace!(target: "sync", "Sending consensus packet to {:?}", lucky_peers);
for peer_id in lucky_peers {
SyncPropagator::send_packet(io, peer_id, CONSENSUS_DATA_PACKET, packet.clone());
SyncPropagator::send_packet(io, WARP_SYNC_PROTOCOL_ID, peer_id, CONSENSUS_DATA_PACKET, packet.clone());
}
}

/// Broadcast private transaction message to peers.
pub fn propagate_private_transaction(sync: &mut ChainSync, io: &mut SyncIo, transaction_hash: H256, packet_id: PacketId, packet: Bytes) {
pub fn propagate_private_transaction(sync: &mut ChainSync, io: &mut SyncIo, transaction_hash: H256, protocol: ProtocolId, packet_id: PacketId, packet: Bytes) {
let lucky_peers = ChainSync::select_random_peers(&sync.get_private_transaction_peers(&transaction_hash));
if lucky_peers.is_empty() {
error!(target: "privatetx", "Cannot propagate the packet, no peers with private tx enabled connected");
Expand All @@ -321,7 +322,7 @@ impl SyncPropagator {
if let Some(ref mut peer) = sync.peers.get_mut(&peer_id) {
peer.last_sent_private_transactions.insert(transaction_hash);
}
SyncPropagator::send_packet(io, peer_id, packet_id, packet.clone());
SyncPropagator::send_packet(io, protocol, peer_id, packet_id, packet.clone());
}
}
}
Expand All @@ -342,8 +343,8 @@ impl SyncPropagator {
}

/// Generic packet sender
pub fn send_packet(sync: &mut SyncIo, peer_id: PeerId, packet_id: PacketId, packet: Bytes) {
if let Err(e) = sync.send(peer_id, packet_id, packet) {
pub fn send_packet(sync: &mut SyncIo, protocol: ProtocolId, peer_id: PeerId, packet_id: PacketId, packet: Bytes) {
if let Err(e) = sync.send_protocol(protocol, peer_id, packet_id, packet) {
elferdo marked this conversation as resolved.
Show resolved Hide resolved
debug!(target:"sync", "Error sending packet: {:?}", e);
sync.disconnect_peer(peer_id);
}
Expand Down
28 changes: 12 additions & 16 deletions ethcore/sync/src/chain/requester.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,11 @@
// You should have received a copy of the GNU General Public License
// along with Parity Ethereum. If not, see <http://www.gnu.org/licenses/>.

use api::WARP_SYNC_PROTOCOL_ID;
use api::{ETH_PROTOCOL, WARP_SYNC_PROTOCOL_ID};
use block_sync::BlockRequest;
use bytes::Bytes;
use ethereum_types::H256;
use network::{PeerId, PacketId};
use network::{PeerId, PacketId, ProtocolId};
use rlp::RlpStream;
use std::time::Instant;
use sync_io::SyncIo;
Expand All @@ -28,7 +28,6 @@ use super::{
BlockSet,
ChainSync,
PeerAsking,
ETH_PROTOCOL_VERSION_63,
GET_BLOCK_BODIES_PACKET,
GET_BLOCK_HEADERS_PACKET,
GET_RECEIPTS_PACKET,
Expand Down Expand Up @@ -62,7 +61,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());
SyncRequester::send_request(sync, io, peer_id, PeerAsking::BlockBodies, ETH_PROTOCOL, 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;
peer.block_set = Some(set);
Expand All @@ -76,7 +75,7 @@ impl SyncRequester {
rlp.append(&1u32);
rlp.append(&0u32);
rlp.append(&0u32);
SyncRequester::send_request(sync, io, peer_id, PeerAsking::ForkHeader, GET_BLOCK_HEADERS_PACKET, rlp.out());
SyncRequester::send_request(sync, io, peer_id, PeerAsking::ForkHeader, ETH_PROTOCOL, GET_BLOCK_HEADERS_PACKET, rlp.out());
}

/// Find some headers or blocks to download for a peer.
Expand All @@ -94,7 +93,7 @@ impl SyncRequester {
pub fn request_snapshot_manifest(sync: &mut ChainSync, io: &mut SyncIo, peer_id: PeerId) {
trace!(target: "sync", "{} <- GetSnapshotManifest", peer_id);
let rlp = RlpStream::new_list(0);
SyncRequester::send_request(sync, io, peer_id, PeerAsking::SnapshotManifest, GET_SNAPSHOT_MANIFEST_PACKET, rlp.out());
SyncRequester::send_request(sync, io, peer_id, PeerAsking::SnapshotManifest, WARP_SYNC_PROTOCOL_ID, GET_SNAPSHOT_MANIFEST_PACKET, rlp.out());
}

/// Request headers from a peer by block hash
Expand All @@ -105,7 +104,7 @@ impl SyncRequester {
rlp.append(&count);
rlp.append(&skip);
rlp.append(&if reverse {1u32} else {0u32});
SyncRequester::send_request(sync, io, peer_id, PeerAsking::BlockHeaders, GET_BLOCK_HEADERS_PACKET, rlp.out());
SyncRequester::send_request(sync, io, peer_id, PeerAsking::BlockHeaders, ETH_PROTOCOL, GET_BLOCK_HEADERS_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_hash = Some(h.clone());
peer.block_set = Some(set);
Expand All @@ -118,7 +117,7 @@ impl SyncRequester {
for h in &hashes {
rlp.append(&h.clone());
}
SyncRequester::send_request(sync, io, peer_id, PeerAsking::BlockReceipts, GET_RECEIPTS_PACKET, rlp.out());
SyncRequester::send_request(sync, io, peer_id, PeerAsking::BlockReceipts, ETH_PROTOCOL, GET_RECEIPTS_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;
peer.block_set = Some(set);
Expand All @@ -129,23 +128,20 @@ impl SyncRequester {
trace!(target: "sync", "{} <- GetSnapshotData {:?}", peer_id, chunk);
let mut rlp = RlpStream::new_list(1);
rlp.append(chunk);
SyncRequester::send_request(sync, io, peer_id, PeerAsking::SnapshotData, GET_SNAPSHOT_DATA_PACKET, rlp.out());
SyncRequester::send_request(sync, io, peer_id, PeerAsking::SnapshotData, WARP_SYNC_PROTOCOL_ID, GET_SNAPSHOT_DATA_PACKET, rlp.out());
}

/// Generic request sender
fn send_request(sync: &mut ChainSync, io: &mut SyncIo, peer_id: PeerId, asking: PeerAsking, packet_id: PacketId, packet: Bytes) {
fn send_request(sync: &mut ChainSync, io: &mut SyncIo, peer_id: PeerId, asking: PeerAsking, protocol: ProtocolId, packet_id: PacketId, packet: Bytes) {
if let Some(ref mut peer) = sync.peers.get_mut(&peer_id) {
if peer.asking != PeerAsking::Nothing {
warn!(target:"sync", "Asking {:?} while requesting {:?}", peer.asking, asking);
}
peer.asking = asking;
peer.ask_time = Instant::now();
// TODO [ToDr] This seems quite fragile. Be careful when protocol is updated.
let result = if packet_id >= ETH_PROTOCOL_VERSION_63.1 {
elferdo marked this conversation as resolved.
Show resolved Hide resolved
io.send_protocol(WARP_SYNC_PROTOCOL_ID, peer_id, packet_id, packet)
} else {
io.send(peer_id, packet_id, packet)
};

let result = io.send_protocol(protocol, peer_id, packet_id, packet);

if let Err(e) = result {
debug!(target:"sync", "Error sending request: {:?}", e);
io.disconnect_peer(peer_id);
Expand Down
6 changes: 0 additions & 6 deletions ethcore/sync/src/sync_io.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,6 @@ pub trait SyncIo {
fn disconnect_peer(&mut self, peer_id: PeerId);
/// Respond to current request with a packet. Can be called from an IO handler for incoming packet.
fn respond(&mut self, packet_id: PacketId, data: Vec<u8>) -> Result<(), Error>;
/// Send a packet to a peer.
fn send(&mut self, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), Error>;
/// Send a packet to a peer using specified protocol.
fn send_protocol(&mut self, protocol: ProtocolId, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), Error>;
/// Get the blockchain
Expand Down Expand Up @@ -98,10 +96,6 @@ impl<'s> SyncIo for NetSyncIo<'s> {
self.network.respond(packet_id, data)
}

fn send(&mut self, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), Error>{
self.network.send(peer_id, packet_id, data)
}

fn send_protocol(&mut self, protocol: ProtocolId, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), Error>{
self.network.send_protocol(protocol, peer_id, packet_id, data)
}
Expand Down
26 changes: 17 additions & 9 deletions ethcore/sync/src/tests/helpers.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,8 @@ use ethcore::miner::Miner;
use ethcore::test_helpers;
use sync_io::SyncIo;
use io::{IoChannel, IoContext, IoHandler};
use api::WARP_SYNC_PROTOCOL_ID;
use chain::{ChainSync, ETH_PROTOCOL_VERSION_63, PAR_PROTOCOL_VERSION_3, PRIVATE_TRANSACTION_PACKET, SIGNED_PRIVATE_TRANSACTION_PACKET, SyncSupplier};
use api::{ETH_PROTOCOL, WARP_SYNC_PROTOCOL_ID};
use chain::{ChainSync, ETH_PROTOCOL_VERSION_63, PAR_PROTOCOL_VERSION_3, PRIVATE_TRANSACTION_PACKET, SyncSupplier, STATUS_PACKET, RECEIPTS_PACKET, GET_SNAPSHOT_MANIFEST_PACKET, SIGNED_PRIVATE_TRANSACTION_PACKET};
use SyncConfig;
use private_tx::SimplePrivateTxHandler;
use types::BlockNumber;
Expand Down Expand Up @@ -80,6 +80,16 @@ impl<'p, C> Drop for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
}
}

fn assert_packet_id_matches_protocol(protocol: &ProtocolId, packet_id: &PacketId) {
match packet_id {
elferdo marked this conversation as resolved.
Show resolved Hide resolved
STATUS_PACKET ... RECEIPTS_PACKET => assert_eq!(*protocol, ETH_PROTOCOL),
GET_SNAPSHOT_MANIFEST_PACKET ... SIGNED_PRIVATE_TRANSACTION_PACKET => assert_eq!(*protocol, WARP_SYNC_PROTOCOL_ID),
// What about light?
_ => assert!(false)
}
}


impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
fn disable_peer(&mut self, peer_id: PeerId) {
self.disconnect_peer(peer_id);
Expand All @@ -102,7 +112,9 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
Ok(())
}

fn send(&mut self, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), network::Error> {
fn send_protocol(&mut self, protocol: ProtocolId, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), network::Error> {
assert_packet_id_matches_protocol(&protocol, &packet_id);

self.packets.push(TestPacket {
data: data,
packet_id: packet_id,
Expand All @@ -111,10 +123,6 @@ impl<'p, C> SyncIo for TestIo<'p, C> where C: FlushingBlockChainClient, C: 'p {
Ok(())
}

fn send_protocol(&mut self, _protocol: ProtocolId, peer_id: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), network::Error> {
self.send(peer_id, packet_id, data)
}

fn chain(&self) -> &BlockChainClient {
&*self.chain
}
Expand Down Expand Up @@ -234,9 +242,9 @@ impl<C> EthPeer<C> where C: FlushingBlockChainClient {
match message {
ChainMessageType::Consensus(data) => self.sync.write().propagate_consensus_packet(&mut io, data),
ChainMessageType::PrivateTransaction(transaction_hash, data) =>
self.sync.write().propagate_private_transaction(&mut io, transaction_hash, PRIVATE_TRANSACTION_PACKET, data),
self.sync.write().propagate_private_transaction(&mut io, transaction_hash, WARP_SYNC_PROTOCOL_ID, PRIVATE_TRANSACTION_PACKET, data),
ChainMessageType::SignedPrivateTransaction(transaction_hash, data) =>
self.sync.write().propagate_private_transaction(&mut io, transaction_hash, SIGNED_PRIVATE_TRANSACTION_PACKET, data),
self.sync.write().propagate_private_transaction(&mut io, transaction_hash, WARP_SYNC_PROTOCOL_ID, SIGNED_PRIVATE_TRANSACTION_PACKET, data),
}
}

Expand Down