diff --git a/vnt/src/channel/punch.rs b/vnt/src/channel/punch.rs index b6ccdbc..0719205 100644 --- a/vnt/src/channel/punch.rs +++ b/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 = (min_port..max_port).collect(); nums.push(max_port); - { - let mut rng = rand::thread_rng(); - nums.shuffle(&mut rng); - } + nums.shuffle(&mut rand::thread_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(()) diff --git a/vnt/src/handle/maintain/mod.rs b/vnt/src/handle/maintain/mod.rs index b685fe4..62f2aab 100644 --- a/vnt/src/handle/maintain/mod.rs +++ b/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; diff --git a/vnt/src/handle/maintain/punch.rs b/vnt/src/handle/maintain/punch.rs index 23ee1d7..6413243 100644 --- a/vnt/src/handle/maintain/punch.rs +++ b/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)>>, current_device: Arc>, 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 = 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)))?; diff --git a/vnt/src/handle/recv_data/client.rs b/vnt/src/handle/recv_data/client.rs index 01c9063..645093d 100644 --- a/vnt/src/handle/recv_data/client.rs +++ b/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, client_cipher: Cipher, - punch_sender: SyncSender<(Ipv4Addr, NatInfo)>, + punch_sender: PunchSender, peer_nat_info_map: Arc>>, nat_test: NatTest, route: AllowExternalRoute, @@ -47,7 +47,7 @@ impl ClientPacketHandler { pub fn new( device: Arc, client_cipher: Cipher, - punch_sender: SyncSender<(Ipv4Addr, NatInfo)>, + punch_sender: PunchSender, peer_nat_info_map: Arc>>, 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); - context.route_table.add_route_if_absent(source, route); + // let route = Route::from_default_rt(route_key, metric); + // 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() - } } diff --git a/vnt/src/handle/recv_data/mod.rs b/vnt/src/handle/recv_data/mod.rs index 408c1fd..d620377 100644 --- a/vnt/src/handle/recv_data/mod.rs +++ b/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 RecvDataHandler { config_info: BaseConfigInfo, nat_test: NatTest, callback: Call, - punch_sender: SyncSender<(Ipv4Addr, NatInfo)>, + punch_sender: PunchSender, peer_nat_info_map: Arc>>, external_route: ExternalRoute, route: AllowExternalRoute,