[mio] 调整代码

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())

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)?;

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)

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;
+        }
+    }
+}

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(());
                     }

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要归零

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));
                 }
             }
         })?