From fa399e01adcb68c27720de8a50f46f411f6d1bba Mon Sep 17 00:00:00 2001 From: rustdesk Date: Thu, 6 Aug 2026 14:21:40 +0800 Subject: [PATCH] fix: the KCP io throttle reset itself every cycle, so it never throttled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The send and recv arms shared one counter, and an ICMP error on a connected socket is reported once and then cleared — so the steady state is an alternation: the send succeeds and clears the counter, the next recv reports the error and finds the counter at 1, and logs. Every error still wrote a line, at the ~100/s the previous commit set out to stop, while the persistent-failure and recovery branches were unreachable. Use one LogThrottle per direction instead of a hand-rolled counter. That removes the shared state the bug lived in, drops a third throttling mechanism in favour of the one already added, and leaves the surrounding `if let Err` untouched rather than reshaping it into a match. Also fix test_udp_uat's socket-error arm, the untreated twin of the punch_udp site: it had no backoff at all, so a persistent error re-armed recv immediately and spun the loop at CPU speed, one warn line per iteration. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01ExUfAkYbq8UC9pQCiLy8TQ --- src/client.rs | 8 ++++++- src/common.rs | 2 +- src/kcp_stream.rs | 58 +++++++++++++++-------------------------------- 3 files changed, 26 insertions(+), 42 deletions(-) diff --git a/src/client.rs b/src/client.rs index f81ff5f10..1516c68f4 100644 --- a/src/client.rs +++ b/src/client.rs @@ -332,6 +332,7 @@ async fn race_transports_prefer_webrtc<'a, T: 'a>( use hbb_common::log_throttle::LogThrottle; const ICE_LOG_INTERVAL: Duration = Duration::from_secs(60); static REJECTED_ICE_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL); +static UDP_UAT_ERR_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL); static UNEXPECTED_ICE_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL); static PENDING_ICE_FULL_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL); @@ -5202,7 +5203,12 @@ async fn test_udp_uat( } } Err(e) => { - log::warn!("UDP NAT test socket error: {}", e); + // Same ICMP-driven errors as punch_udp sees. Without a pause this arm + // re-arms recv immediately and spins the loop at CPU speed. + if let Some(n) = UDP_UAT_ERR_LOG.due() { + log::warn!("UDP NAT test socket error x{n}, last: {e}"); + } + hbb_common::sleep(0.01).await; } } } diff --git a/src/common.rs b/src/common.rs index 2bf64634a..6dee71837 100644 --- a/src/common.rs +++ b/src/common.rs @@ -2653,7 +2653,7 @@ pub async fn punch_udp( tokio::select! { _ = hbb_common::sleep(retry_interval.as_secs_f32()) => { if tm.elapsed() > MAX_TIME { - bail!("UDP punch is timed out, stop sending packets after {packets_sent:?} packets, {recv_errors} recv errors absorbed"); + bail!("UDP punch is timed out, stop sending packets after {:?} packets, {} recv errors absorbed", packets_sent, recv_errors); } let elapsed = last_send_time.elapsed(); diff --git a/src/kcp_stream.rs b/src/kcp_stream.rs index 488552ac2..15175e9d5 100644 --- a/src/kcp_stream.rs +++ b/src/kcp_stream.rs @@ -19,6 +19,12 @@ pub struct KcpStream { stop_sender: Option>, } +const KCP_IO_ERR_LOG_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); +static KCP_SEND_ERR_LOG: hbb_common::log_throttle::LogThrottle = + hbb_common::log_throttle::LogThrottle::new(KCP_IO_ERR_LOG_INTERVAL); +static KCP_RECV_ERR_LOG: hbb_common::log_throttle::LogThrottle = + hbb_common::log_throttle::LogThrottle::new(KCP_IO_ERR_LOG_INTERVAL); + impl KcpStream { // Engage KCP's built-in congestion control (nc=0) unless disabled by option: pure turbo // (nc=1) keeps blasting a full 1024-segment window through loss, which on constrained @@ -36,33 +42,6 @@ impl KcpStream { } } - // One line per run of socket failures rather than per failure: these are absorbed as packet - // loss and can repeat every 10ms, and debug output is written to the log file. - const KCP_IO_FAIL_PERSIST: u64 = 500; // ~5s at the 10ms retry interval - - fn note_io_err(fail_run: &mut u64, what: &str, e: &std::io::Error) { - *fail_run += 1; - if *fail_run == 1 { - log::debug!("KCP {} error (treated as loss): {:?}", what, e); - } else if *fail_run % Self::KCP_IO_FAIL_PERSIST == 0 { - // Still failing well past a transient ICMP: say so once per run length, else a - // socket that never recovers is silent until KCP reaps it 60s later. - log::warn!( - "KCP {} failing persistently: {} consecutive errors, last: {:?}", - what, - fail_run, - e - ); - } - } - - fn note_io_ok(fail_run: &mut u64) { - if *fail_run >= Self::KCP_IO_FAIL_PERSIST { - log::info!("KCP socket recovered after {} failed operations", fail_run); - } - *fail_run = 0; - } - fn create_framed(stream: stream::KcpStream, local_addr: Option) -> Stream { Stream::Tcp(FramedStream( tokio_util::codec::Framed::new(DynTcpStream(Box::new(stream)), BytesCodec::new()), @@ -157,12 +136,11 @@ impl KcpStream { // a truly dead link is reaped by the KCP pong timeout / app-level timeouts. // The short sleep prevents a persistently failing socket from busy-spinning. // - // Log by run, not per occurrence: the 10ms sleep means a persistently failing - // socket would otherwise write ~100 lines a second into the log file. One line - // when a run starts, one per PERSIST run length so a stuck socket stays visible - // (the pong timeout takes 60s to reap it, and the session is frozen meanwhile), - // and one when it recovers carrying the total. - let mut fail_run = 0u64; + // These repeat every 10ms while the socket stays broken, so throttle the line — + // debug output is written to the log file. One throttle per direction: an ICMP + // error on a connected socket is reported once and cleared, so the steady state is + // an alternation (send succeeds, the next recv reports the error), and a shared + // counter would be reset by the succeeding direction on every cycle. loop { tokio::select! { _ = &mut stop_receiver => { @@ -170,18 +148,16 @@ impl KcpStream { break; } Some(data) = output.recv() => { - match udp.send(&data.inner()).await { - Ok(_) => Self::note_io_ok(&mut fail_run), - Err(e) => { - Self::note_io_err(&mut fail_run, "send", &e); - tokio::time::sleep(std::time::Duration::from_millis(10)).await; + if let Err(e) = udp.send(&data.inner()).await { + if let Some(n) = KCP_SEND_ERR_LOG.due() { + log::debug!("KCP send error x{n} (treated as loss), last: {e}"); } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; } } result = udp.recv_from(&mut buf) => { match result { Ok((size, _)) => { - Self::note_io_ok(&mut fail_run); if size < std::mem::size_of::() { continue; } @@ -190,7 +166,9 @@ impl KcpStream { .await.ok(); } Err(e) => { - Self::note_io_err(&mut fail_run, "recv", &e); + if let Some(n) = KCP_RECV_ERR_LOG.due() { + log::debug!("KCP recv error x{n} (treated as loss), last: {e}"); + } tokio::time::sleep(std::time::Duration::from_millis(10)).await; } }