[mio] 调整打洞处理

vnt/src/channel/punch.rs

@@ -1,11 +1,12 @@
 use std::collections::HashMap;
-use std::io;
 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6};
 use std::str::FromStr;
 use std::time::Duration;
+use std::{io, thread};
 
 use mio::net::TcpStream;
 use rand::prelude::SliceRandom;
+use rand::Rng;
 
 use crate::channel::context::Context;
 use crate::channel::sender::AcceptSocketSender;
@@ -277,7 +278,7 @@ impl Punch {
                 //预测范围内最多发送max_k1个包
                 let max_k1 = 60;
                 //全局最多发送max_k2个包
-                let max_k2 = 800;
+                let max_k2 = rand::thread_rng().gen_range(600..800);
                 let port = nat_info.public_ports.get(0).map(|e| *e).unwrap_or(0);
                 if nat_info.public_port_range < max_k1 * 3 {
                     //端口变化不大时，在预测的范围内随机发送
@@ -295,10 +296,7 @@ impl Punch {
                     };
                     let mut nums: Vec<u16> = (min_port..max_port).collect();
                     nums.push(max_port);
-                    {
+                    nums.shuffle(&mut rand::thread_rng());
-                        let mut rng = rand::thread_rng();
-                        nums.shuffle(&mut rng);
-                    }
                     self.punch_symmetric(&nums[..k], buf, &nat_info.public_ips, max_k1 as usize)?;
                 }
                 let start = *self.port_index.entry(id.clone()).or_insert(0);
@@ -326,11 +324,13 @@ impl Punch {
                             continue;
                         }
                         let addr = SocketAddr::V4(SocketAddrV4::new(*ip, port));
-                        self.context.send_main_udp(index, buf, addr)?;
+                        if is_cone {
-                        if !is_cone {
+                            self.context.send_main_udp(index, buf, addr)?;
+                        } else {
                             //只有一方是对称，则对称方要使用全部端口发送数据，符合上述计算的概率
                             self.context.try_send_all(buf, addr);
                         }
+                        thread::sleep(Duration::from_millis(2));
                     }
                     if !is_cone {
                         //对称网络数据只发一遍
@@ -350,6 +350,11 @@ impl Punch {
         max: usize,
     ) -> io::Result<()> {
         let mut count = 0;
+        let index = if self.context.channel_num() == 1 {
+            0
+        } else {
+            rand::thread_rng().gen_range(0..self.context.channel_num())
+        };
         for port in ports {
             for pub_ip in ips {
                 count += 1;
@@ -357,7 +362,8 @@ impl Punch {
                     return Ok(());
                 }
                 let addr = SocketAddr::V4(SocketAddrV4::new(*pub_ip, *port));
-                self.context.send_main_udp(0, buf, addr)?;
+                self.context.send_main_udp(index, buf, addr)?;
+                thread::sleep(Duration::from_millis(2));
             }
         }
         Ok(())

vnt/src/handle/maintain/mod.rs

@@ -9,7 +9,7 @@ mod addr_request;
 pub use addr_request::addr_request;
 
 mod punch;
-pub use punch::punch;
+pub use punch::*;
 
 mod idle;
 pub use idle::idle_gateway;

vnt/src/handle/maintain/punch.rs

@@ -1,5 +1,6 @@
+use std::cmp::Ordering;
 use std::net::Ipv4Addr;
-use std::sync::mpsc::Receiver;
+use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
 use std::sync::Arc;
 use std::time::Duration;
 use std::{io, thread};
@@ -19,6 +20,39 @@ use crate::protocol::body::ENCRYPTION_RESERVED;
 use crate::protocol::{control_packet, other_turn_packet, NetPacket, Protocol, Version, MAX_TTL};
 use crate::util::Scheduler;
 
+#[derive(Clone)]
+pub struct PunchSender {
+    sender_self: SyncSender<(Ipv4Addr, NatInfo)>,
+    sender_peer: SyncSender<(Ipv4Addr, NatInfo)>,
+}
+impl PunchSender {
+    pub fn send(&self, src_peer: bool, ip: Ipv4Addr, info: NatInfo) -> bool {
+        if src_peer {
+            self.sender_peer.send((ip, info)).is_ok()
+        } else {
+            self.sender_self.send((ip, info)).is_ok()
+        }
+    }
+}
+pub struct PunchReceiver {
+    receiver_peer: Receiver<(Ipv4Addr, NatInfo)>,
+    receiver_self: Receiver<(Ipv4Addr, NatInfo)>,
+}
+pub fn punch_channel() -> (PunchSender, PunchReceiver) {
+    let (sender_self, receiver_self) = sync_channel(1);
+    let (sender_peer, receiver_peer) = sync_channel(1);
+    (
+        PunchSender {
+            sender_self,
+            sender_peer,
+        },
+        PunchReceiver {
+            receiver_peer,
+            receiver_self,
+        },
+    )
+}
+
 pub fn punch(
     scheduler: &Scheduler,
     context: Context,
@@ -26,7 +60,7 @@ pub fn punch(
     device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
     client_cipher: Cipher,
-    receiver: Receiver<(Ipv4Addr, NatInfo)>,
+    receiver: PunchReceiver,
     punch: Punch,
 ) {
     punch_request(
@@ -38,8 +72,18 @@ pub fn punch(
         client_cipher.clone(),
         0,
     );
+    let receiver_peer = receiver.receiver_peer;
+    let receiver_self = receiver.receiver_self;
+    {
+        let punch = punch.clone();
+        let current_device = current_device.clone();
+        let client_cipher = client_cipher.clone();
+        thread::spawn(move || {
+            punch_start(receiver_peer, punch, current_device, client_cipher);
+        });
+    }
     thread::spawn(move || {
-        punch_start(receiver, punch, current_device, client_cipher);
+        punch_start(receiver_self, punch, current_device, client_cipher);
     });
 }
 
@@ -116,9 +160,26 @@ fn punch0(
 ) -> io::Result<()> {
     let current_device = current_device.load();
     let nat_info = nat_test.nat_info();
-    let mut list = device_list.lock().clone().1;
+    let current_ip = current_device.virtual_ip;
+    let mut list: Vec<PeerDeviceInfo> = device_list
+        .lock()
+        .1
+        .iter()
+        .filter(|info| info.status.is_online() && info.virtual_ip > current_ip)
+        .cloned()
+        .collect();
     list.shuffle(&mut rand::thread_rng());
     let mut count = 0;
+    // 优先没打洞的
+    list.sort_by(|v1, v2| {
+        if context.route_table.route_one_p2p(&v1.virtual_ip).is_none() {
+            Ordering::Less
+        } else if context.route_table.route_one_p2p(&v2.virtual_ip).is_none() {
+            Ordering::Greater
+        } else {
+            Ordering::Equal
+        }
+    });
     for info in list {
         if !info.status.is_online() {
             continue;
@@ -169,6 +230,7 @@ fn punch_packet(
         punch_reply.ipv6 = ipv6.octets().to_vec();
     }
     punch_reply.nat_type = protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type));
+    log::info!("请求打洞={:?}", punch_reply);
     let bytes = punch_reply
         .write_to_bytes()
         .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("punch_packet {:?}", e)))?;

vnt/src/handle/recv_data/client.rs

@@ -1,7 +1,6 @@
 use std::collections::HashMap;
 use std::io;
 use std::net::{Ipv4Addr, Ipv6Addr};
-use std::sync::mpsc::SyncSender;
 use std::sync::Arc;
 
 use parking_lot::RwLock;
@@ -18,6 +17,7 @@ use crate::channel::punch::NatInfo;
 use crate::channel::{Route, RouteKey};
 use crate::cipher::Cipher;
 use crate::external_route::AllowExternalRoute;
+use crate::handle::maintain::PunchSender;
 use crate::handle::recv_data::PacketHandler;
 use crate::handle::CurrentDeviceInfo;
 #[cfg(feature = "ip_proxy")]
@@ -35,7 +35,7 @@ use crate::protocol::{
 pub struct ClientPacketHandler {
     device: Arc<Device>,
     client_cipher: Cipher,
-    punch_sender: SyncSender<(Ipv4Addr, NatInfo)>,
+    punch_sender: PunchSender,
     peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
     nat_test: NatTest,
     route: AllowExternalRoute,
@@ -47,7 +47,7 @@ impl ClientPacketHandler {
     pub fn new(
         device: Arc<Device>,
         client_cipher: Cipher,
-        punch_sender: SyncSender<(Ipv4Addr, NatInfo)>,
+        punch_sender: PunchSender,
         peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
         nat_test: NatTest,
         route: AllowExternalRoute,
@@ -75,6 +75,9 @@ impl PacketHandler for ClientPacketHandler {
         current_device: &CurrentDeviceInfo,
     ) -> io::Result<()> {
         self.client_cipher.decrypt_ipv4(&mut net_packet)?;
+        context
+            .route_table
+            .update_read_time(&net_packet.source(), &route_key);
         match net_packet.protocol() {
             Protocol::Service => {}
             Protocol::Error => {}
@@ -165,7 +168,6 @@ impl ClientPacketHandler {
     ) -> io::Result<()> {
         let metric = net_packet.source_ttl() - net_packet.ttl() + 1;
         let source = net_packet.source();
-        context.route_table.update_read_time(&source, &route_key);
         match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
             ControlPacket::PingPacket(_) => {
                 net_packet.set_transport_protocol(control_packet::Protocol::Pong.into());
@@ -174,8 +176,8 @@ impl ClientPacketHandler {
                 net_packet.first_set_ttl(MAX_TTL);
                 self.client_cipher.encrypt_ipv4(&mut net_packet)?;
                 context.send_by_key(net_packet.buffer(), route_key)?;
-                let route = Route::from_default_rt(route_key, metric);
+                // let route = Route::from_default_rt(route_key, metric);
-                context.route_table.add_route_if_absent(source, route);
+                // context.route_table.add_route_if_absent(source, route);
             }
             ControlPacket::PongPacket(pong_packet) => {
                 let current_time = crate::handle::now_time() as u16;
@@ -187,6 +189,7 @@ impl ClientPacketHandler {
                 context.route_table.add_route(source, route);
             }
             ControlPacket::PunchRequest => {
+                log::info!("PunchRequest={:?},source={}", route_key, source);
                 if context.use_channel_type().is_only_relay() {
                     return Ok(());
                 }
@@ -201,6 +204,7 @@ impl ClientPacketHandler {
                 context.route_table.add_route_if_absent(source, route);
             }
             ControlPacket::PunchResponse => {
+                log::info!("PunchResponse={:?},source={}", route_key, source);
                 if context.use_channel_type().is_only_relay() {
                     return Ok(());
                 }
@@ -316,12 +320,14 @@ impl ClientPacketHandler {
                     punch_packet.set_source(current_device.virtual_ip());
                     punch_packet.set_destination(source);
                     punch_packet.set_payload(&bytes)?;
-                    if self.punch(source, peer_nat_info) {
+                    log::info!("接收打洞请求={:?}", peer_nat_info);
+                    if self.punch_sender.send(true, source, peer_nat_info) {
                         self.client_cipher.encrypt_ipv4(&mut punch_packet)?;
                         context.send_by_key(punch_packet.buffer(), route_key)?;
                     }
                 } else {
-                    self.punch(source, peer_nat_info);
+                    log::info!("接收打洞请求回复={:?}", peer_nat_info);
+                    self.punch_sender.send(false, source, peer_nat_info);
                 }
             }
             other_turn_packet::Protocol::Unknown(e) => {
@@ -330,7 +336,4 @@ impl ClientPacketHandler {
         }
         Ok(())
     }
-    fn punch(&self, peer_ip: Ipv4Addr, peer_nat_info: NatInfo) -> bool {
-        self.punch_sender.try_send((peer_ip, peer_nat_info)).is_ok()
-    }
 }

vnt/src/handle/recv_data/mod.rs

@@ -1,6 +1,5 @@
 use std::collections::HashMap;
 use std::net::Ipv4Addr;
-use std::sync::mpsc::SyncSender;
 use std::sync::Arc;
 use std::{io, thread};
 
@@ -18,6 +17,7 @@ use crate::cipher::Cipher;
 use crate::cipher::RsaCipher;
 use crate::external_route::{AllowExternalRoute, ExternalRoute};
 use crate::handle::callback::VntCallback;
+use crate::handle::maintain::PunchSender;
 use crate::handle::recv_data::client::ClientPacketHandler;
 use crate::handle::recv_data::server::ServerPacketHandler;
 use crate::handle::recv_data::turn::TurnPacketHandler;
@@ -60,7 +60,7 @@ impl<Call: VntCallback> RecvDataHandler<Call> {
         config_info: BaseConfigInfo,
         nat_test: NatTest,
         callback: Call,
-        punch_sender: SyncSender<(Ipv4Addr, NatInfo)>,
+        punch_sender: PunchSender,
         peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
         external_route: ExternalRoute,
         route: AllowExternalRoute,