From bb8a446a5b950dcb0fb14e9d8d296354d550f2ac Mon Sep 17 00:00:00 2001 From: Arthur Kashfullin Date: Fri, 5 Jun 2026 11:01:27 +0300 Subject: [PATCH 1/2] Handle closed report channel without panic --- src/client.rs | 99 +++++++++++++++++++++++++++++++++++------------- src/client_h3.rs | 20 +++++++--- 2 files changed, 87 insertions(+), 32 deletions(-) diff --git a/src/client.rs b/src/client.rs index 13fc6450..561ff8c0 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1026,6 +1026,10 @@ pub(crate) fn is_cancel_error(res: &Result) -> bool matches!(res, Err(ClientError::Deadline)) || is_too_many_open_files(res) } +pub(crate) fn try_send_report(report_tx: &kanal::Sender, report: T) -> bool { + report_tx.send(report).is_ok() +} + /// Check error was "Too many open file" fn is_too_many_open_files(res: &Result) -> bool { res.as_ref() @@ -1075,7 +1079,9 @@ async fn work_http2_once( if let Some(start_latency_correction) = start_latency_correction { set_start_latency_correction(&mut res, start_latency_correction); } - report_tx.send(res).unwrap(); + if !try_send_report(report_tx, res) { + return (true, is_reconnect); + } (is_cancel, is_reconnect) } @@ -1220,7 +1226,9 @@ pub async fn work( } Err(err) => { if counter.fetch_add(1, Ordering::Relaxed) < n_tasks { - report_tx.send(Err(err)).unwrap(); + if !try_send_report(&report_tx, Err(err)) { + return; + } } else { return; } @@ -1245,7 +1253,9 @@ pub async fn work( while counter.fetch_add(1, Ordering::Relaxed) < n_tasks { let res = client.work_http1(&mut client_state).await; let is_cancel = is_cancel_error(&res); - report_tx.send(res).unwrap(); + if !try_send_report(&report_tx, res) { + break; + } if is_cancel { break; } @@ -1386,7 +1396,9 @@ pub async fn work_with_qps( Err(err) => { // Consume a task if let Ok(()) = rx.recv().await { - report_tx.send(Err(err)).unwrap(); + if !try_send_report(&report_tx, Err(err)) { + return; + } } else { return; } @@ -1413,7 +1425,9 @@ pub async fn work_with_qps( while let Ok(()) = rx.recv().await { let res = client.work_http1(&mut client_state).await; let is_cancel = is_cancel_error(&res); - report_tx.send(res).unwrap(); + if !try_send_report(&report_tx, res) { + break; + } if is_cancel { break; } @@ -1560,7 +1574,9 @@ pub async fn work_with_qps_latency_correction( Err(err) => { // Consume a task if rx.recv().await.is_ok() { - report_tx.send(Err(err)).unwrap(); + if !try_send_report(&report_tx, Err(err)) { + return; + } } else { return; } @@ -1588,7 +1604,9 @@ pub async fn work_with_qps_latency_correction( let mut res = client.work_http1(&mut client_state).await; set_start_latency_correction(&mut res, start); let is_cancel = is_cancel_error(&res); - report_tx.send(res).unwrap(); + if !try_send_report(&report_tx, res) { + break; + } if is_cancel { break; } @@ -1697,7 +1715,9 @@ pub async fn work_until( } } _ = s.acquire() => { - report_tx.send(Err(ClientError::Deadline)).unwrap(); + if !try_send_report(&report_tx, Err(ClientError::Deadline)) { + return; + } connection_gone = true; } } @@ -1708,7 +1728,9 @@ pub async fn work_until( } Err(err) => { - report_tx.send(Err(err)).unwrap(); + if !try_send_report(&report_tx, Err(err)) { + break; + } if s.is_closed() { break; } @@ -1739,7 +1761,9 @@ pub async fn work_until( loop { let res = client.work_http1(&mut client_state).await; let is_cancel = is_cancel_error(&res); - report_tx.send(res).unwrap(); + if !try_send_report(&report_tx, res) { + break; + } if is_cancel || is_end.load(Relaxed) { break; } @@ -1760,8 +1784,9 @@ pub async fn work_until( f.abort(); if let Err(e) = f.await && e.is_cancelled() + && !try_send_report(&report_tx, Err(ClientError::Deadline)) { - report_tx.send(Err(ClientError::Deadline)).unwrap(); + break; } } } @@ -1901,7 +1926,9 @@ pub async fn work_until_with_qps( } } _ = s.acquire() => { - report_tx.send(Err(ClientError::Deadline)).unwrap(); + if !try_send_report(&report_tx, Err(ClientError::Deadline)) { + return; + } connection_gone = true; } } @@ -1913,7 +1940,9 @@ pub async fn work_until_with_qps( Err(err) => { // Consume a task if rx.recv().await.is_ok() { - report_tx.send(Err(err)).unwrap(); + if !try_send_report(&report_tx, Err(err)) { + return; + } } else { return; } @@ -1949,7 +1978,9 @@ pub async fn work_until_with_qps( while let Ok(()) = rx.recv().await { let res = client.work_http1(&mut client_state).await; let is_cancel = is_cancel_error(&res); - report_tx.send(res).unwrap(); + if !try_send_report(&report_tx, res) { + break; + } if is_cancel || is_end.load(Relaxed) { break; } @@ -1970,8 +2001,9 @@ pub async fn work_until_with_qps( f.abort(); if let Err(e) = f.await && e.is_cancelled() + && !try_send_report(&report_tx, Err(ClientError::Deadline)) { - report_tx.send(Err(ClientError::Deadline)).unwrap(); + break; } } } @@ -2109,7 +2141,9 @@ pub async fn work_until_with_qps_latency_correction( } } _ = s.acquire() => { - report_tx.send(Err(ClientError::Deadline)).unwrap(); + if !try_send_report(&report_tx, Err(ClientError::Deadline)) { + return; + } connection_gone = true; } } @@ -2121,7 +2155,9 @@ pub async fn work_until_with_qps_latency_correction( Err(err) => { if rx.recv().await.is_ok() { - report_tx.send(Err(err)).unwrap(); + if !try_send_report(&report_tx, Err(err)) { + return; + } } else { return; } @@ -2158,7 +2194,9 @@ pub async fn work_until_with_qps_latency_correction( let mut res = client.work_http1(&mut client_state).await; set_start_latency_correction(&mut res, start); let is_cancel = is_cancel_error(&res); - report_tx.send(res).unwrap(); + if !try_send_report(&report_tx, res) { + break; + } if is_cancel || is_end.load(Relaxed) { break; } @@ -2179,8 +2217,9 @@ pub async fn work_until_with_qps_latency_correction( f.abort(); if let Err(e) = f.await && e.is_cancelled() + && !try_send_report(&report_tx, Err(ClientError::Deadline)) { - report_tx.send(Err(ClientError::Deadline)).unwrap(); + break; } } } @@ -2200,7 +2239,7 @@ pub mod fast { use crate::{ client::{ ClientError, ClientStateHttp1, ClientStateHttp2, HttpWorkType, is_cancel_error, - is_hyper_error, set_connection_time, setup_http2, + is_hyper_error, set_connection_time, setup_http2, try_send_report, }, pcg64si::Pcg64Si, result_data::ResultData, @@ -2330,7 +2369,13 @@ pub mod fast { } }; - report_tx.send(result_data).unwrap(); + let sent = try_send_report( + &report_tx, + result_data, + ); + if !sent { + return true; + } is_cancel }) }) @@ -2366,7 +2411,7 @@ pub mod fast { } } if has_err { - report_tx.send(result_data_err).unwrap(); + let _sent = try_send_report(&report_tx, result_data_err); } })); } @@ -2412,7 +2457,7 @@ pub mod fast { } } => {} } - report_tx.send(result_data).unwrap(); + let _sent = try_send_report(&report_tx, result_data); })); } rt.block_on(local); @@ -2547,7 +2592,9 @@ pub mod fast { } }; - report_tx.send(result_data).unwrap(); + if !try_send_report(&report_tx, result_data) { + return true; + } is_cancel }) }) @@ -2582,7 +2629,7 @@ pub mod fast { } } if has_err { - report_tx.send(result_data_err).unwrap(); + let _sent = try_send_report(&report_tx, result_data_err); } })); } @@ -2633,7 +2680,7 @@ pub mod fast { result_data.push(Err(ClientError::Deadline)); } } - report_tx.send(result_data).unwrap(); + let _sent = try_send_report(&report_tx, result_data); })); } rt.block_on(local); diff --git a/src/client_h3.rs b/src/client_h3.rs index a60f8ef4..759ee5a2 100644 --- a/src/client_h3.rs +++ b/src/client_h3.rs @@ -41,7 +41,7 @@ pub enum Http3Error { use crate::client::QueryLimit; use crate::client::{ Client, ClientError, ConnectionTime, RequestResult, Stream, is_cancel_error, - set_connection_time, set_start_latency_correction, + set_connection_time, set_start_latency_correction, try_send_report, }; use crate::pcg64si::Pcg64Si; use crate::result_data::ResultData; @@ -341,7 +341,9 @@ async fn create_and_load_up_single_connection_http3( } } _ = s.acquire() => { - report_tx.send(Err(ClientError::Deadline)).unwrap(); + if !try_send_report(&report_tx, Err(ClientError::Deadline)) { + return; + } connection_gone = true; } } @@ -358,7 +360,9 @@ async fn create_and_load_up_single_connection_http3( break; // Consume a task } else if rx.recv().await.is_ok() { - report_tx.send(Err(err)).unwrap(); + if !try_send_report(&report_tx, Err(err)) { + return; + } } else { return; } @@ -426,7 +430,9 @@ pub(crate) async fn work_http3_once( if let Some(start_latency_correction) = start_latency_correction { set_start_latency_correction(&mut res, start_latency_correction); } - report_tx.send(res).unwrap(); + if !try_send_report(report_tx, res) { + return (true, is_reconnect); + } (is_cancel, is_reconnect) } @@ -519,7 +525,9 @@ pub(crate) fn http3_connection_fast_work_until( } }; - report_tx.send(result_data).unwrap(); + if !try_send_report(&report_tx, result_data) { + return true; + } is_cancel }) }) @@ -559,7 +567,7 @@ pub(crate) fn http3_connection_fast_work_until( } } if has_err { - report_tx.send(result_data_err).unwrap(); + let _sent = try_send_report(&report_tx, result_data_err); } })); } From be073f0ae5c9401ae58179e02108280f0f5e14ed Mon Sep 17 00:00:00 2001 From: Arthur Kashfullin Date: Sat, 20 Jun 2026 23:09:06 +0300 Subject: [PATCH 2/2] refactor: inline report channel sends --- src/client.rs | 61 +++++++++++++++++++++--------------------------- src/client_h3.rs | 12 +++++----- 2 files changed, 33 insertions(+), 40 deletions(-) diff --git a/src/client.rs b/src/client.rs index 561ff8c0..b02f7189 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1026,10 +1026,6 @@ pub(crate) fn is_cancel_error(res: &Result) -> bool matches!(res, Err(ClientError::Deadline)) || is_too_many_open_files(res) } -pub(crate) fn try_send_report(report_tx: &kanal::Sender, report: T) -> bool { - report_tx.send(report).is_ok() -} - /// Check error was "Too many open file" fn is_too_many_open_files(res: &Result) -> bool { res.as_ref() @@ -1079,7 +1075,7 @@ async fn work_http2_once( if let Some(start_latency_correction) = start_latency_correction { set_start_latency_correction(&mut res, start_latency_correction); } - if !try_send_report(report_tx, res) { + if report_tx.send(res).is_err() { return (true, is_reconnect); } (is_cancel, is_reconnect) @@ -1226,7 +1222,7 @@ pub async fn work( } Err(err) => { if counter.fetch_add(1, Ordering::Relaxed) < n_tasks { - if !try_send_report(&report_tx, Err(err)) { + if report_tx.send(Err(err)).is_err() { return; } } else { @@ -1253,7 +1249,7 @@ pub async fn work( while counter.fetch_add(1, Ordering::Relaxed) < n_tasks { let res = client.work_http1(&mut client_state).await; let is_cancel = is_cancel_error(&res); - if !try_send_report(&report_tx, res) { + if report_tx.send(res).is_err() { break; } if is_cancel { @@ -1396,7 +1392,7 @@ pub async fn work_with_qps( Err(err) => { // Consume a task if let Ok(()) = rx.recv().await { - if !try_send_report(&report_tx, Err(err)) { + if report_tx.send(Err(err)).is_err() { return; } } else { @@ -1425,7 +1421,7 @@ pub async fn work_with_qps( while let Ok(()) = rx.recv().await { let res = client.work_http1(&mut client_state).await; let is_cancel = is_cancel_error(&res); - if !try_send_report(&report_tx, res) { + if report_tx.send(res).is_err() { break; } if is_cancel { @@ -1574,7 +1570,7 @@ pub async fn work_with_qps_latency_correction( Err(err) => { // Consume a task if rx.recv().await.is_ok() { - if !try_send_report(&report_tx, Err(err)) { + if report_tx.send(Err(err)).is_err() { return; } } else { @@ -1604,7 +1600,7 @@ pub async fn work_with_qps_latency_correction( let mut res = client.work_http1(&mut client_state).await; set_start_latency_correction(&mut res, start); let is_cancel = is_cancel_error(&res); - if !try_send_report(&report_tx, res) { + if report_tx.send(res).is_err() { break; } if is_cancel { @@ -1715,7 +1711,7 @@ pub async fn work_until( } } _ = s.acquire() => { - if !try_send_report(&report_tx, Err(ClientError::Deadline)) { + if report_tx.send(Err(ClientError::Deadline)).is_err() { return; } connection_gone = true; @@ -1728,7 +1724,7 @@ pub async fn work_until( } Err(err) => { - if !try_send_report(&report_tx, Err(err)) { + if report_tx.send(Err(err)).is_err() { break; } if s.is_closed() { @@ -1761,7 +1757,7 @@ pub async fn work_until( loop { let res = client.work_http1(&mut client_state).await; let is_cancel = is_cancel_error(&res); - if !try_send_report(&report_tx, res) { + if report_tx.send(res).is_err() { break; } if is_cancel || is_end.load(Relaxed) { @@ -1784,7 +1780,7 @@ pub async fn work_until( f.abort(); if let Err(e) = f.await && e.is_cancelled() - && !try_send_report(&report_tx, Err(ClientError::Deadline)) + && report_tx.send(Err(ClientError::Deadline)).is_err() { break; } @@ -1926,7 +1922,7 @@ pub async fn work_until_with_qps( } } _ = s.acquire() => { - if !try_send_report(&report_tx, Err(ClientError::Deadline)) { + if report_tx.send(Err(ClientError::Deadline)).is_err() { return; } connection_gone = true; @@ -1940,7 +1936,7 @@ pub async fn work_until_with_qps( Err(err) => { // Consume a task if rx.recv().await.is_ok() { - if !try_send_report(&report_tx, Err(err)) { + if report_tx.send(Err(err)).is_err() { return; } } else { @@ -1978,7 +1974,7 @@ pub async fn work_until_with_qps( while let Ok(()) = rx.recv().await { let res = client.work_http1(&mut client_state).await; let is_cancel = is_cancel_error(&res); - if !try_send_report(&report_tx, res) { + if report_tx.send(res).is_err() { break; } if is_cancel || is_end.load(Relaxed) { @@ -2001,7 +1997,7 @@ pub async fn work_until_with_qps( f.abort(); if let Err(e) = f.await && e.is_cancelled() - && !try_send_report(&report_tx, Err(ClientError::Deadline)) + && report_tx.send(Err(ClientError::Deadline)).is_err() { break; } @@ -2141,7 +2137,7 @@ pub async fn work_until_with_qps_latency_correction( } } _ = s.acquire() => { - if !try_send_report(&report_tx, Err(ClientError::Deadline)) { + if report_tx.send(Err(ClientError::Deadline)).is_err() { return; } connection_gone = true; @@ -2155,7 +2151,7 @@ pub async fn work_until_with_qps_latency_correction( Err(err) => { if rx.recv().await.is_ok() { - if !try_send_report(&report_tx, Err(err)) { + if report_tx.send(Err(err)).is_err() { return; } } else { @@ -2194,7 +2190,7 @@ pub async fn work_until_with_qps_latency_correction( let mut res = client.work_http1(&mut client_state).await; set_start_latency_correction(&mut res, start); let is_cancel = is_cancel_error(&res); - if !try_send_report(&report_tx, res) { + if report_tx.send(res).is_err() { break; } if is_cancel || is_end.load(Relaxed) { @@ -2217,7 +2213,7 @@ pub async fn work_until_with_qps_latency_correction( f.abort(); if let Err(e) = f.await && e.is_cancelled() - && !try_send_report(&report_tx, Err(ClientError::Deadline)) + && report_tx.send(Err(ClientError::Deadline)).is_err() { break; } @@ -2239,7 +2235,7 @@ pub mod fast { use crate::{ client::{ ClientError, ClientStateHttp1, ClientStateHttp2, HttpWorkType, is_cancel_error, - is_hyper_error, set_connection_time, setup_http2, try_send_report, + is_hyper_error, set_connection_time, setup_http2, }, pcg64si::Pcg64Si, result_data::ResultData, @@ -2369,11 +2365,8 @@ pub mod fast { } }; - let sent = try_send_report( - &report_tx, - result_data, - ); - if !sent { + if report_tx.send(result_data).is_err() + { return true; } is_cancel @@ -2411,7 +2404,7 @@ pub mod fast { } } if has_err { - let _sent = try_send_report(&report_tx, result_data_err); + let _sent = report_tx.send(result_data_err); } })); } @@ -2457,7 +2450,7 @@ pub mod fast { } } => {} } - let _sent = try_send_report(&report_tx, result_data); + let _sent = report_tx.send(result_data); })); } rt.block_on(local); @@ -2592,7 +2585,7 @@ pub mod fast { } }; - if !try_send_report(&report_tx, result_data) { + if report_tx.send(result_data).is_err() { return true; } is_cancel @@ -2629,7 +2622,7 @@ pub mod fast { } } if has_err { - let _sent = try_send_report(&report_tx, result_data_err); + let _sent = report_tx.send(result_data_err); } })); } @@ -2680,7 +2673,7 @@ pub mod fast { result_data.push(Err(ClientError::Deadline)); } } - let _sent = try_send_report(&report_tx, result_data); + let _sent = report_tx.send(result_data); })); } rt.block_on(local); diff --git a/src/client_h3.rs b/src/client_h3.rs index 759ee5a2..efb0640b 100644 --- a/src/client_h3.rs +++ b/src/client_h3.rs @@ -41,7 +41,7 @@ pub enum Http3Error { use crate::client::QueryLimit; use crate::client::{ Client, ClientError, ConnectionTime, RequestResult, Stream, is_cancel_error, - set_connection_time, set_start_latency_correction, try_send_report, + set_connection_time, set_start_latency_correction, }; use crate::pcg64si::Pcg64Si; use crate::result_data::ResultData; @@ -341,7 +341,7 @@ async fn create_and_load_up_single_connection_http3( } } _ = s.acquire() => { - if !try_send_report(&report_tx, Err(ClientError::Deadline)) { + if report_tx.send(Err(ClientError::Deadline)).is_err() { return; } connection_gone = true; @@ -360,7 +360,7 @@ async fn create_and_load_up_single_connection_http3( break; // Consume a task } else if rx.recv().await.is_ok() { - if !try_send_report(&report_tx, Err(err)) { + if report_tx.send(Err(err)).is_err() { return; } } else { @@ -430,7 +430,7 @@ pub(crate) async fn work_http3_once( if let Some(start_latency_correction) = start_latency_correction { set_start_latency_correction(&mut res, start_latency_correction); } - if !try_send_report(report_tx, res) { + if report_tx.send(res).is_err() { return (true, is_reconnect); } (is_cancel, is_reconnect) @@ -525,7 +525,7 @@ pub(crate) fn http3_connection_fast_work_until( } }; - if !try_send_report(&report_tx, result_data) { + if report_tx.send(result_data).is_err() { return true; } is_cancel @@ -567,7 +567,7 @@ pub(crate) fn http3_connection_fast_work_until( } } if has_err { - let _sent = try_send_report(&report_tx, result_data_err); + let _sent = report_tx.send(result_data_err); } })); }