Merge branch 'master' into dependabot/cargo/rust-dependencies-c9f23a8216

Signed-off-by: Yuwei Ba <dev@watfaq.com>

Author: ibigbug

clash_lib/src/app/dispatcher/dispatcher_impl.rs

@@ -17,7 +17,7 @@ use std::{
     collections::HashMap,
     fmt::{Debug, Formatter},
     net::SocketAddr,
-    sync::{Arc, Mutex},
+    sync::Arc,
     time::{Duration, Instant},
 };
 use tokio::{
@@ -35,7 +35,7 @@ pub struct Dispatcher {
     outbound_manager: ThreadSafeOutboundManager,
     router: ThreadSafeRouter,
     resolver: ThreadSafeDNSResolver,
-    mode: Arc<Mutex<RunMode>>,
+    mode: Arc<RwLock<RunMode>>,
 
     manager: Arc<Manager>,
 }
@@ -59,19 +59,19 @@ impl Dispatcher {
             outbound_manager,
             router,
             resolver,
-            mode: Arc::new(Mutex::new(mode)),
+            mode: Arc::new(RwLock::new(mode)),
             manager: statistics_manager,
         }
     }
 
     pub async fn set_mode(&self, mode: RunMode) {
         info!("run mode switched to {}", mode);
 
-        *self.mode.lock().unwrap() = mode;
+        *self.mode.write().await = mode;
     }
 
     pub async fn get_mode(&self) -> RunMode {
-        *self.mode.lock().unwrap()
+        *self.mode.read().await
     }
 
     #[instrument(skip(self, sess, lhs))]
@@ -120,7 +120,7 @@ impl Dispatcher {
 
         sess.destination = dest.clone();
 
-        let mode = *self.mode.lock().unwrap();
+        let mode = *self.mode.read().await;
         let (outbound_name, rule) = match mode {
             RunMode::Global => (PROXY_GLOBAL, None),
             RunMode::Rule => self.router.match_route(&mut sess).await,
@@ -241,7 +241,7 @@ impl Dispatcher {
     /// Dispatch a UDP packet to outbound handler
     /// returns the close sender
     #[instrument]
-    pub fn dispatch_datagram(
+    pub async fn dispatch_datagram(
         &self,
         sess: Session,
         udp_inbound: AnyInboundDatagram,
@@ -311,7 +311,7 @@ impl Dispatcher {
                 // do Ip though?
                 packet.dst_addr = dest;
 
-                let mode = *mode.lock().unwrap();
+                let mode = *mode.read().await;
 
                 let (outbound_name, rule) = match mode {
                     RunMode::Global => (PROXY_GLOBAL, None),

clash_lib/src/app/dns/filters.rs

@@ -17,7 +17,7 @@ impl GeoIPFilter {
 impl FallbackIPFilter for GeoIPFilter {
     fn apply(&self, ip: &net::IpAddr) -> bool {
         self.1
-            .lookup_contry(*ip)
+            .lookup_country(*ip)
             .map(|x| x.country)
             .is_ok_and(|x| x.is_some_and(|x| x.iso_code == Some(self.0.as_str())))
     }

clash_lib/src/app/logging.rs

@@ -49,8 +49,8 @@ pub struct LogEvent {
 pub struct EventCollector(Vec<Sender<LogEvent>>);
 
 impl EventCollector {
-    pub fn new(recivers: Vec<Sender<LogEvent>>) -> Self {
-        Self(recivers)
+    pub fn new(receivers: Vec<Sender<LogEvent>>) -> Self {
+        Self(receivers)
     }
 }
 
@@ -180,8 +180,8 @@ pub fn setup_logging(
     tracing::subscriber::set_global_default(subscriber)
         .map_err(|x| anyhow!("setup logging error: {}", x))?;
 
-    if let Ok(jager_endpiont) = std::env::var("JAGER_ENDPOINT") {
-        debug!("jager endpoint: {}", jager_endpiont);
+    if let Ok(jager_endpoint) = std::env::var("JAGER_ENDPOINT") {
+        debug!("jager endpoint: {}", jager_endpoint);
     }
 
     Ok(g)

clash_lib/src/app/profile/mod.rs

@@ -108,7 +108,7 @@ impl CacheFile {
                 Ok(db) => db,
                 Err(e) => {
                     error!(
-                        "failed to parse cache file: {}, initilizing a new one",
+                        "failed to parse cache file: {}, initializing a new one",
                         e
                     );
                     Db {

clash_lib/src/app/router/mod.rs

@@ -108,7 +108,7 @@ impl Router {
             let mayby_ip = sess.resolved_ip.or(sess.destination.ip());
             if let (Some(ip), Some(asn_mmdb)) = (mayby_ip, &self.asn_mmdb) {
                 // try simplified mmdb first
-                let rv = asn_mmdb.lookup_contry(ip);
+                let rv = asn_mmdb.lookup_country(ip);
                 if let Ok(country) = rv {
                     sess.asn = country
                         .country

clash_lib/src/app/router/rules/geoip.rs

@@ -29,7 +29,7 @@ impl RuleMatcher for GeoIP {
         };
 
         if let Some(ip) = ip {
-            match self.mmdb.lookup_contry(ip) {
+            match self.mmdb.lookup_country(ip) {
                 Ok(country) => {
                     country
                         .country

clash_lib/src/common/io.rs

@@ -2,11 +2,13 @@
 use std::future::Future;
 use std::{
     io,
+    mem::MaybeUninit,
     pin::Pin,
     task::{Context, Poll},
     time::Duration,
 };
 
+use bytes::BytesMut;
 use futures::ready;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
@@ -91,7 +93,7 @@ impl CopyBuffer {
         })
     }
 
-    pub fn amount_transfered(&self) -> u64 {
+    pub fn amount_transferred(&self) -> u64 {
         self.amt
     }
 
@@ -235,7 +237,7 @@ where
                                 match delay.as_mut().poll(cx) {
                                     Poll::Ready(()) => {
                                         *a_to_b = TransferState::ShuttingDown(
-                                            buf.amount_transfered(),
+                                            buf.amount_transferred(),
                                         );
                                         continue;
                                     }
@@ -285,7 +287,7 @@ where
                                 match delay.as_mut().poll(cx) {
                                     Poll::Ready(()) => {
                                         *b_to_a = TransferState::ShuttingDown(
-                                            buf.amount_transfered(),
+                                            buf.amount_transferred(),
                                         );
                                         continue;
                                     }
@@ -352,3 +354,56 @@ where
     }
     .await
 }
+
+pub trait ReadExactBase {
+    /// inner stream to be polled
+    type I: AsyncRead + Unpin;
+    /// prepare the inner stream, read buffer and read position
+    fn decompose(&mut self) -> (&mut Self::I, &mut BytesMut, &mut usize);
+}
+
+pub trait ReadExt: ReadExactBase {
+    fn poll_read_exact(
+        &mut self,
+        cx: &mut std::task::Context,
+        size: usize,
+    ) -> Poll<std::io::Result<()>>;
+}
+
+impl<T: ReadExactBase> ReadExt for T {
+    fn poll_read_exact(
+        &mut self,
+        cx: &mut std::task::Context,
+        size: usize,
+    ) -> Poll<std::io::Result<()>> {
+        let (raw, read_buf, read_pos) = self.decompose();
+        read_buf.reserve(size);
+        // # safety: read_buf has reserved `size`
+        unsafe { read_buf.set_len(size) }
+        loop {
+            if *read_pos < size {
+                // # safety: read_pos<size==read_buf.len(), and
+                // read_buf[0..read_pos] is initialized
+                let dst = unsafe {
+                    &mut *((&mut read_buf[*read_pos..size]) as *mut _
+                        as *mut [MaybeUninit<u8>])
+                };
+                let mut buf = ReadBuf::uninit(dst);
+                let ptr = buf.filled().as_ptr();
+                ready!(Pin::new(&mut *raw).poll_read(cx, &mut buf))?;
+                assert_eq!(ptr, buf.filled().as_ptr());
+                if buf.filled().is_empty() {
+                    return Poll::Ready(Err(std::io::Error::new(
+                        std::io::ErrorKind::UnexpectedEof,
+                        "unexpected eof",
+                    )));
+                }
+                *read_pos += buf.filled().len();
+            } else {
+                assert!(*read_pos == size);
+                *read_pos = 0;
+                return Poll::Ready(Ok(()));
+            }
+        }
+    }
+}

clash_lib/src/common/mmdb.rs

@@ -92,7 +92,7 @@ impl Mmdb {
         }
     }
 
-    pub fn lookup_contry(&self, ip: IpAddr) -> std::io::Result<geoip2::Country> {
+    pub fn lookup_country(&self, ip: IpAddr) -> std::io::Result<geoip2::Country> {
         self.reader
             .lookup::<geoip2::Country>(ip)
             .map_err(map_io_error)

clash_lib/src/config/def.rs

@@ -377,7 +377,7 @@ impl FromStr for Config {
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         let mut val: Value = serde_yaml::from_str(s).map_err(|x| {
             Error::InvalidConfig(format!(
-                "cound not parse config content {}: {}",
+                "couldn't not parse config content {}: {}",
                 s, x
             ))
         })?;
@@ -391,7 +391,7 @@ impl FromStr for Config {
 
         serde_yaml::from_value(val).map_err(|x| {
             Error::InvalidConfig(format!(
-                "cound not parse config content {}: {}",
+                "counldn't not parse config content {}: {}",
                 s, x
             ))
         })

clash_lib/src/lib.rs

@@ -249,7 +249,7 @@ async fn start_async(opts: Options) -> Result<(), Error> {
 
             let controller_cfg = config.general.controller.clone();
 
-            let new_componenets = create_components(cwd.clone(), config).await?;
+            let new_components = create_components(cwd.clone(), config).await?;
 
             done.send(()).unwrap();
 
@@ -267,31 +267,31 @@ async fn start_async(opts: Options) -> Result<(), Error> {
             }
 
             debug!("reloading inbound listener");
-            let inbound_listener_handle = new_componenets
+            let inbound_listener_handle = new_components
                 .inbound_manager
                 .lock()
                 .await
                 .get_runner()
                 .map(tokio::spawn)?;
 
             debug!("reloading tun runner");
-            let tun_runner_handle = new_componenets.tun_runner.map(tokio::spawn);
+            let tun_runner_handle = new_components.tun_runner.map(tokio::spawn);
 
             debug!("reloading dns listener");
-            let dns_listener_handle = new_componenets.dns_listener.map(tokio::spawn);
+            let dns_listener_handle = new_components.dns_listener.map(tokio::spawn);
 
             debug!("reloading api listener");
             let api_listener_handle = app::api::get_api_runner(
                 controller_cfg,
                 log_tx.clone(),
-                new_componenets.inbound_manager,
-                new_componenets.dispatcher,
+                new_components.inbound_manager,
+                new_components.dispatcher,
                 global_state.clone(),
-                new_componenets.dns_resolver,
-                new_componenets.outbound_manager,
-                new_componenets.statistics_manager,
-                new_componenets.cache_store,
-                new_componenets.router,
+                new_components.dns_resolver,
+                new_components.outbound_manager,
+                new_components.statistics_manager,
+                new_components.cache_store,
+                new_components.router,
                 cwd.to_string_lossy().to_string(),
             )
             .map(tokio::spawn);