aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/client.rs9
-rw-r--r--src/recv.rs19
-rw-r--r--src/send.rs46
-rw-r--r--src/server.rs7
-rw-r--r--src/stream.rs8
5 files changed, 63 insertions, 26 deletions
diff --git a/src/client.rs b/src/client.rs
index d51236b..2fccdb8 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -179,10 +179,9 @@ impl ClientConn {
})));
}
- trace!(
- "request: query_send {} (serialized message: {} bytes)",
- id,
- req_msg_len
+ debug!(
+ "request: query_send {}, path {}, prio {} (serialized message: {} bytes)",
+ id, path, prio, req_msg_len
);
#[cfg(feature = "telemetry")]
@@ -201,7 +200,7 @@ impl ClientConn {
}
let resp_enc = RespEnc::decode(stream).await?;
- trace!("request response {}", id);
+ debug!("client: got response to request {} (path {})", id, path);
Resp::from_enc(resp_enc)
}
}
diff --git a/src/recv.rs b/src/recv.rs
index e748f18..cba42cb 100644
--- a/src/recv.rs
+++ b/src/recv.rs
@@ -59,7 +59,6 @@ pub(crate) trait RecvLoop: Sync + 'static {
{
let mut streams: HashMap<RequestID, Sender> = 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(_) => (),
@@ -67,22 +66,31 @@ 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);
let has_cont = (size & CHUNK_HAS_CONTINUATION) != 0;
let is_error = (size & ERROR_MARKER) != 0;
let packet = if is_error {
+ trace!(
+ "recv_loop: got id {}, header_size {:04x}, error {}",
+ id,
+ size,
+ size & !ERROR_MARKER
+ );
Err((size & !ERROR_MARKER) as u8)
} else {
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: got id {}, header_size {:04x}, {} bytes",
+ id,
+ size,
+ next_slice.len()
+ );
Ok(Bytes::from(next_slice))
};
@@ -90,6 +98,7 @@ pub(crate) trait RecvLoop: Sync + 'static {
send
} else {
let (send, recv) = mpsc::channel(4);
+ trace!("recv_loop: id {} is new channel", id);
self.recv_handler(
id,
Box::pin(tokio_stream::wrappers::ReceiverStream::new(recv)),
@@ -102,8 +111,10 @@ pub(crate) trait RecvLoop: Sync + 'static {
let _ = sender.send(packet).await;
if has_cont {
+ assert!(!is_error);
streams.insert(id, sender);
} else {
+ trace!("recv_loop: close channel id {}", id);
sender.end();
}
}
diff --git a/src/send.rs b/src/send.rs
index 46c4383..256fe4c 100644
--- a/src/send.rs
+++ b/src/send.rs
@@ -74,36 +74,54 @@ impl<'a> futures::Future for SendQueuePollNextReady<'a> {
type Output = (RequestID, DataFrame);
fn poll(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> {
- for i in 0..self.queue.items.len() {
- let (_prio, items_at_prio) = &mut self.queue.items[i];
-
+ for (i, (_prio, items_at_prio)) in self.queue.items.iter_mut().enumerate() {
let mut ready_item = None;
for (j, item) in items_at_prio.iter_mut().enumerate() {
let mut item_reader = item.data.read_exact_or_eos(MAX_CHUNK_LENGTH as usize);
match Pin::new(&mut item_reader).poll(ctx) {
Poll::Pending => (),
Poll::Ready(ready_v) => {
- ready_item = Some((j, ready_v, item.data.eos()));
+ ready_item = Some((j, ready_v));
break;
}
}
}
- if let Some((j, bytes_or_err, eos)) = ready_item {
+ if let Some((j, bytes_or_err)) = ready_item {
+ let item = items_at_prio.remove(j).unwrap();
+ let id = item.id;
+ let eos = item.data.eos();
+
let data_frame = match bytes_or_err {
- Ok(bytes) => DataFrame::Data(bytes, !eos),
+ Ok(bytes) => {
+ trace!(
+ "send queue poll next ready: id {} eos {:?} bytes {}",
+ id,
+ eos,
+ bytes.len()
+ );
+ DataFrame::Data(bytes, !eos)
+ }
Err(e) => DataFrame::Error(match e {
- ReadExactError::Stream(code) => code,
+ ReadExactError::Stream(code) => {
+ trace!(
+ "send queue poll next ready: id {} eos {:?} ERROR {}",
+ id,
+ eos,
+ code
+ );
+ code
+ }
_ => unreachable!(),
}),
};
- let item = items_at_prio.remove(j).unwrap();
- let id = item.id;
- if !eos {
+
+ if !eos && !matches!(data_frame, DataFrame::Error(_)) {
items_at_prio.push_back(item);
} else if items_at_prio.is_empty() {
self.queue.items.remove(i);
}
+
return Poll::Ready((id, data_frame));
}
}
@@ -173,6 +191,7 @@ pub(crate) trait SendLoop: Sync {
match futures::future::select(recv_fut, send_fut).await {
Either::Left((sth, _send_fut)) => {
if let Some((id, prio, data)) = sth {
+ trace!("send_loop: add stream {} to send", id);
sending.push(SendQueueItem {
id,
prio,
@@ -183,7 +202,12 @@ pub(crate) trait SendLoop: Sync {
};
}
Either::Right(((id, data), _recv_fut)) => {
- trace!("send_loop: sending bytes for {}", id);
+ trace!(
+ "send_loop: id {}, send {} bytes, header_size {}",
+ id,
+ data.data().len(),
+ hex::encode(data.header())
+ );
let header_id = RequestID::to_be_bytes(id);
write.write_all(&header_id[..]).await?;
diff --git a/src/server.rs b/src/server.rs
index 4b232af..57062d8 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -3,7 +3,7 @@ use std::sync::Arc;
use arc_swap::ArcSwapOption;
use async_trait::async_trait;
-use log::{debug, trace};
+use log::*;
use futures::io::{AsyncReadExt, AsyncWriteExt};
use kuska_handshake::async_std::{handshake_server, BoxStream};
@@ -175,7 +175,8 @@ impl RecvLoop for ServerConn {
let self2 = self.clone();
tokio::spawn(async move {
- trace!("ServerConn recv_handler {}", id);
+ debug!("server: recv_handler got {}", id);
+
let (prio, resp_enc) = match ReqEnc::decode(stream).await {
Ok(req_enc) => {
let prio = req_enc.prio;
@@ -192,7 +193,7 @@ impl RecvLoop for ServerConn {
Err(e) => (PRIO_NORMAL, RespEnc::from_err(e)),
};
- trace!("ServerConn sending response to {}: ", id);
+ debug!("server: sending response to {}", id);
resp_send
.send((id, prio, resp_enc.encode()))
diff --git a/src/stream.rs b/src/stream.rs
index beb6b9c..5ba2ed4 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -93,7 +93,7 @@ impl ByteStreamReader {
}
pub fn eos(&self) -> bool {
- self.buf.is_empty() && self.eos
+ self.buf_len == 0 && self.eos
}
fn try_get(&mut self, read_len: usize) -> Option<Bytes> {
@@ -164,8 +164,10 @@ impl<'a> Future for ByteStreamReadExact<'a> {
match futures::ready!(this.reader.stream.as_mut().poll_next(cx)) {
Some(Ok(slice)) => {
- this.reader.buf_len += slice.len();
- this.reader.buf.push_back(slice);
+ if !slice.is_empty() {
+ this.reader.buf_len += slice.len();
+ this.reader.buf.push_back(slice);
+ }
}
Some(Err(e)) => {
this.reader.err = Some(e);
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-21 22:22:26 +0000