From 700f783956697ef9d5aff4d904167f50367409e9 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Wed, 31 Aug 2022 15:08:51 +0200 Subject: Add dump of sending queue --- src/proto.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) (limited to 'src/proto.rs') diff --git a/src/proto.rs b/src/proto.rs index e843bff..56afede 100644 --- a/src/proto.rs +++ b/src/proto.rs @@ -1,5 +1,6 @@ use std::collections::{HashMap, VecDeque}; use std::sync::Arc; +use std::fmt::Write; use log::trace; @@ -94,6 +95,15 @@ 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 @@ -117,6 +127,7 @@ 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 = {}", sending.dump()); if let Ok((id, prio, data)) = msg_recv.try_recv() { trace!("send_loop: got {}, {} bytes", id, data.len()); sending.push(SendQueueItem { -- cgit v1.2.3 From 01db3c43193d9a24ef4cd1a1b1993f4d44a9aa9b Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Wed, 31 Aug 2022 15:58:05 +0200 Subject: add debug_name in proto to differenciate messages --- src/proto.rs | 24 +++++++++++++++--------- 1 file changed, 15 insertions(+), 9 deletions(-) (limited to 'src/proto.rs') diff --git a/src/proto.rs b/src/proto.rs index 56afede..41a2e47 100644 --- a/src/proto.rs +++ b/src/proto.rs @@ -120,6 +120,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, @@ -127,9 +128,9 @@ 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 = {}", sending.dump()); + 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({}): got {}, {} bytes", debug_name, id, data.len()); sending.push(SendQueueItem { id, prio, @@ -138,7 +139,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 @@ -168,7 +170,7 @@ 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({}): got {}, {} bytes", debug_name, id, data.len()); sending.push(SendQueueItem { id, prio, @@ -197,13 +199,17 @@ 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"); + trace!("recv_loop({}): reading packet", debug_name); let mut header_id = [0u8; RequestID::BITS as usize / 8]; match read.read_exact(&mut header_id[..]).await { Ok(_) => (), @@ -211,19 +217,19 @@ 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); + trace!("recv_loop({}): got header id: {:04x}", debug_name, 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 size: {:04x}", debug_name, size); 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[..]); -- cgit v1.2.3 From 984ba65e658a920f2d8e67ce808e9acd257c4d86 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Wed, 31 Aug 2022 16:10:14 +0200 Subject: Better messages in proto.rs --- src/proto.rs | 43 +++++++++++++++++++++++++++++++------------ 1 file changed, 31 insertions(+), 12 deletions(-) (limited to 'src/proto.rs') diff --git a/src/proto.rs b/src/proto.rs index 41a2e47..8f7e70f 100644 --- a/src/proto.rs +++ b/src/proto.rs @@ -1,6 +1,6 @@ use std::collections::{HashMap, VecDeque}; -use std::sync::Arc; use std::fmt::Write; +use std::sync::Arc; use log::trace; @@ -99,7 +99,14 @@ impl SendQueue { 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(); + write!( + &mut ret, + " [{} {} ({})]", + prio, + item.data.len() - item.cursor, + item.id + ) + .unwrap(); } } ret @@ -130,7 +137,13 @@ pub(crate) trait SendLoop: Sync { 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", debug_name, id, data.len()); + trace!( + "send_loop({}): new message to send, id = {}, prio = {}, {} bytes", + debug_name, + id, + prio, + data.len() + ); sending.push(SendQueueItem { id, prio, @@ -170,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", debug_name, id, data.len()); + trace!( + "send_loop({}): new message to send, id = {}, prio = {}, {} bytes", + debug_name, + id, + prio, + data.len() + ); sending.push(SendQueueItem { id, prio, @@ -199,17 +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, - debug_name: String, - ) -> 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", debug_name); let mut header_id = [0u8; RequestID::BITS as usize / 8]; match read.read_exact(&mut header_id[..]).await { Ok(_) => (), @@ -217,12 +231,17 @@ 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}", debug_name, 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}", debug_name, 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; -- cgit v1.2.3