修改地址探测和转发路径

vnt/src/channel/context.rs

@@ -353,18 +353,18 @@ impl RouteTable {
         }
         Err(io::Error::new(io::ErrorKind::NotFound, "route not found"))
     }
-    pub fn add_route_if_absent(&self, id: Ipv4Addr, route: Route) {
+    pub fn add_route_if_absent(&self, id: Ipv4Addr, route: Route) -> bool {
         self.add_route_(id, route, true)
     }
-    pub fn add_route(&self, id: Ipv4Addr, route: Route) {
+    pub fn add_route(&self, id: Ipv4Addr, route: Route) -> bool {
         self.add_route_(id, route, false)
     }
-    fn add_route_(&self, id: Ipv4Addr, route: Route, only_if_absent: bool) {
+    fn add_route_(&self, id: Ipv4Addr, route: Route, only_if_absent: bool) -> bool {
         // 限制通道类型
         match self.use_channel_type {
             UseChannelType::P2p => {
                 if !route.is_p2p() {
-                    return;
+                    return false;
                 }
             }
             _ => {}
@@ -372,10 +372,18 @@ impl RouteTable {
         let key = route.route_key();
         if only_if_absent {
             if let Some((_, list)) = self.route_table.read().get(&id) {
+                let mut p2p_num = 0;
                 for (x, _) in list {
-                    if x.route_key() == key {
+                    if x.is_p2p() {
-                        return;
+                        p2p_num += 1;
                     }
+                    if x.route_key() == key {
+                        return true;
+                    }
+                }
+                if !self.first_latency && p2p_num >= self.channel_num {
+                    // 非优先延迟的情况下，通道满了则不用再添加
+                    return false;
                 }
             }
         }
@@ -387,11 +395,11 @@ impl RouteTable {
         for (x, time) in list.iter_mut() {
             if x.metric < route.metric && !self.first_latency {
                 //非优先延迟的情况下 不能比当前的路径更长
-                return;
+                return false;
             }
             if x.route_key() == key {
                 if only_if_absent {
-                    return;
+                    return true;
                 }
                 x.metric = route.metric;
                 x.rt = route.rt;
@@ -406,7 +414,7 @@ impl RouteTable {
             //如果延迟都稳定了，则去除多余通道
             for (route, _) in list.iter() {
                 if route.rt == DEFAULT_RT {
-                    return;
+                    return true;
                 }
             }
             //延迟优先模式需要更多的通道探测延迟最低的路线
@@ -422,6 +430,9 @@ impl RouteTable {
                     //非优先延迟的情况下 添加了直连的则排除非直连的
                     list.retain(|(k, _)| k.is_p2p());
                 }
+                if self.channel_num <= list.len() {
+                    return false;
+                }
             };
             //增加路由表容量，避免波动
             let limit_len = self.channel_num * 2;
@@ -429,6 +440,7 @@ impl RouteTable {
             self.truncate_(list, limit_len);
             list.push((route, AtomicCell::new(Instant::now())));
         }
+        return true;
     }
     fn truncate_(&self, list: &mut Vec<(Route, AtomicCell<Instant>)>, len: usize) {
         if list.len() <= len {

vnt/src/core/conn.rs

@@ -385,7 +385,6 @@ pub fn start<Call: VntCallback>(
             &scheduler,
             context.clone(),
             current_device.clone(),
-            server_cipher.clone(),
             nat_test.clone(),
             config_info.clone(),
         );

vnt/src/handle/maintain/addr_request.rs

@@ -5,48 +5,30 @@ use crossbeam_utils::atomic::AtomicCell;
 
 use crate::channel::context::ChannelContext;
 use crate::channel::punch::NatType;
-use crate::cipher::Cipher;
 use crate::handle::{BaseConfigInfo, CurrentDeviceInfo};
 use crate::nat::NatTest;
-use crate::protocol::body::ENCRYPTION_RESERVED;
-use crate::protocol::{control_packet, NetPacket, Protocol, MAX_TTL};
 use crate::util::Scheduler;
 
 pub fn addr_request(
     scheduler: &Scheduler,
     context: ChannelContext,
     current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
-    server_cipher: Cipher,
     nat_test: NatTest,
     _config: BaseConfigInfo,
 ) {
-    pub_address_request(
+    pub_address_request(scheduler, context, current_device_info.clone(), nat_test, 0);
-        scheduler,
-        context,
-        current_device_info.clone(),
-        server_cipher,
-        nat_test,
-        0,
-    );
 }
 
 fn pub_address_request(
     scheduler: &Scheduler,
     context: ChannelContext,
     current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
-    server_cipher: Cipher,
     nat_test: NatTest,
     count: usize,
 ) {
     let channel_num = context.channel_num();
     let index = count % channel_num;
-    if let Err(e) = addr_request0(
+    if let Err(e) = addr_request0(&context, &current_device_info, &nat_test, index) {
-        &context,
-        &current_device_info,
-        &server_cipher,
-        &nat_test,
-        index,
-    ) {
         log::warn!("{:?}", e);
     }
     let nat_info = nat_test.nat_info();
@@ -58,7 +40,7 @@ fn pub_address_request(
             if index == channel_num - 1 {
                 19
             } else {
-                7
+                9
             }
         }
     } else {
@@ -66,14 +48,7 @@ fn pub_address_request(
     };
 
     let rs = scheduler.timeout(Duration::from_secs(time), move |s| {
-        pub_address_request(
+        pub_address_request(s, context, current_device_info, nat_test, index + 1)
-            s,
-            context,
-            current_device_info,
-            server_cipher,
-            nat_test,
-            index + 1,
-        )
     });
     if !rs {
         log::info!("定时任务停止");
@@ -83,7 +58,6 @@ fn pub_address_request(
 fn addr_request0(
     context: &ChannelContext,
     current_device: &AtomicCell<CurrentDeviceInfo>,
-    server_cipher: &Cipher,
     nat_test: &NatTest,
     index: usize,
 ) -> anyhow::Result<()> {
@@ -91,24 +65,7 @@ fn addr_request0(
     if current_dev.status.offline() {
         return Ok(());
     }
-
+    let (data, addr) = nat_test.send_data()?;
-    if current_dev.connect_server.is_ipv4() && !context.main_protocol().is_base_tcp() {
+    context.send_main_udp(index, &data, addr)?;
-        // 如果连接的是ipv4服务，则探测公网端口
-        let gateway_ip = current_dev.virtual_gateway;
-        let src_ip = current_dev.virtual_ip;
-        let mut packet = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED]).unwrap();
-        packet.set_default_version();
-        packet.set_gateway_flag(true);
-        packet.set_protocol(Protocol::Control);
-        packet.set_transport_protocol(control_packet::Protocol::AddrRequest.into());
-        packet.first_set_ttl(MAX_TTL);
-        packet.set_source(src_ip);
-        packet.set_destination(gateway_ip);
-        server_cipher.encrypt_ipv4(&mut packet)?;
-        context.send_main_udp(index, packet.buffer(), current_dev.connect_server)?;
-    } else {
-        let (data, addr) = nat_test.send_data()?;
-        context.send_main_udp(index, &data, addr)?;
-    }
     Ok(())
 }

vnt/src/handle/maintain/punch.rs

@@ -233,7 +233,7 @@ fn punch0(
             || nat_info.public_ports.iter().filter(|&&v| v == 0).count()
                 > nat_info.public_ports.len() / 2)
     {
-        log::info!("公网地址为空，暂时放弃打洞,第{}轮", total_count);
+        log::info!("未获取到公网地址，暂时放弃打洞,第{}轮", total_count);
         return Ok(());
     }
     let current_ip = current_device.virtual_ip;

vnt/src/handle/maintain/re_nat_type.rs

@@ -44,7 +44,7 @@ fn retrieve_nat_type0(
                 };
                 #[cfg(feature = "upnp")]
                 nat_test.reset_upnp();
-                log::info!("刷新nat成功")
+                log::info!("刷新nat结束")
             }
         })
         .expect("natTest");

vnt/src/handle/recv_data/client.rs

@@ -212,14 +212,18 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
         let source = net_packet.source();
         match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
             ControlPacket::PingPacket(_) => {
-                net_packet.set_transport_protocol(control_packet::Protocol::Pong.into());
-                net_packet.set_source(current_device.virtual_ip);
-                net_packet.set_destination(source);
-                net_packet.first_set_ttl(MAX_TTL);
-                self.client_cipher.encrypt_ipv4(&mut net_packet)?;
-                context.send_by_key(&net_packet, route_key)?;
                 let route = Route::from_default_rt(route_key, metric);
-                context.route_table.add_route_if_absent(source, route);
+                if context.route_table.add_route_if_absent(source, route)
+                    || net_packet.source() < current_device.virtual_ip
+                {
+                    //在路由表中，或者来源比自己小，就需要回复，注意不能调换顺序
+                    net_packet.set_transport_protocol(control_packet::Protocol::Pong.into());
+                    net_packet.set_source(current_device.virtual_ip);
+                    net_packet.set_destination(source);
+                    net_packet.first_set_ttl(MAX_TTL);
+                    self.client_cipher.encrypt_ipv4(&mut net_packet)?;
+                    context.send_by_key(&net_packet, route_key)?;
+                }
             }
             ControlPacket::PongPacket(pong_packet) => {
                 let current_time = crate::handle::now_time() as u16;

vnt/src/nat/mod.rs

@@ -1,4 +1,4 @@
-use anyhow::Context;
+use anyhow::{anyhow, Context};
 use std::io;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, ToSocketAddrs};
 use std::net::{SocketAddr, UdpSocket};
@@ -139,17 +139,12 @@ impl Into<NatType> for PunchNatType {
 impl NatTest {
     pub fn new(
         _channel_num: usize,
-        mut stun_server: Vec<String>,
+        stun_server: Vec<String>,
         local_ipv4: Option<Ipv4Addr>,
         ipv6: Option<Ipv6Addr>,
         udp_ports: Vec<u16>,
         tcp_port: u16,
     ) -> NatTest {
-        if stun_server.len() > 5 {
-            stun_server.shuffle(&mut rand::thread_rng());
-            stun_server.truncate(5);
-            log::info!("stun_server truncate {:?}", stun_server);
-        }
         let ports = vec![0; udp_ports.len()];
         let nat_info = NatInfo::new(
             Vec::new(),
@@ -262,8 +257,17 @@ impl NatTest {
         &self,
         local_ipv4: Option<Ipv4Addr>,
         ipv6: Option<Ipv6Addr>,
-    ) -> io::Result<NatInfo> {
+    ) -> anyhow::Result<NatInfo> {
-        let (nat_type, public_ips, port_range) = stun::stun_test_nat(self.stun_server.clone())?;
+        let mut stun_server = self.stun_server.clone();
+        if stun_server.len() > 5 {
+            stun_server.shuffle(&mut rand::thread_rng());
+            stun_server.truncate(5);
+            log::info!("stun_server truncate {:?}", stun_server);
+        }
+        let (nat_type, public_ips, port_range) = stun::stun_test_nat(stun_server)?;
+        if public_ips.is_empty() {
+            Err(anyhow!("public_ips.is_empty"))?
+        }
         let mut guard = self.info.lock();
         guard.nat_type = nat_type;
         guard.public_ips = public_ips;