Fixing the performance issues related to stream management

blocking_client/src/quiche_client_loop.rs

@@ -233,7 +233,10 @@ pub fn create_quiche_client_thread(
             }
 
             for stream_id in connection.writable() {
-                handle_writable(&mut connection, &mut partial_responses, stream_id);
+                if let Err(e) = handle_writable(&mut connection, &mut partial_responses, stream_id)
+                {
+                    log::error!("Error writing message on writable stream : {e:?}");
+                }
             }
 
             if connection.is_closed() {
@@ -375,7 +378,6 @@ mod tests {
                 rx_sent_queue,
                 CompressionType::Lz4Fast(8),
                 true,
-                100,
             ) {
                 log::error!("Server loop closed by error : {e}");
             }

examples/tester-client/src/main.rs

@@ -133,7 +133,12 @@ pub async fn main() {
     sleep(Duration::from_secs(1));
     println!("Subscribing");
     client
-        .subscribe(vec![Filter::BlockAll, Filter::Slot, Filter::BlockMeta])
+        .subscribe(vec![
+            Filter::BlockAll,
+            Filter::Slot,
+            Filter::BlockMeta,
+            Filter::AccountsAll,
+        ])
         .await
         .unwrap();
     println!("Subscribed");

quiche/src/quiche_sender.rs

@@ -15,6 +15,7 @@ pub fn send_message(
 ) -> std::result::Result<(), quiche::Error> {
     let written = match connection.stream_send(stream_id, message, true) {
         Ok(v) => v,
+        Err(quiche::Error::Done) => 0,
         Err(e) => {
             return Err(e);
         }
@@ -36,19 +37,25 @@ pub fn handle_writable(
     conn: &mut quiche::Connection,
     partial_responses: &mut PartialResponses,
     stream_id: u64,
-) {
+) -> std::result::Result<(), quiche::Error> {
     log::trace!("{} stream {} is writable", conn.trace_id(), stream_id);
 
     let resp = match partial_responses.get_mut(&stream_id) {
         Some(s) => s,
         None => {
-            return;
+            // stream has finished
+            let _ = conn.stream_send(stream_id, b"", true);
+            return Ok(());
         }
     };
     let body = &resp.binary;
 
     let written = match conn.stream_send(stream_id, body, true) {
         Ok(v) => v,
+        Err(quiche::Error::Done) => {
+            // done writing
+            return Ok(());
+        }
         Err(e) => {
             partial_responses.remove(&stream_id);
 
@@ -56,12 +63,12 @@ pub fn handle_writable(
                 "{} stream id :{stream_id} send failed {e:?}",
                 conn.trace_id()
             );
-            return;
+            return Err(e);
         }
     };
 
     if written == 0 {
-        return;
+        return Ok(());
     }
 
     if written == resp.binary.len() {
@@ -78,4 +85,5 @@ pub fn handle_writable(
         resp.binary = resp.binary[written..].to_vec();
         resp.written += written;
     }
+    Ok(())
 }

server/src/quiche_server_loop.rs

@@ -412,7 +412,11 @@ fn create_client_task(
             {
                 for stream_id in connection.writable() {
                     number_of_writable_streams.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-                    handle_writable(&mut connection, &mut partial_responses, stream_id);
+                    if let Err(e) =
+                        handle_writable(&mut connection, &mut partial_responses, stream_id)
+                    {
+                        log::error!("Error writing {e:?}");
+                    }
                 }
 
                 while partial_responses.len() < max_allowed_partial_responses {
@@ -433,13 +437,22 @@ fn create_client_task(
                                 true
                             } else {
                                 messages_added.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-                                send_message(
+                                match send_message(
                                     &mut connection,
                                     &mut partial_responses,
                                     stream_id,
                                     &message,
-                                )
+                                ) {
-                                .is_err()
+                                    Ok(_) => false,
+                                    Err(quiche::Error::Done) => {
+                                        // done writing / queue is full
+                                        break;
+                                    }
+                                    Err(e) => {
+                                        log::error!("error sending message : {e:?}");
+                                        true
+                                    }
+                                }
                             }
                         }
                         Err(e) => {