From 8a4d21849e7d34b71668fa2ad21d696926c87050 Mon Sep 17 00:00:00 2001
From: lubeilin <1791778603@qq.com>
Date: Wed, 13 Mar 2024 21:23:32 +0800
Subject: [PATCH] =?UTF-8?q?[mio]=20=E8=B0=83=E6=95=B4=E4=BB=A3=E7=A0=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 vnt/src/channel/context.rs            | 51 +++++++++++++++++----------
 vnt/src/handle/handshaker.rs          |  2 +-
 vnt/src/handle/maintain/idle.rs       |  2 +-
 vnt/src/handle/mod.rs                 | 14 ++++++++
 vnt/src/handle/recv_data/client.rs    |  2 +-
 vnt/src/handle/recv_data/server.rs    |  2 +-
 vnt/src/handle/tun_tap/tun_handler.rs |  4 +--
 7 files changed, 52 insertions(+), 25 deletions(-)

diff --git a/vnt/src/channel/context.rs b/vnt/src/channel/context.rs
index 47a4a83..869a963 100644
--- a/vnt/src/channel/context.rs
+++ b/vnt/src/channel/context.rs
@@ -13,7 +13,6 @@ use rand::Rng;
 use crate::channel::punch::NatType;
 use crate::channel::sender::{AcceptSocketSender, ChannelSender, PacketSender};
 use crate::channel::{Route, RouteKey, UseChannelType, DEFAULT_RT};
-use crate::handle::{ConnectStatus, CurrentDeviceInfo};
 
 /// 传输通道上下文，持有udp socket、tcp socket和路由信息
 #[derive(Clone)]
@@ -153,20 +152,7 @@ impl ContextInner {
         }
         Ok(())
     }
-    pub fn change_status(
-        &self,
-        current_device: &AtomicCell<CurrentDeviceInfo>,
-        connect_status: ConnectStatus,
-    ) -> CurrentDeviceInfo {
-        loop {
-            let cur = current_device.load();
-            let mut new_info = cur;
-            new_info.status = connect_status;
-            if current_device.compare_exchange(cur, new_info).is_ok() {
-                return new_info;
-            }
-        }
-    }
+
     pub fn channel_num(&self) -> usize {
         self.main_udp_socket.len()
     }
@@ -392,6 +378,7 @@ impl RouteTable {
             }
         }
         if exist {
+            // 这个排序还有待优化，因为后加入的大概率排最后，被直接淘汰的概率也大，可能导致更好的通道被移除了
             list.sort_by_key(|(k, _)| k.rt);
             //如果延迟都稳定了，则去除多余通道
             for (route, _) in list.iter() {
@@ -399,7 +386,13 @@ impl RouteTable {
                     return;
                 }
             }
-            list.truncate(self.channel_num);
+            //延迟优先模式需要更多的通道探测延迟最低的路线
+            let limit_len = if self.first_latency {
+                self.channel_num + 2
+            } else {
+                self.channel_num
+            };
+            self.truncate_(list, limit_len);
         } else {
             if !self.first_latency {
                 if route.is_p2p() {
@@ -410,12 +403,32 @@ impl RouteTable {
             //增加路由表容量，避免波动
             let limit_len = self.channel_num * 2;
             list.sort_by_key(|(k, _)| k.rt);
-            if list.len() > limit_len {
-                list.truncate(limit_len);
-            }
+            self.truncate_(list, limit_len);
             list.push((route, AtomicCell::new(Instant::now())));
         }
     }
+    fn truncate_(&self, list: &mut Vec<(Route, AtomicCell<Instant>)>, len: usize) {
+        if list.len() <= len {
+            return;
+        }
+        if self.first_latency {
+            //找到第一个p2p通道
+            if let Some(index) =
+                list.iter()
+                    .enumerate()
+                    .find_map(|(index, (route, _))| if route.is_p2p() { Some(index) } else { None })
+            {
+                if index >= len {
+                    //保留第一个p2p通道
+                    let route = list.remove(index);
+                    list.truncate(len - 1);
+                    list.push(route);
+                    return;
+                }
+            }
+        }
+        list.truncate(len);
+    }
     pub fn route(&self, id: &Ipv4Addr) -> Option<Vec<Route>> {
         if let Some((_, v)) = self.route_table.read().get(id) {
             Some(v.iter().map(|(i, _)| *i).collect())
diff --git a/vnt/src/handle/handshaker.rs b/vnt/src/handle/handshaker.rs
index 092a1a2..ea3972b 100644
--- a/vnt/src/handle/handshaker.rs
+++ b/vnt/src/handle/handshaker.rs
@@ -37,7 +37,7 @@ impl Handshake {
     pub fn send(&self, context: &Context, secret: bool, addr: SocketAddr) -> io::Result<()> {
         let last = self.time.load();
         //短时间不重复发送
-        if last.elapsed() < Duration::from_secs(2) {
+        if last.elapsed() < Duration::from_secs(5) {
             return Ok(());
         }
         let request_packet = handshake_request_packet(secret)?;
diff --git a/vnt/src/handle/maintain/idle.rs b/vnt/src/handle/maintain/idle.rs
index e18203a..ddbd513 100644
--- a/vnt/src/handle/maintain/idle.rs
+++ b/vnt/src/handle/maintain/idle.rs
@@ -102,7 +102,7 @@ fn idle_route0<Call: VntCallback>(
             context.remove_route(&ip, route.route_key());
             if cur.is_gateway(&ip) {
                 //网关路由过期，则需要改变状态
-                context.change_status(current_device, ConnectStatus::Connecting);
+                crate::handle::change_status(current_device, ConnectStatus::Connecting);
                 call.error(ErrorInfo::new(ErrorType::Disconnect));
             }
             Duration::from_millis(100)
diff --git a/vnt/src/handle/mod.rs b/vnt/src/handle/mod.rs
index efb9c00..33d166d 100644
--- a/vnt/src/handle/mod.rs
+++ b/vnt/src/handle/mod.rs
@@ -1,3 +1,4 @@
+use crossbeam_utils::atomic::AtomicCell;
 use std::net::{Ipv4Addr, SocketAddr};
 
 pub mod callback;
@@ -201,3 +202,16 @@ impl CurrentDeviceInfo {
         &self.virtual_gateway == ip || ip == &GATEWAY_IP
     }
 }
+pub fn change_status(
+    current_device: &AtomicCell<CurrentDeviceInfo>,
+    connect_status: ConnectStatus,
+) -> CurrentDeviceInfo {
+    loop {
+        let cur = current_device.load();
+        let mut new_info = cur;
+        new_info.status = connect_status;
+        if current_device.compare_exchange(cur, new_info).is_ok() {
+            return new_info;
+        }
+    }
+}
diff --git a/vnt/src/handle/recv_data/client.rs b/vnt/src/handle/recv_data/client.rs
index 645093d..eb76625 100644
--- a/vnt/src/handle/recv_data/client.rs
+++ b/vnt/src/handle/recv_data/client.rs
@@ -139,7 +139,7 @@ impl ClientPacketHandler {
                         || real_dest == current_device.broadcast_ip
                         || real_dest.is_unspecified())
                 {
-                    if !self.route.allow(&ipv4.destination_ip()) {
+                    if !self.route.allow(&real_dest) {
                         //拦截不符合的目标
                         return Ok(());
                     }
diff --git a/vnt/src/handle/recv_data/server.rs b/vnt/src/handle/recv_data/server.rs
index 92a5f4c..55b68ca 100644
--- a/vnt/src/handle/recv_data/server.rs
+++ b/vnt/src/handle/recv_data/server.rs
@@ -398,7 +398,7 @@ impl<Call: VntCallback> ServerPacketHandler<Call> {
                 self.callback.error(err);
             }
             InErrorPacket::Disconnect => {
-                context.change_status(&self.current_device, ConnectStatus::Connecting);
+                crate::handle::change_status(&self.current_device, ConnectStatus::Connecting);
                 let err = ErrorInfo::new(ErrorType::Disconnect);
                 self.callback.error(err);
                 //掉线epoch要归零
diff --git a/vnt/src/handle/tun_tap/tun_handler.rs b/vnt/src/handle/tun_tap/tun_handler.rs
index 3e50361..05dd165 100644
--- a/vnt/src/handle/tun_tap/tun_handler.rs
+++ b/vnt/src/handle/tun_tap/tun_handler.rs
@@ -93,8 +93,8 @@ pub fn start(
             #[cfg(target_os = "macos")]
             {
                 let ip = current_device.load().virtual_ip;
-                if let Ok(udp) = std::net::UdpSocket::bind("0.0.0.0:0"){
-                    let _ = udp.send_to(b"stop",format!("{:?}:1234",ip));
+                if let Ok(udp) = std::net::UdpSocket::bind("0.0.0.0:0") {
+                    let _ = udp.send_to(b"stop", format!("{:?}:1234", ip));
                 }
             }
         })?