diff --git a/Cargo.lock b/Cargo.lock index 9a55766..aa6b4a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -428,7 +428,7 @@ dependencies = [ [[package]] name = "netapp" -version = "0.4.4" +version = "0.4.5" dependencies = [ "arc-swap", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index af3f1ab..9a69d66 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "netapp" -version = "0.4.4" +version = "0.4.5" authors = ["Alex Auvolat "] edition = "2018" license-file = "LICENSE" diff --git a/src/client.rs b/src/client.rs index 8227e8f..5c5a05b 100644 --- a/src/client.rs +++ b/src/client.rs @@ -102,13 +102,16 @@ impl ClientConn { netapp.connected_as_client(peer_id, conn.clone()); + let debug_name = format!("CLI {}", hex::encode(&peer_id[..8])); + tokio::spawn(async move { - let send_future = tokio::spawn(conn.clone().send_loop(query_recv, write)); + let debug_name_2 = debug_name.clone(); + let send_future = tokio::spawn(conn.clone().send_loop(query_recv, write, debug_name_2)); let conn2 = conn.clone(); let recv_future = tokio::spawn(async move { select! { - r = conn2.recv_loop(read) => r, + r = conn2.recv_loop(read, debug_name) => r, _ = await_exit(stop_recv_loop_recv) => Ok(()) } }); diff --git a/src/peering/fullmesh.rs b/src/peering/fullmesh.rs index 012c5a0..208cfe4 100644 --- a/src/peering/fullmesh.rs +++ b/src/peering/fullmesh.rs @@ -22,10 +22,10 @@ use crate::NodeID; const CONN_RETRY_INTERVAL: Duration = Duration::from_secs(30); const CONN_MAX_RETRIES: usize = 10; -const PING_INTERVAL: Duration = Duration::from_secs(10); +const PING_INTERVAL: Duration = Duration::from_secs(15); const LOOP_DELAY: Duration = Duration::from_secs(1); -const PING_TIMEOUT: Duration = Duration::from_secs(5); -const FAILED_PING_THRESHOLD: usize = 3; +const PING_TIMEOUT: Duration = Duration::from_secs(10); +const FAILED_PING_THRESHOLD: usize = 4; // -- Protocol messages -- @@ -60,11 +60,26 @@ struct PeerInfoInternal { all_addrs: Vec, state: PeerConnState, + last_send_ping: Option, last_seen: Option, ping: VecDeque, failed_pings: usize, } +impl PeerInfoInternal { + fn new(addr: SocketAddr, state: PeerConnState) -> Self { + Self { + addr, + all_addrs: vec![addr], + state, + last_send_ping: None, + last_seen: None, + ping: VecDeque::new(), + failed_pings: 0, + } + } +} + #[derive(Copy, Clone, Debug)] pub struct PeerInfo { /// The node's identifier (its public key) @@ -184,14 +199,7 @@ impl FullMeshPeeringStrategy { if id != netapp.id { known_hosts.list.insert( id, - PeerInfoInternal { - addr, - all_addrs: vec![addr], - state: PeerConnState::Waiting(0, Instant::now()), - last_seen: None, - ping: VecDeque::new(), - failed_pings: 0, - }, + PeerInfoInternal::new(addr, PeerConnState::Waiting(0, Instant::now())), ); } } @@ -199,14 +207,7 @@ impl FullMeshPeeringStrategy { if let Some(addr) = our_addr { known_hosts.list.insert( netapp.id, - PeerInfoInternal { - addr, - all_addrs: vec![addr], - state: PeerConnState::Ourself, - last_seen: None, - ping: VecDeque::new(), - failed_pings: 0, - }, + PeerInfoInternal::new(addr, PeerConnState::Ourself), ); } @@ -254,7 +255,7 @@ impl FullMeshPeeringStrategy { trace!("{}, {:?}", hex::encode(&id[..8]), info); match info.state { PeerConnState::Connected => { - let must_ping = match info.last_seen { + let must_ping = match info.last_send_ping { None => true, Some(t) => Instant::now() - t > PING_INTERVAL, }; @@ -274,9 +275,16 @@ impl FullMeshPeeringStrategy { }; // 2. Dispatch ping to hosts - trace!("to_ping: {} peers", to_retry.len()); - for id in to_ping { - tokio::spawn(self.clone().ping(id)); + trace!("to_ping: {} peers", to_ping.len()); + if !to_ping.is_empty() { + let mut known_hosts = self.known_hosts.write().unwrap(); + for id in to_ping.iter() { + known_hosts.list.get_mut(id).unwrap().last_send_ping = Some(Instant::now()); + } + drop(known_hosts); + for id in to_ping { + tokio::spawn(self.clone().ping(id)); + } } // 3. Try reconnects @@ -534,17 +542,9 @@ impl FullMeshPeeringStrategy { host.all_addrs.push(addr); } } else { - known_hosts.list.insert( - id, - PeerInfoInternal { - state: PeerConnState::Connected, - addr, - all_addrs: vec![addr], - last_seen: None, - ping: VecDeque::new(), - failed_pings: 0, - }, - ); + known_hosts + .list + .insert(id, PeerInfoInternal::new(addr, PeerConnState::Connected)); } } known_hosts.update_hash(); @@ -569,14 +569,7 @@ impl FullMeshPeeringStrategy { } else { PeerConnState::Waiting(0, Instant::now()) }; - PeerInfoInternal { - addr, - all_addrs: vec![addr], - state, - last_seen: None, - ping: VecDeque::new(), - failed_pings: 0, - } + PeerInfoInternal::new(addr, state) } } diff --git a/src/proto.rs b/src/proto.rs index e843bff..8f7e70f 100644 --- a/src/proto.rs +++ b/src/proto.rs @@ -1,4 +1,5 @@ use std::collections::{HashMap, VecDeque}; +use std::fmt::Write; use std::sync::Arc; use log::trace; @@ -94,6 +95,22 @@ impl SendQueue { fn is_empty(&self) -> bool { self.items.iter().all(|(_k, v)| v.is_empty()) } + fn dump(&self) -> String { + let mut ret = String::new(); + for (prio, q) in self.items.iter() { + for item in q.iter() { + write!( + &mut ret, + " [{} {} ({})]", + prio, + item.data.len() - item.cursor, + item.id + ) + .unwrap(); + } + } + ret + } } /// The SendLoop trait, which is implemented both by the client and the server @@ -110,6 +127,7 @@ pub(crate) trait SendLoop: Sync { self: Arc, mut msg_recv: mpsc::UnboundedReceiver<(RequestID, RequestPriority, Vec)>, mut write: BoxStreamWrite, + debug_name: String, ) -> Result<(), Error> where W: AsyncWriteExt + Unpin + Send + Sync, @@ -117,8 +135,15 @@ pub(crate) trait SendLoop: Sync { let mut sending = SendQueue::new(); let mut should_exit = false; while !should_exit || !sending.is_empty() { + trace!("send_loop({}): queue = {}", debug_name, sending.dump()); if let Ok((id, prio, data)) = msg_recv.try_recv() { - trace!("send_loop: got {}, {} bytes", id, data.len()); + trace!( + "send_loop({}): new message to send, id = {}, prio = {}, {} bytes", + debug_name, + id, + prio, + data.len() + ); sending.push(SendQueueItem { id, prio, @@ -127,7 +152,8 @@ pub(crate) trait SendLoop: Sync { }); } else if let Some(mut item) = sending.pop() { trace!( - "send_loop: sending bytes for {} ({} bytes, {} already sent)", + "send_loop({}): sending bytes for {} ({} bytes, {} already sent)", + debug_name, item.id, item.data.len(), item.cursor @@ -157,7 +183,13 @@ pub(crate) trait SendLoop: Sync { } else { let sth = msg_recv.recv().await; if let Some((id, prio, data)) = sth { - trace!("send_loop: got {}, {} bytes", id, data.len()); + trace!( + "send_loop({}): new message to send, id = {}, prio = {}, {} bytes", + debug_name, + id, + prio, + data.len() + ); sending.push(SendQueueItem { id, prio, @@ -186,13 +218,12 @@ pub(crate) trait SendLoop: Sync { pub(crate) trait RecvLoop: Sync + 'static { fn recv_handler(self: &Arc, id: RequestID, msg: Vec); - async fn recv_loop(self: Arc, mut read: R) -> Result<(), Error> + async fn recv_loop(self: Arc, mut read: R, debug_name: String) -> Result<(), Error> where R: AsyncReadExt + Unpin + Send + Sync, { let mut receiving = HashMap::new(); loop { - trace!("recv_loop: reading packet"); let mut header_id = [0u8; RequestID::BITS as usize / 8]; match read.read_exact(&mut header_id[..]).await { Ok(_) => (), @@ -200,19 +231,24 @@ pub(crate) trait RecvLoop: Sync + 'static { Err(e) => return Err(e.into()), }; let id = RequestID::from_be_bytes(header_id); - trace!("recv_loop: got header id: {:04x}", id); let mut header_size = [0u8; ChunkLength::BITS as usize / 8]; read.read_exact(&mut header_size[..]).await?; let size = ChunkLength::from_be_bytes(header_size); - trace!("recv_loop: got header size: {:04x}", size); + trace!( + "recv_loop({}): got header id = {}, size = 0x{:04x} ({} bytes)", + debug_name, + id, + size, + size & !CHUNK_HAS_CONTINUATION + ); let has_cont = (size & CHUNK_HAS_CONTINUATION) != 0; let size = size & !CHUNK_HAS_CONTINUATION; let mut next_slice = vec![0; size as usize]; read.read_exact(&mut next_slice[..]).await?; - trace!("recv_loop: read {} bytes", next_slice.len()); + trace!("recv_loop({}): read {} bytes", debug_name, next_slice.len()); let mut msg_bytes: Vec<_> = receiving.remove(&id).unwrap_or_default(); msg_bytes.extend_from_slice(&next_slice[..]); diff --git a/src/server.rs b/src/server.rs index 5465307..a835959 100644 --- a/src/server.rs +++ b/src/server.rs @@ -105,14 +105,17 @@ impl ServerConn { netapp.connected_as_server(peer_id, conn.clone()); + let debug_name = format!("SRV {}", hex::encode(&peer_id[..8])); + let debug_name_2 = debug_name.clone(); + let conn2 = conn.clone(); let recv_future = tokio::spawn(async move { select! { - r = conn2.recv_loop(read) => r, + r = conn2.recv_loop(read, debug_name_2) => r, _ = await_exit(must_exit) => Ok(()) } }); - let send_future = tokio::spawn(conn.clone().send_loop(resp_recv, write)); + let send_future = tokio::spawn(conn.clone().send_loop(resp_recv, write, debug_name)); recv_future.await.log_err("ServerConn recv_loop"); conn.resp_send.store(None);