Manage UDP associations with LFU strategy (#506)

* manage UDP associations with LFU

- DNS associations could be evicted firstly then others like HTTP/3

* UDP redir caches inbound non-local socket

- Optimization: prevent creating new sockets for the same remotes

* updated libc to v0.2.94 for unified TCP options

* make clippy happy

* Pin lfu-cache to v1.2.1 for edward-shen/lfu-cache#2

* UDP target should cache with expiration 1hrs

* UDP tunnel test with an echo server

- CI crashes occasionally because of 8.8.8.8:53 doesn't respond

Author: zonyitoo

Cargo.lock

@@ -357,7 +357,7 @@ fn main() {
 
         #[cfg(any(target_os = "linux", target_os = "android", target_os = "macos", target_os = "ios"))]
         if let Some(iface) = matches.value_of("OUTBOUND_BIND_INTERFACE") {
-            config.outbound_bind_interface = Some(From::from(iface.to_owned()));
+            config.outbound_bind_interface = Some(iface.to_owned());
         }
 
         #[cfg(all(unix, not(target_os = "android")))]

bin/sslocal.rs

@@ -163,9 +163,9 @@ fn main() {
             config.outbound_fwmark = Some(mark.parse::<u32>().expect("an unsigned integer for `outbound-fwmark`"));
         }
 
-        #[cfg(any(target_os = "linux", target_os = "android"))]
+        #[cfg(any(target_os = "linux", target_os = "android", target_os = "macos", target_os = "ios"))]
         if let Some(iface) = matches.value_of("OUTBOUND_BIND_INTERFACE") {
-            config.outbound_bind_interface = Some(From::from(iface.to_owned()));
+            config.outbound_bind_interface = Some(iface.to_owned());
         }
 
         if let Some(m) = matches.value_of("MANAGER_ADDRESS") {

bin/ssmanager.rs

@@ -204,9 +204,9 @@ fn main() {
             config.outbound_fwmark = Some(mark.parse::<u32>().expect("an unsigned integer for `outbound-fwmark`"));
         }
 
-        #[cfg(any(target_os = "linux", target_os = "android"))]
+        #[cfg(any(target_os = "linux", target_os = "android", target_os = "macos", target_os = "ios"))]
         if let Some(iface) = matches.value_of("OUTBOUND_BIND_INTERFACE") {
-            config.outbound_bind_interface = Some(From::from(iface.to_owned()));
+            config.outbound_bind_interface = Some(iface.to_owned());
         }
 
         if let Some(m) = matches.value_of("MANAGER_ADDRESS") {

bin/ssserver.rs

@@ -77,7 +77,7 @@ once_cell = "1.7"
 thiserror = "1.0"
 
 spin = { version = "0.9", features = ["std"] }
-lru_time_cache = "0.11"
+lfu_cache = "1.2.1"
 bytes = "1.0"
 byte_string = "1.0"
 byteorder = "1.3"

crates/shadowsocks-service/Cargo.toml

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use std::{net::IpAddr, time::Duration};
 
 #[cfg(feature = "local-dns")]
-use lru_time_cache::LruCache;
+use lfu_cache::TimedLfuCache;
 use shadowsocks::{
     config::ServerType,
     context::{Context, SharedContext},
@@ -32,7 +32,7 @@ pub struct ServiceContext {
 
     // For DNS relay's ACL domain name reverse lookup -- whether the IP shall be forwarded
     #[cfg(feature = "local-dns")]
-    reverse_lookup_cache: Mutex<LruCache<IpAddr, bool>>,
+    reverse_lookup_cache: Mutex<TimedLfuCache<IpAddr, bool>>,
 }
 
 impl Default for ServiceContext {
@@ -51,7 +51,10 @@ impl ServiceContext {
             acl: None,
             flow_stat: Arc::new(FlowStat::new()),
             #[cfg(feature = "local-dns")]
-            reverse_lookup_cache: Mutex::new(LruCache::with_expiry_duration(Duration::from_secs(3 * 24 * 60 * 60))),
+            reverse_lookup_cache: Mutex::new(TimedLfuCache::with_capacity_and_expiration(
+                10240, // XXX: It should be enough for a normal user.
+                Duration::from_secs(3 * 24 * 60 * 60),
+            )),
         }
     }
 

crates/shadowsocks-service/src/local/context.rs

@@ -3,7 +3,7 @@
 use std::{sync::Arc, time::Duration};
 
 use hyper::{Body, Client};
-use lru_time_cache::LruCache;
+use lfu_cache::TimedLfuCache;
 use shadowsocks::config::ServerAddr;
 use tokio::sync::Mutex;
 
@@ -14,14 +14,14 @@ use super::{connector::ProxyConnector, http_client::ProxyHttpClient};
 /// Cached HTTP client for remote servers
 pub struct ProxyClientCache {
     context: Arc<ServiceContext>,
-    cache: Mutex<LruCache<ServerAddr, ProxyHttpClient>>,
+    cache: Mutex<TimedLfuCache<ServerAddr, ProxyHttpClient>>,
 }
 
 impl ProxyClientCache {
     pub fn new(context: Arc<ServiceContext>) -> ProxyClientCache {
         ProxyClientCache {
             context,
-            cache: Mutex::new(LruCache::with_expiry_duration_and_capacity(Duration::from_secs(60), 5)),
+            cache: Mutex::new(TimedLfuCache::with_capacity_and_expiration(5, Duration::from_secs(60))),
         }
     }
 

crates/shadowsocks-service/src/local/http/client_cache.rs

@@ -59,7 +59,7 @@ impl AutoProxyClientStream {
         let addr = addr.into();
         let stream =
             TcpStream::connect_remote_with_opts(context.context_ref(), &addr, context.connect_opts_ref()).await?;
-        Ok(AutoProxyClientStream::Bypassed(stream.into()))
+        Ok(AutoProxyClientStream::Bypassed(stream))
     }
 
     /// Connect to target `addr` via shadowsocks' server configured by `svr_cfg`

crates/shadowsocks-service/src/local/net/tcp/auto_proxy_stream.rs

@@ -10,8 +10,8 @@ use std::{
 use async_trait::async_trait;
 use bytes::Bytes;
 use futures::future::{self, AbortHandle};
+use lfu_cache::TimedLfuCache;
 use log::{debug, error, trace, warn};
-use lru_time_cache::{Entry, LruCache};
 use shadowsocks::{
     lookup_then,
     net::UdpSocket as ShadowUdpSocket,
@@ -42,14 +42,17 @@ pub trait UdpInboundWrite {
     async fn send_to(&self, peer_addr: SocketAddr, remote_addr: &Address, data: &[u8]) -> io::Result<()>;
 }
 
+type AssociationMap<W> = TimedLfuCache<SocketAddr, UdpAssociation<W>>;
+type SharedAssociationMap<W> = Arc<Mutex<AssociationMap<W>>>;
+
 /// UDP association manager
 pub struct UdpAssociationManager<W>
 where
     W: UdpInboundWrite + Clone + Send + Sync + Unpin + 'static,
 {
     respond_writer: W,
     context: Arc<ServiceContext>,
-    assoc_map: Arc<Mutex<LruCache<SocketAddr, UdpAssociation<W>>>>,
+    assoc_map: SharedAssociationMap<W>,
     cleanup_abortable: AbortHandle,
     balancer: PingBalancer,
 }
@@ -77,8 +80,8 @@ where
     ) -> UdpAssociationManager<W> {
         let time_to_live = time_to_live.unwrap_or(crate::DEFAULT_UDP_EXPIRY_DURATION);
         let assoc_map = Arc::new(Mutex::new(match capacity {
-            Some(capacity) => LruCache::with_expiry_duration_and_capacity(time_to_live, capacity),
-            None => LruCache::with_expiry_duration(time_to_live),
+            Some(capacity) => TimedLfuCache::with_capacity_and_expiration(capacity, time_to_live),
+            None => TimedLfuCache::with_expiration(time_to_live),
         }));
 
         let cleanup_abortable = {
@@ -87,8 +90,8 @@ where
                 loop {
                     time::sleep(time_to_live).await;
 
-                    // iter() will trigger a cleanup of expired associations
-                    let _ = assoc_map.lock().await.iter();
+                    // cleanup expired associations
+                    let _ = assoc_map.lock().await.evict_expired();
                 }
             });
             tokio::spawn(cleanup_task);
@@ -107,23 +110,27 @@ where
     /// Sends `data` from `peer_addr` to `target_addr`
     pub async fn send_to(&self, peer_addr: SocketAddr, target_addr: Address, data: &[u8]) -> io::Result<()> {
         // Check or (re)create an association
-        match self.assoc_map.lock().await.entry(peer_addr) {
-            Entry::Occupied(occ) => {
-                let assoc = occ.into_mut();
-                assoc.try_send((target_addr, Bytes::copy_from_slice(data)))
-            }
-            Entry::Vacant(vac) => {
-                let assoc = vac.insert(UdpAssociation::new(
-                    self.context.clone(),
-                    peer_addr,
-                    self.assoc_map.clone(),
-                    self.balancer.clone(),
-                    self.respond_writer.clone(),
-                ));
-                trace!("created udp association for {}", peer_addr);
-                assoc.try_send((target_addr, Bytes::copy_from_slice(data)))
-            }
+
+        let mut assoc_map = self.assoc_map.lock().await;
+
+        if let Some(assoc) = assoc_map.get(&peer_addr) {
+            return assoc.try_send((target_addr, Bytes::copy_from_slice(data)));
         }
+
+        let assoc = UdpAssociation::new(
+            self.context.clone(),
+            peer_addr,
+            self.assoc_map.clone(),
+            self.balancer.clone(),
+            self.respond_writer.clone(),
+        );
+
+        trace!("created udp association for {}", peer_addr);
+
+        assoc.try_send((target_addr, Bytes::copy_from_slice(data)))?;
+        assoc_map.insert(peer_addr, assoc);
+
+        Ok(())
     }
 }
 
@@ -153,7 +160,7 @@ where
     fn new(
         context: Arc<ServiceContext>,
         peer_addr: SocketAddr,
-        assoc_map: Arc<Mutex<LruCache<SocketAddr, UdpAssociation<W>>>>,
+        assoc_map: SharedAssociationMap<W>,
         balancer: PingBalancer,
         respond_writer: W,
     ) -> UdpAssociation<W> {
@@ -245,7 +252,7 @@ where
     bypassed_ipv4_socket: SpinMutex<UdpAssociationBypassState>,
     bypassed_ipv6_socket: SpinMutex<UdpAssociationBypassState>,
     proxied_socket: SpinMutex<UdpAssociationProxyState>,
-    assoc_map: Arc<Mutex<LruCache<SocketAddr, UdpAssociation<W>>>>,
+    assoc_map: SharedAssociationMap<W>,
     balancer: PingBalancer,
     respond_writer: W,
 }
@@ -266,7 +273,7 @@ where
     fn new(
         context: Arc<ServiceContext>,
         peer_addr: SocketAddr,
-        assoc_map: Arc<Mutex<LruCache<SocketAddr, UdpAssociation<W>>>>,
+        assoc_map: SharedAssociationMap<W>,
         balancer: PingBalancer,
         respond_writer: W,
     ) -> (Arc<UdpAssociationContext<W>>, mpsc::Sender<(Address, Bytes)>) {