mirror of
https://github.com/rustdesk/rustdesk.git
synced 2026-09-10 14:31:02 +03:00
fix: KCP/UDP resilience to ICMP resets; optional KCP congestion control
- treat ICMP-driven UDP socket errors (WSAECONNRESET 10054 on Windows, ECONNREFUSED on Linux) as packet loss in punch_udp and the KCP pump instead of tearing the session down; KCP retransmits through them and a truly dead link is still reaped by the pong/app-level timeouts - resolve STUN hostnames via tokio::net::lookup_host so DNS never blocks a runtime worker; fix the inverted non-IPv4 error message - add enable-kcp-congestion-control option (default on): switch the turbo profile to nc=0 so brief loss on constrained links no longer spirals into stalls; sender-side only, no wire negotiation - pin kcp-sys to the rustdesk-patches branch: upstream main lost the RustDesk patches on the EasyTier sync, and this branch also wires set_kcp_config_factory into connection setup, making the option effective Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
8
Cargo.lock
generated
8
Cargo.lock
generated
@@ -755,9 +755,9 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "bindgen"
|
name = "bindgen"
|
||||||
version = "0.71.1"
|
version = "0.72.1"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "5f58bf3d7db68cfbac37cfc485a8d711e87e064c3d0fe0435b92f7a407f9d6b3"
|
checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"bitflags 2.9.1",
|
"bitflags 2.9.1",
|
||||||
"cexpr",
|
"cexpr",
|
||||||
@@ -4276,11 +4276,11 @@ dependencies = [
|
|||||||
[[package]]
|
[[package]]
|
||||||
name = "kcp-sys"
|
name = "kcp-sys"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
source = "git+https://github.com/rustdesk-org/kcp-sys#32a6c09fc6223f54aea83981a6aa8995931d29be"
|
source = "git+https://github.com/rustdesk-org/kcp-sys?branch=rustdesk-patches#acd13bad5248c7ea0f4c0595a627c308af210fb1"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"auto_impl",
|
"auto_impl",
|
||||||
"bindgen 0.71.1",
|
"bindgen 0.72.1",
|
||||||
"bitflags 2.9.1",
|
"bitflags 2.9.1",
|
||||||
"bytes",
|
"bytes",
|
||||||
"cc",
|
"cc",
|
||||||
|
|||||||
@@ -83,7 +83,7 @@ fon = "0.6"
|
|||||||
shutdown_hooks = "0.1"
|
shutdown_hooks = "0.1"
|
||||||
totp-rs = { version = "5.4", default-features = false, features = ["gen_secret", "otpauth"] }
|
totp-rs = { version = "5.4", default-features = false, features = ["gen_secret", "otpauth"] }
|
||||||
stunclient = "0.4"
|
stunclient = "0.4"
|
||||||
kcp-sys= { git = "https://github.com/rustdesk-org/kcp-sys"}
|
kcp-sys= { git = "https://github.com/rustdesk-org/kcp-sys", branch = "rustdesk-patches" }
|
||||||
reqwest = { version = "0.12", features = ["blocking", "socks", "json", "native-tls", "rustls-tls", "rustls-tls-native-roots", "gzip", "zstd"], default-features=false }
|
reqwest = { version = "0.12", features = ["blocking", "socks", "json", "native-tls", "rustls-tls", "rustls-tls-native-roots", "gzip", "zstd"], default-features=false }
|
||||||
|
|
||||||
[target.'cfg(not(target_os = "linux"))'.dependencies]
|
[target.'cfg(not(target_os = "linux"))'.dependencies]
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
use std::{
|
use std::{
|
||||||
collections::HashMap,
|
collections::HashMap,
|
||||||
future::Future,
|
future::Future,
|
||||||
net::{SocketAddr, ToSocketAddrs},
|
net::SocketAddr,
|
||||||
sync::{Arc, Mutex, RwLock},
|
sync::{Arc, Mutex, RwLock},
|
||||||
task::Poll,
|
task::Poll,
|
||||||
};
|
};
|
||||||
@@ -2442,16 +2442,27 @@ pub fn is_udp_disabled() -> bool {
|
|||||||
Config::get_option(keys::OPTION_DISABLE_UDP) == "Y"
|
Config::get_option(keys::OPTION_DISABLE_UDP) == "Y"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub const OPTION_ENABLE_KCP_CC: &str = "enable-kcp-congestion-control";
|
||||||
|
|
||||||
|
// Default ON ("enable-" option2bool semantics); set "N" to fall back to the pure
|
||||||
|
// turbo profile (nc=1, no congestion window).
|
||||||
|
#[inline]
|
||||||
|
pub fn get_kcp_cc_enabled() -> bool {
|
||||||
|
config::option2bool(
|
||||||
|
OPTION_ENABLE_KCP_CC,
|
||||||
|
&Config::get_option(OPTION_ENABLE_KCP_CC),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
// this crate https://github.com/yoshd/stun-client supports nat type
|
// this crate https://github.com/yoshd/stun-client supports nat type
|
||||||
async fn stun_ipv6_test(stun_server: &str) -> ResultType<(SocketAddr, String)> {
|
async fn stun_ipv6_test(stun_server: &str) -> ResultType<(SocketAddr, String)> {
|
||||||
use std::net::ToSocketAddrs;
|
|
||||||
use stunclient::StunClient;
|
use stunclient::StunClient;
|
||||||
let local_addr = SocketAddr::from(([0u16; 8], 0)); // [::]:0
|
let local_addr = SocketAddr::from(([0u16; 8], 0)); // [::]:0
|
||||||
let socket = UdpSocket::bind(&local_addr).await?;
|
let socket = UdpSocket::bind(&local_addr).await?;
|
||||||
let Some(stun_addr) = stun_server
|
// Resolve via tokio so DNS never blocks the async runtime worker.
|
||||||
.to_socket_addrs()?
|
let Some(stun_addr) = tokio::net::lookup_host(stun_server)
|
||||||
.filter(|x| x.is_ipv6())
|
.await?
|
||||||
.next()
|
.find(|x| x.is_ipv6())
|
||||||
else {
|
else {
|
||||||
bail!(
|
bail!(
|
||||||
"Failed to resolve STUN ipv6 server address: {}",
|
"Failed to resolve STUN ipv6 server address: {}",
|
||||||
@@ -2468,14 +2479,13 @@ async fn stun_ipv6_test(stun_server: &str) -> ResultType<(SocketAddr, String)> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn stun_ipv4_test(stun_server: &str) -> ResultType<(SocketAddr, String)> {
|
async fn stun_ipv4_test(stun_server: &str) -> ResultType<(SocketAddr, String)> {
|
||||||
use std::net::ToSocketAddrs;
|
|
||||||
use stunclient::StunClient;
|
use stunclient::StunClient;
|
||||||
let local_addr = SocketAddr::from(([0u8; 4], 0));
|
let local_addr = SocketAddr::from(([0u8; 4], 0));
|
||||||
let socket = UdpSocket::bind(&local_addr).await?;
|
let socket = UdpSocket::bind(&local_addr).await?;
|
||||||
let Some(stun_addr) = stun_server
|
// Resolve via tokio so DNS never blocks the async runtime worker.
|
||||||
.to_socket_addrs()?
|
let Some(stun_addr) = tokio::net::lookup_host(stun_server)
|
||||||
.filter(|x| x.is_ipv4())
|
.await?
|
||||||
.next()
|
.find(|x| x.is_ipv4())
|
||||||
else {
|
else {
|
||||||
bail!(
|
bail!(
|
||||||
"Failed to resolve STUN ipv4 server address: {}",
|
"Failed to resolve STUN ipv4 server address: {}",
|
||||||
@@ -2487,7 +2497,7 @@ async fn stun_ipv4_test(stun_server: &str) -> ResultType<(SocketAddr, String)> {
|
|||||||
Ok(if addr.ip().is_ipv4() {
|
Ok(if addr.ip().is_ipv4() {
|
||||||
(addr, stun_server.to_owned())
|
(addr, stun_server.to_owned())
|
||||||
} else {
|
} else {
|
||||||
bail!("STUN server returned non-IPv6 address: {}", addr)
|
bail!("STUN server returned non-IPv4 address: {}", addr)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2526,10 +2536,9 @@ pub async fn test_nat_ipv4() -> ResultType<(SocketAddr, String)> {
|
|||||||
async fn test_bind_ipv6() -> ResultType<SocketAddr> {
|
async fn test_bind_ipv6() -> ResultType<SocketAddr> {
|
||||||
let local_addr = SocketAddr::from(([0u16; 8], 0)); // [::]:0
|
let local_addr = SocketAddr::from(([0u16; 8], 0)); // [::]:0
|
||||||
let socket = UdpSocket::bind(local_addr).await?;
|
let socket = UdpSocket::bind(local_addr).await?;
|
||||||
let addr = STUNS_V6[0]
|
let addr = tokio::net::lookup_host(STUNS_V6[0])
|
||||||
.to_socket_addrs()?
|
.await?
|
||||||
.filter(|x| x.is_ipv6())
|
.find(|x| x.is_ipv6())
|
||||||
.next()
|
|
||||||
.ok_or_else(|| {
|
.ok_or_else(|| {
|
||||||
anyhow!(
|
anyhow!(
|
||||||
"Failed to resolve STUN ipv6 server address: {}",
|
"Failed to resolve STUN ipv6 server address: {}",
|
||||||
@@ -2660,7 +2669,14 @@ pub async fn punch_udp(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
res = socket.recv(&mut data) => match res {
|
res = socket.recv(&mut data) => match res {
|
||||||
Err(e) => bail!("UDP punch failed, {packets_sent} packets sent: {e}"),
|
Err(e) => {
|
||||||
|
// While the hole is still forming, ICMP unreachable from the peer's NAT
|
||||||
|
// is expected and surfaces as ConnectionReset/Refused on a connected
|
||||||
|
// socket (notably 10054 on Windows). Treat it as loss and keep punching;
|
||||||
|
// MAX_TIME above still bounds the whole attempt.
|
||||||
|
log::debug!("UDP punch recv error (treated as loss): {e}");
|
||||||
|
hbb_common::sleep(0.01).await;
|
||||||
|
}
|
||||||
Ok(n) => {
|
Ok(n) => {
|
||||||
// log::debug!("UDP punch succeeded after sending {} packets after {:?}", packets_sent, tm.elapsed());
|
// log::debug!("UDP punch succeeded after sending {} packets after {:?}", packets_sent, tm.elapsed());
|
||||||
if listen {
|
if listen {
|
||||||
|
|||||||
@@ -20,6 +20,22 @@ pub struct KcpStream {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl KcpStream {
|
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
|
||||||
|
// links amplifies brief loss into a spiral users experience as stalls or drops. This is
|
||||||
|
// sender-side only, so no wire negotiation is needed and either peer may run either
|
||||||
|
// profile. Requires kcp-sys from the `rustdesk-patches` branch, which wires the config
|
||||||
|
// factory into connection setup (on older revs the factory was stored but never consulted).
|
||||||
|
fn apply_kcp_config(endpoint: &mut KcpEndpoint) {
|
||||||
|
if crate::get_kcp_cc_enabled() {
|
||||||
|
endpoint.set_kcp_config_factory(Box::new(|conv| {
|
||||||
|
let mut config = kcp_sys::ffi_safe::KcpConfig::new_turbo(conv);
|
||||||
|
config.nc = Some(0);
|
||||||
|
config
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn create_framed(stream: stream::KcpStream, local_addr: Option<SocketAddr>) -> Stream {
|
fn create_framed(stream: stream::KcpStream, local_addr: Option<SocketAddr>) -> Stream {
|
||||||
Stream::Tcp(FramedStream(
|
Stream::Tcp(FramedStream(
|
||||||
tokio_util::codec::Framed::new(DynTcpStream(Box::new(stream)), BytesCodec::new()),
|
tokio_util::codec::Framed::new(DynTcpStream(Box::new(stream)), BytesCodec::new()),
|
||||||
@@ -35,6 +51,7 @@ impl KcpStream {
|
|||||||
init_packet: Option<BytesMut>,
|
init_packet: Option<BytesMut>,
|
||||||
) -> ResultType<(Self, Stream)> {
|
) -> ResultType<(Self, Stream)> {
|
||||||
let mut endpoint = KcpEndpoint::new();
|
let mut endpoint = KcpEndpoint::new();
|
||||||
|
Self::apply_kcp_config(&mut endpoint);
|
||||||
endpoint.run().await;
|
endpoint.run().await;
|
||||||
|
|
||||||
let (input, output) = (
|
let (input, output) = (
|
||||||
@@ -70,6 +87,7 @@ impl KcpStream {
|
|||||||
timeout: std::time::Duration,
|
timeout: std::time::Duration,
|
||||||
) -> ResultType<(Self, Stream)> {
|
) -> ResultType<(Self, Stream)> {
|
||||||
let mut endpoint = KcpEndpoint::new();
|
let mut endpoint = KcpEndpoint::new();
|
||||||
|
Self::apply_kcp_config(&mut endpoint);
|
||||||
endpoint.run().await;
|
endpoint.run().await;
|
||||||
|
|
||||||
let (input, output) = (
|
let (input, output) = (
|
||||||
@@ -104,6 +122,13 @@ impl KcpStream {
|
|||||||
let udp = udp_socket.clone();
|
let udp = udp_socket.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
let mut buf = vec![0; 1500];
|
let mut buf = vec![0; 1500];
|
||||||
|
// A connected UDP socket surfaces ICMP port-unreachable as an error on
|
||||||
|
// send/recv (WSAECONNRESET 10054 on Windows, ECONNREFUSED on Linux). For UDP
|
||||||
|
// these are advisory: a stray ICMP from a NAT rebind glitch or a momentary
|
||||||
|
// peer hiccup does not mean the path is dead, and KCP retransmits through it.
|
||||||
|
// Treat socket errors as packet loss instead of tearing the session down;
|
||||||
|
// 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.
|
||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
_ = &mut stop_receiver => {
|
_ = &mut stop_receiver => {
|
||||||
@@ -112,8 +137,8 @@ impl KcpStream {
|
|||||||
}
|
}
|
||||||
Some(data) = output.recv() => {
|
Some(data) = output.recv() => {
|
||||||
if let Err(e) = udp.send(&data.inner()).await {
|
if let Err(e) = udp.send(&data.inner()).await {
|
||||||
log::debug!("KCP send error: {:?}", e);
|
log::debug!("KCP send error (treated as loss): {:?}", e);
|
||||||
break;
|
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
result = udp.recv_from(&mut buf) => {
|
result = udp.recv_from(&mut buf) => {
|
||||||
@@ -127,8 +152,8 @@ impl KcpStream {
|
|||||||
.await.ok();
|
.await.ok();
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
log::debug!("KCP recv_from error: {:?}", e);
|
log::debug!("KCP recv_from error (treated as loss): {:?}", e);
|
||||||
break;
|
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user