yeyinlaoying/vnt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[mio] 修复重连问题

Browse Source
This commit is contained in:
lubeilin
2024-03-11 21:59:13 +08:00
parent c63813e590
commit e78994ceb4
8 changed files with 84 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnt/src/channel/context.rs
+5 -24
View File
@@ -156,33 +156,14 @@ impl ContextInner {
pub fn change_status(
&self,
current_device: &AtomicCell<CurrentDeviceInfo>,
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;
}
}
}
vnt/src/core/conn.rs
+4
View File
@@ -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();
vnt/src/handle/handshaker.rs
+28
View File
@@ -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<AtomicCell<Instant>>,
}
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<NetPacket<Vec<u8>>> {
vnt/src/handle/maintain/heartbeat.rs
+7
View File
@@ -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<PeerDeviceInfo>)>,
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 {
vnt/src/handle/maintain/idle.rs
+13 -11
View File
@@ -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<Call: VntCallback>(
tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
call: Call,
mut connect_count: usize,
handshake: Handshake,
) {
idle_gateway0(
&context,
@@ -43,6 +45,7 @@ pub fn idle_gateway<Call: VntCallback>(
&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<Call: VntCallback>(
tcp_socket_sender,
call,
connect_count,
handshake,
)
});
if !rs {
@@ -66,6 +70,7 @@ fn idle_gateway0<Call: VntCallback>(
tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
call: &Call,
connect_count: &mut usize,
handshake: &Handshake,
) {
if let Err(e) = check_gateway_channel(
context,
@@ -74,6 +79,7 @@ fn idle_gateway0<Call: VntCallback>(
tcp_socket_sender,
call,
connect_count,
handshake,
) {
let cur = current_device.load();
call.error(ErrorInfo::new_msg(
@@ -96,10 +102,8 @@ fn idle_route0<Call: VntCallback>(
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<Call: VntCallback>(
tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
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(
&current_device.connect_server,
vnt/src/handle/maintain/punch.rs
+11 -13
View File
@@ -123,17 +123,16 @@ fn punch_request(
client_cipher: Cipher,
count: usize,
) {
if let Err(e) = punch0(
&context,
&nat_test,
&device_list,
&current_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<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
current_device: &Arc<AtomicCell<CurrentDeviceInfo>>,
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<PeerDeviceInfo> = device_list
vnt/src/handle/recv_data/mod.rs
+3
View File
@@ -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<Call: VntCallback> RecvDataHandler<Call> {
route: AllowExternalRoute,
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
counter: U64Adder,
handshake: Handshake,
) -> Self {
let server = ServerPacketHandler::new(
#[cfg(feature = "server_encrypt")]
@@ -78,6 +80,7 @@ impl<Call: VntCallback> RecvDataHandler<Call> {
nat_test.clone(),
callback,
external_route,
handshake,
);
let client = ClientPacketHandler::new(
device.clone(),
vnt/src/handle/recv_data/server.rs
+13 -4
View File
@@ -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<Call> {
up_key_time: Arc<AtomicCell<Instant>>,
route_record: Arc<Mutex<Vec<(Ipv4Addr, Ipv4Addr)>>>,
external_route: ExternalRoute,
handshake: Handshake,
}
impl<Call> ServerPacketHandler<Call> {
@@ -62,6 +66,7 @@ impl<Call> ServerPacketHandler<Call> {
nat_test: NatTest,
callback: Call,
external_route: ExternalRoute,
handshake: Handshake,
) -> Self {
Self {
#[cfg(feature = "server_encrypt")]
@@ -77,6 +82,7 @@ impl<Call> ServerPacketHandler<Call> {
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<Call: VntCallback> ServerPacketHandler<Call> {
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<Call: VntCallback> ServerPacketHandler<Call> {
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<Call: VntCallback> ServerPacketHandler<Call> {
}
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<Call: VntCallback> ServerPacketHandler<Call> {
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<Call: VntCallback> ServerPacketHandler<Call> {
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 => {