From e78994ceb49572fd9c96a0ac30fc21a3505b48af Mon Sep 17 00:00:00 2001 From: lubeilin <1791778603@qq.com> Date: Mon, 11 Mar 2024 21:59:13 +0800 Subject: [PATCH] =?UTF-8?q?[mio]=20=E4=BF=AE=E5=A4=8D=E9=87=8D=E8=BF=9E?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vnt/src/channel/context.rs | 29 +++++----------------------- vnt/src/core/conn.rs | 4 ++++ vnt/src/handle/handshaker.rs | 28 +++++++++++++++++++++++++++ vnt/src/handle/maintain/heartbeat.rs | 7 +++++++ vnt/src/handle/maintain/idle.rs | 24 ++++++++++++----------- vnt/src/handle/maintain/punch.rs | 24 +++++++++++------------ vnt/src/handle/recv_data/mod.rs | 3 +++ vnt/src/handle/recv_data/server.rs | 17 ++++++++++++---- 8 files changed, 84 insertions(+), 52 deletions(-) diff --git a/vnt/src/channel/context.rs b/vnt/src/channel/context.rs index 99b2d53..b7f78c6 100644 --- a/vnt/src/channel/context.rs +++ b/vnt/src/channel/context.rs @@ -156,33 +156,14 @@ impl ContextInner { pub fn change_status( &self, current_device: &AtomicCell, + connect_status: ConnectStatus, ) -> CurrentDeviceInfo { - let mut cur = current_device.load(); loop { - let status = if self.route_table.route_one(&cur.virtual_gateway).is_some() { - //已连接 - if cur.status.online() { - return cur; - } - //状态变为已连接 - ConnectStatus::Connected - } else { - //未连接 - if cur.status.offline() { - return cur; - } - //状态变为未连接 - ConnectStatus::Connecting - }; + let cur = current_device.load(); let mut new_info = cur; - new_info.status = status; - match current_device.compare_exchange(cur, new_info) { - Ok(_) => { - return new_info; - } - Err(c) => { - cur = c; - } + new_info.status = connect_status; + if current_device.compare_exchange(cur, new_info).is_ok() { + return new_info; } } } diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs index c492b4a..00806b6 100644 --- a/vnt/src/core/conn.rs +++ b/vnt/src/core/conn.rs @@ -19,6 +19,7 @@ use crate::cipher::Cipher; use crate::cipher::RsaCipher; use crate::core::Config; use crate::external_route::{AllowExternalRoute, ExternalRoute}; +use crate::handle::handshaker::Handshake; use crate::handle::maintain::PunchReceiver; use crate::handle::recv_data::RecvDataHandler; use crate::handle::{ @@ -143,6 +144,7 @@ impl Vnt { let down_counter = U64Adder::with_capacity(config.ports.as_ref().map(|v| v.len()).unwrap_or_default() + 8); let down_count_watcher = down_counter.watch(); + let handshake = Handshake::new(); let handler = RecvDataHandler::new( #[cfg(feature = "server_encrypt")] rsa_cipher, @@ -161,6 +163,7 @@ impl Vnt { #[cfg(feature = "ip_proxy")] proxy_map.clone(), down_counter, + handshake.clone(), ); //初始化网络数据通道 @@ -196,6 +199,7 @@ impl Vnt { tcp_socket_sender.clone(), callback.clone(), 0, + handshake, ); { let context = context.clone(); diff --git a/vnt/src/handle/handshaker.rs b/vnt/src/handle/handshaker.rs index 62658f5..092a1a2 100644 --- a/vnt/src/handle/handshaker.rs +++ b/vnt/src/handle/handshaker.rs @@ -1,7 +1,12 @@ use std::io; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use crossbeam_utils::atomic::AtomicCell; use protobuf::Message; +use crate::channel::context::Context; #[cfg(feature = "server_encrypt")] use crate::cipher::RsaCipher; use crate::handle::{GATEWAY_IP, SELF_IP}; @@ -19,6 +24,29 @@ pub enum HandshakeEnum { ServerError(String), Other(String), } +#[derive(Clone)] +pub struct Handshake { + time: Arc>, +} +impl Handshake { + pub fn new() -> Self { + Handshake { + time: Arc::new(AtomicCell::new(Instant::now() - Duration::from_secs(60))), + } + } + pub fn send(&self, context: &Context, secret: bool, addr: SocketAddr) -> io::Result<()> { + let last = self.time.load(); + //短时间不重复发送 + if last.elapsed() < Duration::from_secs(2) { + return Ok(()); + } + let request_packet = handshake_request_packet(secret)?; + log::info!("发送握手请求,secret={},{:?}", secret, addr); + context.send_default(request_packet.buffer(), addr)?; + self.time.store(Instant::now()); + Ok(()) + } +} /// 第一次握手数据 pub fn handshake_request_packet(secret: bool) -> io::Result>> { diff --git a/vnt/src/handle/maintain/heartbeat.rs b/vnt/src/handle/maintain/heartbeat.rs index 23500c7..82be33d 100644 --- a/vnt/src/handle/maintain/heartbeat.rs +++ b/vnt/src/handle/maintain/heartbeat.rs @@ -102,6 +102,9 @@ fn heartbeat0( if current_device.is_gateway(&peer.virtual_ip) { continue; } + if current_device.status.offline() { + continue; + } if context.route_table.route_one(&peer.virtual_ip).is_none() { //路由为空，则向服务端地址发送 let net_packet = match heartbeat_packet_client(client_cipher, src_ip, peer.virtual_ip) { @@ -165,6 +168,10 @@ fn client_relay0( device_list: &Mutex<(u16, Vec)>, client_cipher: &Cipher, ) -> io::Result<()> { + // 离线了不再探测 + if current_device.status.offline() { + return Ok(()); + } let peer_list = { device_list.lock().1.clone() }; let mut routes = context.route_table.route_table_p2p(); for peer in &peer_list { diff --git a/vnt/src/handle/maintain/idle.rs b/vnt/src/handle/maintain/idle.rs index f59ff28..e18203a 100644 --- a/vnt/src/handle/maintain/idle.rs +++ b/vnt/src/handle/maintain/idle.rs @@ -2,7 +2,8 @@ use crate::channel::context::Context; use crate::channel::idle::{Idle, IdleType}; use crate::channel::sender::AcceptSocketSender; use crate::handle::callback::{ConnectInfo, ErrorType}; -use crate::handle::{handshaker, BaseConfigInfo, CurrentDeviceInfo}; +use crate::handle::handshaker::Handshake; +use crate::handle::{handshaker, BaseConfigInfo, ConnectStatus, CurrentDeviceInfo}; use crate::util::Scheduler; use crate::{ErrorInfo, VntCallback}; use crossbeam_utils::atomic::AtomicCell; @@ -35,6 +36,7 @@ pub fn idle_gateway( tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option>)>, call: Call, mut connect_count: usize, + handshake: Handshake, ) { idle_gateway0( &context, @@ -43,6 +45,7 @@ pub fn idle_gateway( &tcp_socket_sender, &call, &mut connect_count, + &handshake, ); let rs = scheduler.timeout(Duration::from_secs(5), move |s| { idle_gateway( @@ -53,6 +56,7 @@ pub fn idle_gateway( tcp_socket_sender, call, connect_count, + handshake, ) }); if !rs { @@ -66,6 +70,7 @@ fn idle_gateway0( tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option>)>, call: &Call, connect_count: &mut usize, + handshake: &Handshake, ) { if let Err(e) = check_gateway_channel( context, @@ -74,6 +79,7 @@ fn idle_gateway0( tcp_socket_sender, call, connect_count, + handshake, ) { let cur = current_device.load(); call.error(ErrorInfo::new_msg( @@ -96,10 +102,8 @@ fn idle_route0( context.remove_route(&ip, route.route_key()); if cur.is_gateway(&ip) { //网关路由过期，则需要改变状态 - let cur = context.change_status(current_device); - if cur.status.offline() { - call.error(ErrorInfo::new(ErrorType::Disconnect)); - } + context.change_status(current_device, ConnectStatus::Connecting); + call.error(ErrorInfo::new(ErrorType::Disconnect)); } Duration::from_millis(100) } @@ -115,21 +119,19 @@ fn check_gateway_channel( tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option>)>, call: &Call, count: &mut usize, + handshake: &Handshake, ) -> io::Result<()> { - let current_device = context.change_status(current_device); + let current_device = current_device.load(); if current_device.status.offline() { *count += 1; - if *count % 4 == 0 { - context.change_main_index(); - } //需要重连 call.connect(ConnectInfo::new(*count, current_device.connect_server)); - let request_packet = handshaker::handshake_request_packet(config.client_secret)?; log::info!("发送握手请求,{:?}", config); - if let Err(e) = context.send_default(request_packet.buffer(), current_device.connect_server) + if let Err(e) = handshake.send(context, config.client_secret, current_device.connect_server) { log::warn!("{:?}", e); if context.is_main_tcp() { + let request_packet = handshaker::handshake_request_packet(config.client_secret)?; //tcp需要重连 let tcp_stream = std::net::TcpStream::connect_timeout( ¤t_device.connect_server, diff --git a/vnt/src/handle/maintain/punch.rs b/vnt/src/handle/maintain/punch.rs index 6413243..3663eb2 100644 --- a/vnt/src/handle/maintain/punch.rs +++ b/vnt/src/handle/maintain/punch.rs @@ -123,17 +123,16 @@ fn punch_request( client_cipher: Cipher, count: usize, ) { - if let Err(e) = punch0( - &context, - &nat_test, - &device_list, - ¤t_device, - &client_cipher, - ) { - log::warn!("{:?}", e) - } - let sleep_time = [3, 5, 7, 11, 13, 17, 19, 23, 29]; - let secs = Duration::from_secs(sleep_time[count % sleep_time.len()]); + let curr = current_device.load(); + let secs = if curr.status.online() { + if let Err(e) = punch0(&context, &nat_test, &device_list, curr, &client_cipher) { + log::warn!("{:?}", e) + } + let sleep_time = [3, 5, 7, 11, 13, 17, 19, 23, 29]; + Duration::from_secs(sleep_time[count % sleep_time.len()]) + } else { + Duration::from_secs(3) + }; let rs = scheduler.timeout(secs, move |s| { punch_request( s, @@ -155,10 +154,9 @@ fn punch0( context: &Context, nat_test: &NatTest, device_list: &Arc)>>, - current_device: &Arc>, + current_device: CurrentDeviceInfo, client_cipher: &Cipher, ) -> io::Result<()> { - let current_device = current_device.load(); let nat_info = nat_test.nat_info(); let current_ip = current_device.virtual_ip; let mut list: Vec = device_list diff --git a/vnt/src/handle/recv_data/mod.rs b/vnt/src/handle/recv_data/mod.rs index d620377..fa9730d 100644 --- a/vnt/src/handle/recv_data/mod.rs +++ b/vnt/src/handle/recv_data/mod.rs @@ -17,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::handshaker::Handshake; use crate::handle::maintain::PunchSender; use crate::handle::recv_data::client::ClientPacketHandler; use crate::handle::recv_data::server::ServerPacketHandler; @@ -66,6 +67,7 @@ impl RecvDataHandler { route: AllowExternalRoute, #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, counter: U64Adder, + handshake: Handshake, ) -> Self { let server = ServerPacketHandler::new( #[cfg(feature = "server_encrypt")] @@ -78,6 +80,7 @@ impl RecvDataHandler { nat_test.clone(), callback, external_route, + handshake, ); let client = ClientPacketHandler::new( device.clone(), diff --git a/vnt/src/handle/recv_data/server.rs b/vnt/src/handle/recv_data/server.rs index 906281a..92a5f4c 100644 --- a/vnt/src/handle/recv_data/server.rs +++ b/vnt/src/handle/recv_data/server.rs @@ -23,8 +23,11 @@ use crate::external_route::ExternalRoute; use crate::handle::callback::{ErrorInfo, ErrorType, HandshakeInfo, RegisterInfo, VntCallback}; #[cfg(feature = "server_encrypt")] use crate::handle::handshaker; +use crate::handle::handshaker::Handshake; use crate::handle::recv_data::PacketHandler; -use crate::handle::{registrar, BaseConfigInfo, CurrentDeviceInfo, PeerDeviceInfo, GATEWAY_IP}; +use crate::handle::{ + registrar, BaseConfigInfo, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo, GATEWAY_IP, +}; use crate::nat::NatTest; use crate::proto; use crate::proto::message::{DeviceList, HandshakeResponse, RegistrationResponse}; @@ -49,6 +52,7 @@ pub struct ServerPacketHandler { up_key_time: Arc>, route_record: Arc>>, external_route: ExternalRoute, + handshake: Handshake, } impl ServerPacketHandler { @@ -62,6 +66,7 @@ impl ServerPacketHandler { nat_test: NatTest, callback: Call, external_route: ExternalRoute, + handshake: Handshake, ) -> Self { Self { #[cfg(feature = "server_encrypt")] @@ -77,6 +82,7 @@ impl ServerPacketHandler { up_key_time: Arc::new(AtomicCell::new(Instant::now() - Duration::from_secs(60))), route_record: Arc::new(Mutex::default()), external_route, + handshake, } } } @@ -239,6 +245,7 @@ impl ServerPacketHandler { new_current_device.virtual_ip = virtual_ip; new_current_device.virtual_netmask = virtual_netmask; new_current_device.virtual_gateway = virtual_gateway; + new_current_device.status = crate::handle::ConnectStatus::Connected; if let Err(c) = self .current_device .compare_exchange(cur, new_current_device) @@ -248,7 +255,6 @@ impl ServerPacketHandler { break; } } - let _ = context.change_status(&self.current_device); let public_ip = response.public_ip.into(); let public_port = response.public_port as u16; @@ -380,10 +386,10 @@ impl ServerPacketHandler { } fn error( &self, - _context: &Context, + context: &Context, _current_device: &CurrentDeviceInfo, net_packet: NetPacket<&mut [u8]>, - _route_key: RouteKey, + route_key: RouteKey, ) -> io::Result<()> { match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload())? { InErrorPacket::TokenError => { @@ -392,6 +398,7 @@ impl ServerPacketHandler { self.callback.error(err); } InErrorPacket::Disconnect => { + context.change_status(&self.current_device, ConnectStatus::Connecting); let err = ErrorInfo::new(ErrorType::Disconnect); self.callback.error(err); //掉线epoch要归零 @@ -400,6 +407,8 @@ impl ServerPacketHandler { dev.0 = 0; drop(dev); } + self.handshake + .send(context, self.config_info.client_secret, route_key.addr)?; // self.register(current_device, context, route_key)?; } InErrorPacket::AddressExhausted => {