Consensus messages are now targeted at NodeIDs (H512) instead of the internally used peer_id, to make sure that they are sent to the correct peer.

DMDcoin#248

Author: SurfingNerd

crates/ethcore/src/client/chain_notify.rs

@@ -184,7 +184,7 @@ pub trait ChainNotify: Send + Sync {
     }
 
     /// fires when chain sends a message to a specific peer
-    fn send(&self, _message_type: ChainMessageType, _node_id: Option<H512>) {
+    fn send(&self, _message_type: ChainMessageType, _node_id: &H512) {
         // does nothing by default
     }
 

crates/ethcore/src/client/client.rs

@@ -3307,7 +3307,9 @@ impl super::traits::EngineClient for Client {
             .sent_consensus_messages_bytes
             .fetch_add(message.len() as u64, std::sync::atomic::Ordering::Relaxed);
 
-        self.notify(|notify| notify.send(ChainMessageType::Consensus(message.clone()), node_id));
+        if let Some(n) = node_id {
+            self.notify(|notify| notify.send(ChainMessageType::Consensus(message.clone()), &n));
+        }
     }
 
     fn epoch_transition_for(&self, parent_hash: H256) -> Option<crate::engines::EpochTransition> {

crates/ethcore/src/test_helpers.rs

@@ -695,11 +695,13 @@ impl ChainNotify for TestNotify {
         self.messages.write().push(data);
     }
 
-    fn send(&self, message: ChainMessageType, node_id: Option<H512>) {
+    fn send(&self, message: ChainMessageType, node_id: &H512) {
         let data = match message {
             ChainMessageType::Consensus(data) => data,
         };
-        self.targeted_messages.write().push((data, node_id));
+        self.targeted_messages
+            .write()
+            .push((data, Some(node_id.clone())));
     }
 }
 

crates/ethcore/sync/src/api.rs

@@ -494,24 +494,18 @@ impl SyncProtocolHandler {
         let mut sync_io = NetSyncIo::new(nc, &*self.chain, &*self.snapshot_service, &self.overlay);
 
         for node_id in pub_keys.iter() {
-            if let Some(peer_id) = nc.node_id_to_peer_id(*node_id) {
+            if let Some(peer_id) = nc.node_id_to_peer_id(node_id) {
                 let found_peers = self.sync.peer_info(&[peer_id]);
                 if let Some(peer_info) = found_peers.first() {
                     if let Some(_) = peer_info {
-                        self.send_cached_consensus_messages_for(&mut sync_io, node_id, peer_id);
+                        self.send_cached_consensus_messages_for(&mut sync_io, node_id);
                     }
                 }
             }
         }
     }
 
-    fn send_cached_consensus_messages_for(
-        &self,
-        sync_io: &mut dyn SyncIo,
-        node_id: &NodeId,
-        peer_id: PeerId,
-    ) {
-        
+    fn send_cached_consensus_messages_for(&self, sync_io: &mut dyn SyncIo, node_id: &NodeId) {
         // now since we are connected, lets send any cached messages
         if let Some(vec_msg) = self.message_cache.write().remove(&Some(*node_id)) {
             trace!(target: "consensus", "Cached Messages: Trying to send cached messages to {:?}", node_id);
@@ -521,17 +515,18 @@ impl SyncProtocolHandler {
             for msg in vec_msg {
                 match msg {
                     ChainMessageType::Consensus(message) => {
-                        let send_consensus_result = self
-                        .sync
-                        .write()
-                        .send_consensus_packet(sync_io, message.clone(), peer_id);
+                        let send_consensus_result = self.sync.write().send_consensus_packet(
+                            sync_io,
+                            message.clone(),
+                            node_id,
+                        );
 
                         match send_consensus_result {
-                            Ok(_) => {},
+                            Ok(_) => {}
                             Err(e) => {
-                                info!(target: "consensus", "Error sending cached consensus message to peer (re-adding) {:?}: {:?}", peer_id, e);
+                                info!(target: "consensus", "Error sending cached consensus message to peer (re-adding) {:?}: {:?}", node_id, e);
                                 failed_messages.push(ChainMessageType::Consensus(message));
-                            },
+                            }
                         }
                     }
                 }
@@ -540,7 +535,9 @@ impl SyncProtocolHandler {
             if !failed_messages.is_empty() {
                 // If we failed to send some messages, cache them for later
                 let mut lock = self.message_cache.write();
-                lock.entry(Some(*node_id)).or_default().extend(failed_messages);
+                lock.entry(Some(*node_id))
+                    .or_default()
+                    .extend(failed_messages);
             } else {
                 trace!(target: "consensus", "Cached Messages: Successfully sent all cached messages to {:?}", node_id);
             }
@@ -736,41 +733,21 @@ impl ChainNotify for EthSync {
         });
     }
 
-    fn send(&self, message_type: ChainMessageType, node_id: Option<H512>) {
+    fn send(&self, message_type: ChainMessageType, node_id: &H512) {
         self.network.with_context(PAR_PROTOCOL, |context| {
-            let peer_ids = self.network.connected_peers();
-            let target_peer_id = peer_ids.into_iter().find(|p| {
-                match context.session_info(*p){
-                    Some(session_info) => {
-                        session_info.id == node_id
-                    },
-                    None => { warn!(target:"sync", "No session exists for peerId {:?} Node: {:?}", p, node_id); false},
-                }
-            });
-
-            let my_peer_id = match target_peer_id {
-                None => {
-                    trace!(target: "consensus", "Cached Messages: peer {:?} not connected, caching message...", node_id);
-                    let mut lock = self.eth_handler.message_cache.write();
-                    lock.entry(node_id.clone()).or_default().push(message_type);
-                    return;
-                }
-                Some(n) => n,
-            };
-
             let mut sync_io = NetSyncIo::new(context,
                                              &*self.eth_handler.chain,
                                              &*self.eth_handler.snapshot_service,
                                              &self.eth_handler.overlay);
 
             match message_type {
                 ChainMessageType::Consensus(message) => {
-                    let send_result = self.eth_handler.sync.write().send_consensus_packet(&mut sync_io, message.clone(), my_peer_id);
+                    let send_result = self.eth_handler.sync.write().send_consensus_packet(&mut sync_io, message.clone(), node_id);
                     if let Err(e) = send_result {
-                        info!(target: "consensus", "Error sending consensus message to peer - caching message {:?}: {:?}", my_peer_id, e);
+                        info!(target: "consensus", "Error sending consensus message to peer - caching message {:?}: {:?}", node_id, e);
                         // If we failed to send the message, cache it for later
                         let mut lock = self.eth_handler.message_cache.write();
-                        lock.entry(node_id.clone()).or_default().push(ChainMessageType::Consensus(message));
+                        lock.entry(Some(node_id.clone())).or_default().push(ChainMessageType::Consensus(message));
                     }
                 },
             }

crates/ethcore/sync/src/chain/mod.rs

@@ -577,7 +577,8 @@ impl ChainSyncApi {
                     for peers in sync.get_peers(&chain_info, PeerState::SameBlock).chunks(10) {
                         check_deadline(deadline)?;
                         for peer in peers {
-                            let send_result = ChainSync::send_packet(io, *peer, NewBlockPacket, rlp.clone());
+                            let send_result =
+                                ChainSync::send_packet(io, *peer, NewBlockPacket, rlp.clone());
                             if send_result.is_ok() {
                                 if let Some(ref mut peer) = sync.peers.get_mut(peer) {
                                     peer.latest_hash = hash;

crates/ethcore/sync/src/chain/propagator.rs

@@ -25,9 +25,9 @@ use crate::{
     types::{BlockNumber, blockchain_info::BlockChainInfo, transaction::SignedTransaction},
 };
 use bytes::Bytes;
-use ethereum_types::H256;
+use ethereum_types::{H256, H512};
 use fastmap::H256FastSet;
-use network::{client_version::ClientCapabilities, Error, PeerId};
+use network::{Error, PeerId, client_version::ClientCapabilities};
 use rand::RngCore;
 use rlp::RlpStream;
 
@@ -202,7 +202,6 @@ impl ChainSync {
                            rlp: Bytes| {
             let size = rlp.len();
 
-
             let send_result = ChainSync::send_packet(
                 io,
                 peer_id,
@@ -214,17 +213,14 @@ impl ChainSync {
                 rlp,
             );
 
-
-
             if send_result.is_ok() {
                 if is_hashes {
                     stats.log_propagated_hashes(sent, size);
                 } else {
                     stats.log_propagated_transactions(sent, size);
-                }    
+                }
                 trace!(target: "sync", "{:02} <- {} ({} entries; {} bytes)", peer_id, if is_hashes { "NewPooledTransactionHashes" } else { "Transactions" }, sent, size);
             }
-            
         };
 
         let mut sent_to_peers = HashSet::new();
@@ -403,32 +399,46 @@ impl ChainSync {
         let lucky_peers = ChainSync::select_random_peers(&self.get_consensus_peers());
         trace!(target: "sync", "Sending consensus packet to {:?}", lucky_peers);
 
-        self.statistics
-            .log_consensus_broadcast(lucky_peers.len(), packet.len());
+        let mut num_sent_messages = 0;
         for peer_id in lucky_peers {
-            let send_result = ChainSync::send_packet(io, peer_id, ConsensusDataPacket, packet.clone());
+            let send_result =
+                ChainSync::send_packet(io, peer_id, ConsensusDataPacket, packet.clone());
 
             if let Err(e) = send_result {
                 info!(target: "sync", "Error broadcast consensus packet to peer {}: {:?}", peer_id, e);
+            } else {
+                num_sent_messages += 1;
             }
         }
+
+        self.statistics
+            .log_consensus_broadcast(num_sent_messages, packet.len());
     }
 
+    /// Sends a packet to a specific peer.
+    /// The caller has to take care about Errors, and reshedule if an error occurs.
     pub(crate) fn send_consensus_packet(
         &mut self,
         io: &mut dyn SyncIo,
         packet: Bytes,
-        peer_id: usize,
+        peer: &H512,
     ) -> Result<(), Error> {
+        let peer_id = match io.node_id_to_peer_id(peer) {
+            Some(id) => id,
+            None => {
+                warn!(target: "sync", "Peer with node id {} not found in peers list.", peer);
+                return Err("No Session for Peer".into());
+            }
+        };
         let packet_len = packet.len();
         let send_result = ChainSync::send_packet(io, peer_id, ConsensusDataPacket, packet.clone());
         match &send_result {
             Ok(_) => {
-                self.statistics.log_consensus(peer_id, packet_len);
-            },
+                self.statistics.log_consensus(packet_len);
+            }
             Err(e) => {
                 warn!(target: "sync", "Error sending consensus packet to peer {}: {:?}", peer_id, e);
-            },
+            }
         }
         return send_result;
     }
@@ -467,7 +477,7 @@ impl ChainSync {
         packet: Bytes,
     ) -> Result<(), Error> {
         let result = sync.send(peer_id, packet_id, packet);
-        if let  Err(e) = &result {
+        if let Err(e) = &result {
             debug!(target:"sync", "Error sending packet: {:?}", e);
             sync.disconnect_peer(peer_id);
         }

crates/ethcore/sync/src/chain/propagator_statistics.rs

@@ -79,7 +79,7 @@ impl SyncPropagatorStatistics {
         }
     }
 
-    pub(crate) fn log_consensus(&mut self, _peer_id: usize, bytelen: usize) {
+    pub(crate) fn log_consensus(&mut self, bytelen: usize) {
         if self.logging_enabled {
             self.consensus_bytes += bytelen as i64;
             self.consensus_packages += 1;

crates/ethcore/sync/src/sync_io.rs

@@ -20,6 +20,7 @@ use crate::{
 };
 use bytes::Bytes;
 use ethcore::{client::BlockChainClient, snapshot::SnapshotService};
+use ethereum_types::H512;
 use network::{
     Error, NetworkContext, PacketId, PeerId, ProtocolId, SessionInfo, client_version::ClientVersion,
 };
@@ -58,6 +59,9 @@ pub trait SyncIo {
     fn is_expired(&self) -> bool;
     /// Return sync overlay
     fn chain_overlay(&self) -> &RwLock<HashMap<BlockNumber, Bytes>>;
+
+    /// Returns the peer ID for a given node id, if a corresponding peer exists.
+    fn node_id_to_peer_id(&self, node_id: &H512) -> Option<PeerId>;
 }
 
 /// Wraps `NetworkContext` and the blockchain client
@@ -132,4 +136,8 @@ impl<'s> SyncIo for NetSyncIo<'s> {
     fn peer_version(&self, peer_id: PeerId) -> ClientVersion {
         self.network.peer_client_version(peer_id)
     }
+
+    fn node_id_to_peer_id(&self, node_id: &H512) -> Option<PeerId> {
+        self.network.node_id_to_peer_id(node_id)
+    }
 }

crates/ethcore/sync/src/tests/helpers.rs

@@ -181,6 +181,10 @@ where
     fn chain_overlay(&self) -> &RwLock<HashMap<BlockNumber, Bytes>> {
         &self.overlay
     }
+
+    fn node_id_to_peer_id(&self, node_id: &ethereum_types::H512) -> Option<PeerId> {
+        return Some(node_id.to_low_u64_le() as PeerId);
+    }
 }
 
 /// Mock for emulution of async run of new blocks

crates/net/network-devp2p/src/host.rs

@@ -296,14 +296,17 @@ impl<'s> NetworkContextTrait for NetworkContext<'s> {
             .unwrap_or(false)
     }
 
-    fn node_id_to_peer_id(&self, node_id: NodeId) -> Option<PeerId> {
+    fn node_id_to_peer_id(&self, node_id: &NodeId) -> Option<PeerId> {
         let sessions = self.sessions.read();
         let sessions = &*sessions;
 
         for i in (0..MAX_SESSIONS).map(|x| x + FIRST_SESSION) {
             if let Some(session) = sessions.get(i) {
-                if session.lock().info.id == Some(node_id) {
-                    return Some(i);
+                let session_node_id_o = session.lock().info.id;
+                if let Some(session_node_id) = session_node_id_o {
+                    if session_node_id == *node_id {
+                        return Some(i);
+                    }
                 }
             }
         }