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

Zombie connections

Browse Source
This commit is contained in:
arkpar
2016-02-20 01:10:27 +01:00
parent 69df91de68
commit b1bfd00875
6 changed files with 120 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -190,11 +190,11 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
let session = { self.sessions.read().unwrap().get(peer).cloned() };
if let Some(session) = session {
session.lock().unwrap().deref_mut().send_packet(self.protocol, packet_id as u8, &data).unwrap_or_else(|e| {
warn!(target: "net", "Send error: {:?}", e);
warn!(target: "network", "Send error: {:?}", e);
}); //TODO: don't copy vector data
try!(self.io.update_registration(peer));
} else {
trace!(target: "net", "Send: Peer no longer exist")
trace!(target: "network", "Send: Peer no longer exist")
}
Ok(())
}
@@ -470,18 +470,18 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
.take(min(MAX_HANDSHAKES_PER_ROUND, handshake_limit - handshake_count)) {
self.connect_peer(&id, io);
}
debug!(target: "net", "Connecting peers: {} sessions, {} pending", self.session_count(), self.handshake_count());
debug!(target: "network", "Connecting peers: {} sessions, {} pending", self.session_count(), self.handshake_count());
}
#[cfg_attr(feature="dev", allow(single_match))]
fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage<Message>>) {
if self.have_session(id)
{
trace!("Aborted connect. Node already connected.");
trace!(target: "network", "Aborted connect. Node already connected.");
return;
}
if self.connecting_to(id) {
trace!("Aborted connect. Node already connecting.");
trace!(target: "network", "Aborted connect. Node already connecting.");
return;
}
@@ -493,7 +493,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
node.endpoint.address
}
else {
debug!("Connection to expired node aborted");
debug!(target: "network", "Connection to expired node aborted");
return;
}
};
@@ -515,16 +515,16 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
if handshakes.insert_with(|token| {
let mut handshake = Handshake::new(token, id, socket, &nonce, self.stats.clone()).expect("Can't create handshake");
handshake.start(io, &self.info.read().unwrap(), id.is_some()).and_then(|_| io.register_stream(token)).unwrap_or_else (|e| {
debug!(target: "net", "Handshake create error: {:?}", e);
debug!(target: "network", "Handshake create error: {:?}", e);
});
Arc::new(Mutex::new(handshake))
}).is_none() {
debug!("Max handshakes reached");
debug!(target: "network", "Max handshakes reached");
}
}
fn accept(&self, io: &IoContext<NetworkIoMessage<Message>>) {
trace!(target: "net", "accept");
trace!(target: "network", "Accepting incoming connection");
loop {
let socket = match self.tcp_listener.lock().unwrap().accept() {
Ok(None) => break,
@@ -544,7 +544,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
if let Some(handshake) = handshake {
let mut h = handshake.lock().unwrap();
if let Err(e) = h.writable(io, &self.info.read().unwrap()) {
trace!(target: "net", "Handshake write error: {}: {:?}", token, e);
trace!(target: "network", "Handshake write error: {}: {:?}", token, e);
}
}
}
@@ -554,9 +554,9 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
if let Some(session) = session {
let mut s = session.lock().unwrap();
if let Err(e) = s.writable(io, &self.info.read().unwrap()) {
trace!(target: "net", "Session write error: {}: {:?}", token, e);
trace!(target: "network", "Session write error: {}: {:?}", token, e);
}
io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Session registration error: {:?}", e));
io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Session registration error: {:?}", e));
}
}
@@ -571,7 +571,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
if let Some(handshake) = handshake {
let mut h = handshake.lock().unwrap();
if let Err(e) = h.readable(io, &self.info.read().unwrap()) {
debug!(target: "net", "Handshake read error: {}: {:?}", token, e);
debug!(target: "network", "Handshake read error: {}: {:?}", token, e);
kill = true;
}
if h.done() {
@@ -585,7 +585,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
self.start_session(token, io);
return;
}
io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Token registration error: {:?}", e));
io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Token registration error: {:?}", e));
}
fn session_readable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
@@ -597,7 +597,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
let mut s = session.lock().unwrap();
match s.readable(io, &self.info.read().unwrap()) {
Err(e) => {
debug!(target: "net", "Session read error: {}: {:?}", token, e);
debug!(target: "network", "Session read error: {}: {:?}", token, e);
kill = true;
},
Ok(SessionData::Ready) => {
@@ -613,7 +613,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
packet_id,
}) => {
match self.handlers.read().unwrap().get(protocol) {
None => { warn!(target: "net", "No handler found for protocol: {:?}", protocol) },
None => { warn!(target: "network", "No handler found for protocol: {:?}", protocol) },
Some(_) => packet_data = Some((protocol, packet_id, data)),
}
},
@@ -631,7 +631,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
h.read(&NetworkContext::new(io, p, Some(token), self.sessions.clone()), &token, packet_id, &data[1..]);
}
io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Token registration error: {:?}", e));
io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Token registration error: {:?}", e));
}
fn start_session(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
@@ -643,19 +643,24 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
// turn a handshake into a session
let mut sessions = self.sessions.write().unwrap();
let mut h = handshakes.get_mut(token).unwrap().lock().unwrap();
if h.expired {
return;
}
let originated = h.originated;
let mut session = match Session::new(&mut h, &self.info.read().unwrap()) {
Ok(s) => s,
Err(e) => {
debug!("Session creation error: {:?}", e);
debug!(target: "network", "Session creation error: {:?}", e);
return;
}
};
let result = sessions.insert_with(move |session_token| {
session.set_token(session_token);
io.deregister_stream(token).expect("Error deleting handshake registration");
h.set_expired();
io.register_stream(session_token).expect("Error creating session registration");
self.stats.inc_sessions();
trace!(target: "network", "Creating session {} -> {}", token, session_token);
if !originated {
// Add it no node table
if let Ok(address) = session.remote_addr() {
@@ -684,26 +689,34 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
FIRST_HANDSHAKE ... LAST_HANDSHAKE => {
let handshakes = self.handshakes.write().unwrap();
if let Some(handshake) = handshakes.get(token).cloned() {
failure_id = Some(handshake.lock().unwrap().id().clone());
let mut handshake = handshake.lock().unwrap();
if !handshake.expired() {
handshake.set_expired();
failure_id = Some(handshake.id().clone());
io.deregister_stream(token).expect("Error deregistering stream");
}
}
},
FIRST_SESSION ... LAST_SESSION => {
let sessions = self.sessions.write().unwrap();
if let Some(session) = sessions.get(token).cloned() {
let s = session.lock().unwrap();
if s.is_ready() {
for (p, _) in self.handlers.read().unwrap().iter() {
if s.have_capability(p) {
to_disconnect.push(p);
let mut s = session.lock().unwrap();
if !s.expired() {
if s.is_ready() {
for (p, _) in self.handlers.read().unwrap().iter() {
if s.have_capability(p) {
to_disconnect.push(p);
}
}
}
s.set_expired();
failure_id = Some(s.id().clone());
io.deregister_stream(token).expect("Error deregistering stream");
}
failure_id = Some(s.id().clone());
}
},
_ => {},
}
io.deregister_stream(token).expect("Error deregistering stream");
if let Some(id) = failure_id {
if remote {
self.nodes.write().unwrap().note_failure(&id);
@@ -758,11 +771,11 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
}
fn stream_hup(&self, io: &IoContext<NetworkIoMessage<Message>>, stream: StreamToken) {
trace!(target: "net", "Hup: {}", stream);
trace!(target: "network", "Hup: {}", stream);
match stream {
FIRST_SESSION ... LAST_SESSION => self.connection_closed(stream, io),
FIRST_HANDSHAKE ... LAST_HANDSHAKE => self.connection_closed(stream, io),
_ => warn!(target: "net", "Unexpected hup"),
_ => warn!(target: "network", "Unexpected hup"),
};
}
@@ -810,7 +823,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
},
_ => match self.timers.read().unwrap().get(&token).cloned() {
Some(timer) => match self.handlers.read().unwrap().get(timer.protocol).cloned() {
None => { warn!(target: "net", "No handler found for protocol: {:?}", timer.protocol) },
None => { warn!(target: "network", "No handler found for protocol: {:?}", timer.protocol) },
Some(h) => { h.timeout(&NetworkContext::new(io, timer.protocol, None, self.sessions.clone()), timer.token); }
},
None => { warn!("Unknown timer token: {}", token); } // timer is not registerd through us