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

Merge pull request #519 from ethcore/net-fix

Browse Source 
Fixed a race condition when connecting peer disconnects immediately
This commit is contained in:
Nikolay Volf 2016-02-26 14:48:15 +03:00
commit 9823e30c2e
2 changed files with 14 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
sync/src/chain.rs
View File

@ -555,7 +555,10 @@ impl ChainSync {
/// Called when a new peer is connected /// Called when a new peer is connected
pub fn on_peer_connected(&mut self, io: &mut SyncIo, peer: PeerId) { pub fn on_peer_connected(&mut self, io: &mut SyncIo, peer: PeerId) {
trace!(target: "sync", "== Connected {}", peer); trace!(target: "sync", "== Connected {}", peer);
self.send_status(io, peer); if let Err(e) = self.send_status(io) {
warn!(target:"sync", "Error sending status request: {:?}", e);
io.disable_peer(peer);
}
} }
/// Resume downloading /// Resume downloading
@ -887,7 +890,7 @@ impl ChainSync {
} }
/// Send Status message /// Send Status message
fn send_status(&mut self, io: &mut SyncIo, peer_id: PeerId) { fn send_status(&mut self, io: &mut SyncIo) -> Result<(), UtilError> {
let mut packet = RlpStream::new_list(5); let mut packet = RlpStream::new_list(5);
let chain = io.chain().chain_info(); let chain = io.chain().chain_info();
packet.append(&(PROTOCOL_VERSION as u32)); packet.append(&(PROTOCOL_VERSION as u32));
@ -895,11 +898,7 @@ impl ChainSync {
packet.append(&chain.total_difficulty); packet.append(&chain.total_difficulty);
packet.append(&chain.best_block_hash); packet.append(&chain.best_block_hash);
packet.append(&chain.genesis_hash); packet.append(&chain.genesis_hash);
//TODO: handle timeout for status request io.respond(STATUS_PACKET, packet.out())
if let Err(e) = io.send(peer_id, STATUS_PACKET, packet.out()) {
warn!(target:"sync", "Error sending status request: {:?}", e);
io.disable_peer(peer_id);
}
} }
/// Respond to GetBlockHeaders request /// Respond to GetBlockHeaders request

12
util/src/network/host.rs
View File

@ -579,7 +579,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
} }
if kill { if kill {
self.kill_connection(token, io, true); //TODO: mark connection as dead an check in kill_connection self.kill_connection(token, io, true);
return; return;
} else if create_session { } else if create_session {
self.start_session(token, io); self.start_session(token, io);
@ -621,7 +621,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
} }
if kill { if kill {
self.kill_connection(token, io, true); //TODO: mark connection as dead an check in kill_connection self.kill_connection(token, io, true);
} }
for p in ready_data { for p in ready_data {
let h = self.handlers.read().unwrap().get(p).unwrap().clone(); let h = self.handlers.read().unwrap().get(p).unwrap().clone();
@ -685,6 +685,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
fn kill_connection(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>, remote: bool) { fn kill_connection(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>, remote: bool) {
let mut to_disconnect: Vec<ProtocolId> = Vec::new(); let mut to_disconnect: Vec<ProtocolId> = Vec::new();
let mut failure_id = None; let mut failure_id = None;
let mut deregister = false;
match token { match token {
FIRST_HANDSHAKE ... LAST_HANDSHAKE => { FIRST_HANDSHAKE ... LAST_HANDSHAKE => {
let handshakes = self.handshakes.write().unwrap(); let handshakes = self.handshakes.write().unwrap();
@ -693,7 +694,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
if !handshake.expired() { if !handshake.expired() {
handshake.set_expired(); handshake.set_expired();
failure_id = Some(handshake.id().clone()); failure_id = Some(handshake.id().clone());
io.deregister_stream(token).expect("Error deregistering stream"); deregister = true;
} }
} }
}, },
@ -711,7 +712,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
s.set_expired(); s.set_expired();
failure_id = Some(s.id().clone()); failure_id = Some(s.id().clone());
io.deregister_stream(token).expect("Error deregistering stream"); deregister = true;
} }
} }
}, },
@ -726,6 +727,9 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
let h = self.handlers.read().unwrap().get(p).unwrap().clone(); let h = self.handlers.read().unwrap().get(p).unwrap().clone();
h.disconnected(&NetworkContext::new(io, p, Some(token), self.sessions.clone()), &token); h.disconnected(&NetworkContext::new(io, p, Some(token), self.sessions.clone()), &token);
} }
if deregister {
io.deregister_stream(token).expect("Error deregistering stream");
}
} }
fn update_nodes(&self, io: &IoContext<NetworkIoMessage<Message>>, node_changes: TableUpdates) { fn update_nodes(&self, io: &IoContext<NetworkIoMessage<Message>>, node_changes: TableUpdates) {