wip

clash_lib/Cargo.toml

@@ -121,6 +121,8 @@ console-subscriber = { version = "0.2.0" }
 tracing-timing = { version = "0.6.0" }
 criterion = { version = "0.5", features = ["html_reports", "async_tokio"], optional = true }
 
+cmd_lib = "1"
+
 [dev-dependencies]
 tempfile = "3.10"
 ctor = "0.2"

clash_lib/src/app/dispatcher/dispatcher_impl.rs

@@ -317,7 +317,7 @@ impl Dispatcher {
                                 match remote_w.send(packet).await {
                                     Ok(_) => {}
                                     Err(err) => {
-                                        warn!("failed to send packet to remote: {}", err);
+                                        warn!("failed to send packet to remote: {:?}", err);
                                     }
                                 }
                             }

clash_lib/src/lib.rs

@@ -16,11 +16,12 @@ use common::auth;
 use common::http::new_http_client;
 use common::mmdb;
 use config::def::LogLevel;
+use once_cell::sync::OnceCell;
 use proxy::tun::get_tun_runner;
 
 use std::io;
 use std::path::PathBuf;
-use std::sync::{Arc, OnceLock};
+use std::sync::Arc;
 use thiserror::Error;
 use tokio::sync::{broadcast, mpsc, oneshot, Mutex};
 use tokio::task::JoinHandle;
@@ -102,9 +103,10 @@ pub struct GlobalState {
 
 pub struct RuntimeController {
     shutdown_tx: mpsc::Sender<()>,
+    broadcast_shutdown: broadcast::Sender<()>
 }
 
-static RUNTIME_CONTROLLER: OnceLock<std::sync::RwLock<RuntimeController>> = OnceLock::new();
+static RUNTIME_CONTROLLER: OnceCell<RuntimeController> = OnceCell::new();
 
 pub fn start(opts: Options) -> Result<(), Error> {
     let rt = match opts.rt.as_ref().unwrap_or(&TokioRuntime::MultiThread) {
@@ -127,17 +129,32 @@ pub fn start(opts: Options) -> Result<(), Error> {
     })
 }
 
-pub fn shutdown() -> bool {
-    match RUNTIME_CONTROLLER.get().unwrap().write() {
-        Ok(rt) => rt.shutdown_tx.blocking_send(()).is_ok(),
-        _ => false,
+pub fn shutdown() -> anyhow::Result<()> {
+    match RUNTIME_CONTROLLER.get() {
+        Some(rt) => {
+            _ = rt.broadcast_shutdown.send(()); // fail when there is no receiver
+            rt.shutdown_tx.blocking_send(())?;
+            Ok(())
+        },
+        _ => Err(anyhow::anyhow!("Empty RUNTIME_CONTROLLER")),
+    }
+}
+
+pub async fn listen_shutdown()-> anyhow::Result<()> {
+    match RUNTIME_CONTROLLER.get() {
+        Some(rt) => {
+            _ = rt.broadcast_shutdown.subscribe().recv().await; // fail when there is no sender or buffer is full
+            Ok(())
+        },
+        _ => Err(anyhow::anyhow!("Empty RUNTIME_CONTROLLER")),
     }
 }
 
 async fn start_async(opts: Options) -> Result<(), Error> {
     let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1);
+    let (broadcast_shutdown, _) = broadcast::channel(1);
 
-    let _ = RUNTIME_CONTROLLER.set(std::sync::RwLock::new(RuntimeController { shutdown_tx }));
+    let _ = RUNTIME_CONTROLLER.set(RuntimeController { shutdown_tx, broadcast_shutdown });
 
     let config: InternalConfig = opts.config.try_parse()?;
 

clash_lib/src/proxy/converters/tuic.rs

@@ -37,7 +37,8 @@ impl TryFrom<&OutboundTuic> for AnyOutboundHandler {
                 .unwrap_or_default(),
             heartbeat_interval: Duration::from_millis(s.heartbeat_interval.unwrap_or(3000)),
             reduce_rtt: s.reduce_rtt.unwrap_or(false) || s.fast_open.unwrap_or(false),
-            request_timeout: Duration::from_millis(s.request_timeout.unwrap_or(8000)),
+            request_timeout: Duration::from_millis(s.request_timeout.unwrap_or(4000)),
+            idle_timeout: VarInt::from_u64(s.request_timeout.unwrap_or(4000)).unwrap_or(VarInt::MAX),
             congestion_controller: s
                 .congestion_controller
                 .clone()

clash_lib/src/proxy/tuic/handle_task.rs

@@ -26,9 +26,12 @@ impl TuicConnection {
             .authenticate(self.uuid, self.password.clone())
             .await
         {
-            Ok(()) => tracing::info!("[auth] {uuid}", uuid = self.uuid),
+            Ok(()) => tracing::info!("[auth] success {uuid}", uuid = self.uuid),
             Err(err) => {
-                tracing::warn!("[auth] authentication sending error: {err}")
+                tracing::warn!(
+                    "[auth] authentication sending error: {:?}",
+                    anyhow::anyhow!(err)
+                )
             }
         }
     }
@@ -141,7 +144,7 @@ impl TuicConnection {
         }
 
         match self.inner.heartbeat().await {
-            Ok(()) => tracing::trace!("[heartbeat]"),
+            Ok(()) => tracing::debug!("[heartbeat]"),
             Err(err) => tracing::error!("[heartbeat] {err}"),
         }
         Ok(())

clash_lib/src/proxy/tuic/mod.rs

@@ -6,6 +6,7 @@ pub(crate) mod types;
 use crate::proxy::tuic::types::SocketAdderTrans;
 use anyhow::Result;
 use axum::async_trait;
+use quinn::congestion::{BbrConfig, NewRenoConfig};
 use quinn::{EndpointConfig, TokioRuntime};
 use std::net::SocketAddr;
 use std::{
@@ -61,6 +62,7 @@ pub struct HandlerOptions {
     pub heartbeat_interval: Duration,
     pub reduce_rtt: bool,
     pub request_timeout: Duration,
+    pub idle_timeout: VarInt,
     pub congestion_controller: CongestionControl,
     pub max_udp_relay_packet_size: u64,
     pub max_open_stream: VarInt,
@@ -147,15 +149,26 @@ impl Handler {
         crypto.enable_early_data = true;
         crypto.enable_sni = !opts.disable_sni;
         let mut quinn_config = QuinnConfig::new(Arc::new(crypto));
-        let mut quinn_transport_config = QuinnTransportConfig::default();
-        quinn_transport_config
+        let mut transport_config = QuinnTransportConfig::default();
+        transport_config
             .max_concurrent_bidi_streams(opts.max_open_stream)
             .max_concurrent_uni_streams(opts.max_open_stream)
             .send_window(opts.send_window)
             .stream_receive_window(opts.receive_window)
-            .max_idle_timeout(None)
-            .congestion_controller_factory(Arc::new(CubicConfig::default()));
-        quinn_config.transport_config(Arc::new(quinn_transport_config));
+            .max_idle_timeout(Some(VarInt::from_u32(3_000).into()));
+        match opts.congestion_controller {
+            CongestionControl::Cubic => {
+                transport_config.congestion_controller_factory(Arc::new(CubicConfig::default()))
+            }
+            CongestionControl::NewReno => {
+                transport_config.congestion_controller_factory(Arc::new(NewRenoConfig::default()))
+            }
+            CongestionControl::Bbr => {
+                transport_config.congestion_controller_factory(Arc::new(BbrConfig::default()))
+            }
+        };
+
+        quinn_config.transport_config(Arc::new(transport_config));
         // Try to create an IPv4 socket as the placeholder first, if it fails, try IPv6.
         let socket =
             UdpSocket::bind(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))).or_else(|err| {

clash_lib/src/proxy/tuic/types.rs

@@ -4,9 +4,9 @@ use quinn::Connection as QuinnConnection;
 use quinn::{Endpoint as QuinnEndpoint, ZeroRttAccepted};
 use register_count::Counter;
 use std::collections::HashMap;
+use std::str::FromStr;
 use std::{
     net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket},
-    str::FromStr,
     sync::{atomic::AtomicU32, Arc},
     time::Duration,
 };
@@ -177,7 +177,7 @@ impl TuicConnection {
             };
         };
 
-        tracing::warn!("connection error: {err}");
+        tracing::warn!("connection error: {err:?}");
     }
 }
 

clash_lib/src/proxy/tun/auto_route.rs

@@ -0,0 +1,56 @@
+use crate::config::internal::config::TunConfig;
+
+#[cfg(target_os = "linux")]
+pub async fn setup(cfg: &mut TunConfig, tun_name: &str) -> anyhow::Result<()> {
+    use cmd_lib::run_cmd;
+
+    use crate::listen_shutdown;
+
+    if !cfg.auto_route.unwrap_or(false) {
+        return Ok(());
+    }
+    let mark = cfg.mark.unwrap_or(6969);
+    cfg.mark = Some(mark);
+    let table = cfg.table.take().unwrap_or("2233".into());
+    cfg.table = Some(table.clone());
+
+    let ipv6 = false; // TODO read from conf
+    // TODO chen perm
+    run_cmd! {
+        ip route add default dev $tun_name table $table;
+        ip rule add ipproto icmp table main;
+        ip rule add not fwmark $mark table $table;
+        ip rule add table main suppress_prefixlength 0;
+    }?;
+    if ipv6 {
+        run_cmd! {
+            ip -6 route add default dev $tun_name table $table;
+            ip -6 rule add ipproto icmp table main;
+            ip -6 rule add not fwmark $mark table $table;
+            ip -6 rule add table main suppress_prefixlength 0;
+        }?;
+    }
+    tokio::spawn(async move {
+        listen_shutdown().await.unwrap();
+        tracing::info!("cleaning routes");
+        run_cmd!{
+            ip rule del not from all fwmark $mark lookup $table;
+            ip rule del from all lookup main suppress_prefixlength 0;
+            ip rule del from all ipproto icmp lookup main;
+        }.unwrap();
+        if ipv6 {
+            run_cmd!{
+                ip -6 rule del not from all fwmark $mark lookup $table;
+                ip -6 rule del from all lookup main suppress_prefixlength 0;
+                ip -6 rule del from all ipproto icmp lookup main;
+            }.unwrap();
+        }
+    });
+    Ok(())
+}
+
+#[cfg(not(target_os = "linux"))]
+pub fn setup(cfg: &mut TunConfig, tun_name: &str) -> anyhow::Result<()> {
+    tracing::error!("Auto route not impl!");
+    Ok(())
+}