通道设置为0

优化从linux或wsl系统获取id

.gitignore

@@ -0,0 +1,4 @@
+target/*
+vnt/src/proto/message.rs
+vnt-cli/src/generated_serial_number.rs
+Cargo.lock

common/src/identifier.rs

@@ -51,6 +51,13 @@ pub fn get_unique_identifier() -> Option<String> {
 #[cfg(target_os = "linux")]
 pub fn get_unique_identifier() -> Option<String> {
     use std::process::Command;
+
+    // 对 linux 或 wsl 来说，读取 /etc/machine-id 即可获取当前操作系统的
+    // 唯一标识，而且某些环境没有预装`dmidecode`命令
+    if let Ok(identifier) = std::fs::read_to_string("/etc/machine-id") {
+        return Some(identifier);
+    }
+
     let output = match Command::new("dmidecode")
         .arg("-s")
         .arg("system-uuid")

vnt-cli/Cargo.toml

@@ -12,7 +12,6 @@ getopts = "0.2.21"
 console = "0.15.2"
 os_info = "3.7.0"
 serde = "1.0"
-#serde_json = "1.0.94"
 serde_yaml = "0.9.32"
 log = "0.4.17"
 log4rs = "1.2.0"
@@ -41,4 +40,5 @@ server_encrypt=["vnt/server_encrypt"]
 ip_proxy=["vnt/ip_proxy"]
 [build-dependencies]
 embed-manifest = "1.4.0"
-rand = "0.9.0-alpha.0"
+rand = "0.9.0-alpha.0"
+chrono = "0.4.23"

vnt-cli/build.rs

@@ -3,11 +3,10 @@ use std::fs::File;
 use std::io::Write;
 
 fn main() {
-    // 生成随机序列号
+    let now_time = chrono::Local::now();
     let serial_number = format!(
-        "{}-{}-{}",
-        rand::thread_rng().gen_range(100..1000),
-        rand::thread_rng().gen_range(100..1000),
+        "{}-{}",
+        &now_time.format("%y%m%d%H%M").to_string(),
         rand::thread_rng().gen_range(100..1000)
     );
     let generated_code = format!(r#"pub const SERIAL_NUMBER: &str = "{}";"#, serial_number);

vnt-cli/src/main.rs

@@ -338,6 +338,11 @@ fn main() {
     };
     println!("version {}", vnt::VNT_VERSION);
     println!("Serial:{}", generated_serial_number::SERIAL_NUMBER);
+    log::info!(
+        "version:{},Serial:{}",
+        vnt::VNT_VERSION,
+        generated_serial_number::SERIAL_NUMBER
+    );
     main0(config, cmd);
     std::process::exit(0);
 }

vnt/src/channel/context.rs

@@ -289,15 +289,7 @@ impl ContextInner {
         }
     }
     pub fn remove_route(&self, ip: &Ipv4Addr, route_key: RouteKey) {
-        if self.route_table.remove_route(ip, route_key) {
-            if route_key.is_tcp {
-                if let Some(tcp) = self.tcp_map.write().remove(&route_key.addr) {
-                    if let Err(e) = tcp.shutdown() {
-                        log::warn!("{:?}", e);
-                    }
-                }
-            }
-        }
+        self.route_table.remove_route(ip, route_key)
     }
 }
 
@@ -501,18 +493,13 @@ impl RouteTable {
         }
         list
     }
-    pub fn remove_route(&self, id: &Ipv4Addr, route_key: RouteKey) -> bool {
+    pub fn remove_route(&self, id: &Ipv4Addr, route_key: RouteKey) {
         let mut write_guard = self.route_table.write();
         if let Some((_, routes)) = write_guard.get_mut(id) {
             routes.retain(|(x, _)| x.route_key() != route_key);
             if routes.is_empty() {
                 write_guard.remove(id);
-                true
-            } else {
-                false
             }
-        } else {
-            return true;
         }
     }
     /// 更新路由入栈包的时刻，长时间没有收到数据的路由将会被剔除

vnt/src/channel/mod.rs

@@ -179,9 +179,6 @@ pub fn init_context(
             )
         };
 
-        io_convert(socket.set_reuse_address(true), |_| {
-            format!("set_reuse_address failed: {}", &address)
-        })?;
         io_convert(socket.set_send_buffer_size(2 * 1024 * 1024), |_| {
             format!("set_send_buffer_size failed: {}", &address)
         })?;
@@ -192,7 +189,6 @@ pub fn init_context(
             format!("bind failed: {}", &address)
         })?;
         let main_channel: UdpSocket = socket.into();
-        main_channel.set_nonblocking(true)?;
         udps.push(main_channel);
     }
     let context = Context::new(
@@ -220,9 +216,6 @@ pub fn init_context(
         (socket, address)
     };
 
-    io_convert(socket.set_reuse_address(true), |_| {
-        format!("set_reuse_address failed: {}", &address)
-    })?;
     if let Err(e) = socket.bind(&address.into()) {
         if ports[0] == 0 {
             //端口可能冲突，则使用任意端口

vnt/src/channel/punch.rs

@@ -100,6 +100,9 @@ impl NatInfo {
     pub fn update_addr(&mut self, index: usize, ip: Ipv4Addr, port: u16) {
         if port != 0 {
             if let Some(public_port) = self.public_ports.get_mut(index) {
+                if *public_port != port {
+                    log::info!("端口变化={}:{}", ip, port)
+                }
                 *public_port = port;
             }
         }
@@ -111,6 +114,7 @@ impl NatInfo {
         {
             if !self.public_ips.contains(&ip) {
                 self.public_ips.push(ip);
+                log::info!("ip变化={},{:?}", ip, self.public_ips)
             }
         }
     }
@@ -208,7 +212,7 @@ impl Punch {
 impl Punch {
     fn connect_tcp(&self, buf: &[u8], addr: SocketAddr) -> bool {
         // mio是非阻塞的，不能立马判断是否能连接成功，所以用标准库的tcp
-        match std::net::TcpStream::connect_timeout(&addr, Duration::from_secs(3)) {
+        match std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(100)) {
             Ok(tcp_stream) => {
                 if tcp_stream.set_nonblocking(true).is_err() {
                     return false;
@@ -224,29 +228,35 @@ impl Punch {
         }
         false
     }
-    pub fn punch(&mut self, buf: &[u8], id: Ipv4Addr, nat_info: NatInfo) -> io::Result<()> {
+    pub fn punch(
+        &mut self,
+        buf: &[u8],
+        id: Ipv4Addr,
+        nat_info: NatInfo,
+        punch_tcp: bool,
+    ) -> io::Result<()> {
         if !self.context.route_table.need_punch(&id) {
             log::info!("已打洞成功,无需打洞:{:?}", id);
             return Ok(());
         }
-        if self.is_tcp && nat_info.tcp_port != 0 {
+        if punch_tcp && self.is_tcp && nat_info.tcp_port != 0 {
             //向tcp发起连接
             if let Some(ipv6_addr) = nat_info.local_tcp_ipv6addr() {
                 if self.connect_tcp(buf, ipv6_addr) {
-                    return Ok(());
+                    // return Ok(());
                 }
             }
             //向tcp发起连接
             if let Some(ipv4_addr) = nat_info.local_tcp_ipv4addr() {
                 if self.connect_tcp(buf, ipv4_addr) {
-                    return Ok(());
+                    // return Ok(());
                 }
             }
             if nat_info.nat_type == NatType::Cone && nat_info.public_ips.len() == 1 {
                 let addr =
                     SocketAddr::V4(SocketAddrV4::new(nat_info.public_ips[0], nat_info.tcp_port));
                 if self.connect_tcp(buf, addr) {
-                    return Ok(());
+                    // return Ok(());
                 }
             }
         }

vnt/src/channel/udp_channel.rs

@@ -1,6 +1,5 @@
 use std::collections::HashMap;
 use std::sync::mpsc::{sync_channel, Receiver};
-use std::sync::Arc;
 use std::{io, thread};
 
 use mio::event::Source;
@@ -135,7 +134,74 @@ where
     }
 }
 
-/// 阻塞监听
+// /// 阻塞
+// fn main_udp_listen<H>(
+//     stop_manager: StopManager,
+//     recv_handler: H,
+//     context: Context,
+// ) -> io::Result<()>
+// where
+//     H: RecvChannelHandler,
+// {
+//     for index in 0..context.main_udp_socket.len() {
+//         let stop_manager = stop_manager.clone();
+//         let context = context.clone();
+//         let recv_handler = recv_handler.clone();
+//         thread::Builder::new()
+//             .name(format!("mainUdp{}", index))
+//             .spawn(move || {
+//                 if let Err(e) = main_udp_listen0(stop_manager, index, recv_handler, context) {
+//                     log::error!("{:?}", e);
+//                 }
+//             })?;
+//     }
+//     Ok(())
+// }
+//
+// pub fn main_udp_listen0<H>(
+//     stop_manager: StopManager,
+//     index: usize,
+//     mut recv_handler: H,
+//     context: Context,
+// ) -> io::Result<()>
+// where
+//     H: RecvChannelHandler,
+// {
+//     use std::time::Duration;
+//     let udp_socket = &context.main_udp_socket[index];
+//     udp_socket.set_read_timeout(Some(Duration::from_secs(5)))?;
+//     udp_socket.set_write_timeout(Some(Duration::from_secs(1)))?;
+//     let local_addr = udp_socket.local_addr()?;
+//     let worker = stop_manager.add_listener(format!("main_udp_{}", index), move || {
+//         if let Ok(udp) = std::net::UdpSocket::bind("0.0.0.0:0") {
+//             let _ = udp.send_to(b"stop", format!("127.0.0.1:{}", local_addr.port()));
+//         }
+//     })?;
+//
+//     let mut buf = [0; BUFFER_SIZE];
+//     loop {
+//         match udp_socket.recv_from(&mut buf) {
+//             Ok((len, addr)) => {
+//                 if &buf[..len] == b"stop" {
+//                     if stop_manager.is_stop() {
+//                         break;
+//                     }
+//                 }
+//                 recv_handler.handle(&mut buf[..len], RouteKey::new(false, index, addr), &context);
+//             }
+//             Err(e) => {
+//                 if stop_manager.is_stop() {
+//                     break;
+//                 }
+//                 log::error!("index={},{:?},{}", index, udp_socket.local_addr(), e)
+//             }
+//         }
+//     }
+//     worker.stop_all();
+//     Ok(())
+// }
+
+/// 非阻塞
 fn main_udp_listen<H>(
     stop_manager: StopManager,
     recv_handler: H,
@@ -144,6 +210,7 @@ fn main_udp_listen<H>(
 where
     H: RecvChannelHandler,
 {
+    use std::sync::Arc;
     let poll = Poll::new()?;
     let waker = Arc::new(Waker::new(poll.registry(), NOTIFY)?);
     let _waker = waker.clone();

vnt/src/core/conn.rs

@@ -190,6 +190,7 @@ impl Vnt {
             server_cipher.clone(),
             config.parallel,
             up_counter,
+            device_list.clone(),
         )?;
         maintain::idle_gateway(
             &scheduler,

vnt/src/handle/maintain/addr_request.rs

@@ -31,8 +31,8 @@ pub fn pub_address_request(
     server_cipher: Cipher,
 ) {
     addr_request0(&context, &current_device_info, &server_cipher);
-    // 17秒发送一次
-    let rs = scheduler.timeout(Duration::from_secs(17), |s| {
+    // 9秒发送一次
+    let rs = scheduler.timeout(Duration::from_secs(9), |s| {
         pub_address_request(s, context, current_device_info, server_cipher)
     });
     if !rs {

vnt/src/handle/maintain/punch.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::net::Ipv4Addr;
 use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
 use std::sync::Arc;
@@ -26,6 +27,7 @@ pub struct PunchSender {
     sender_cone_self: SyncSender<(Ipv4Addr, NatInfo)>,
     sender_cone_peer: SyncSender<(Ipv4Addr, NatInfo)>,
 }
+
 impl PunchSender {
     pub fn send(&self, src_peer: bool, ip: Ipv4Addr, info: NatInfo) -> bool {
         log::info!(
@@ -53,17 +55,19 @@ impl PunchSender {
         sender.try_send((ip, info)).is_ok()
     }
 }
+
 pub struct PunchReceiver {
     receiver_peer: Receiver<(Ipv4Addr, NatInfo)>,
     receiver_self: Receiver<(Ipv4Addr, NatInfo)>,
     receiver_cone_peer: Receiver<(Ipv4Addr, NatInfo)>,
     receiver_cone_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);
-    let (sender_cone_peer, receiver_cone_peer) = sync_channel(1);
-    let (sender_cone_self, receiver_cone_self) = sync_channel(1);
+    let (sender_self, receiver_self) = sync_channel(0);
+    let (sender_peer, receiver_peer) = sync_channel(0);
+    let (sender_cone_peer, receiver_cone_peer) = sync_channel(0);
+    let (sender_cone_self, receiver_cone_self) = sync_channel(0);
     (
         PunchSender {
             sender_self,
@@ -90,6 +94,8 @@ pub fn punch(
     receiver: PunchReceiver,
     punch: Punch,
 ) {
+    let punch_record = Arc::new(Mutex::new(HashMap::new()));
+    let last_punch_record = HashMap::new();
     punch_request(
         scheduler,
         context,
@@ -98,15 +104,18 @@ pub fn punch(
         current_device.clone(),
         client_cipher.clone(),
         0,
+        punch_record.clone(),
+        last_punch_record,
     );
     let f = |receiver: Receiver<(Ipv4Addr, NatInfo)>| {
         let punch = punch.clone();
         let current_device = current_device.clone();
         let client_cipher = client_cipher.clone();
+        let punch_record = punch_record.clone();
         thread::Builder::new()
             .name("punch".into())
             .spawn(move || {
-                punch_start(receiver, punch, current_device, client_cipher);
+                punch_start(receiver, punch, current_device, client_cipher, punch_record);
             })
             .expect("punch");
     };
@@ -122,6 +131,7 @@ fn punch_start(
     mut punch: Punch,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
     client_cipher: Cipher,
+    punch_record: Arc<Mutex<HashMap<Ipv4Addr, usize>>>,
 ) {
     while let Ok((peer_ip, nat_info)) = receiver.recv() {
         let mut packet = NetPacket::new_encrypt([0u8; 12 + ENCRYPTION_RESERVED]).unwrap();
@@ -131,12 +141,23 @@ fn punch_start(
         packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into());
         packet.set_source(current_device.load().virtual_ip());
         packet.set_destination(peer_ip);
-        log::info!("发起打洞，目标:{:?},{:?}", peer_ip, nat_info);
+        let count = {
+            let mut guard = punch_record.lock();
+            if let Some(v) = guard.get_mut(&peer_ip) {
+                *v += 1;
+                *v
+            } else {
+                guard.insert(peer_ip, 1);
+                0
+            }
+        };
+        log::info!("第{}次发起打洞,目标:{:?},{:?} ", count, peer_ip, nat_info);
+
         if let Err(e) = client_cipher.encrypt_ipv4(&mut packet) {
             log::error!("{:?}", e);
             continue;
         }
-        if let Err(e) = punch.punch(packet.buffer(), peer_ip, nat_info) {
+        if let Err(e) = punch.punch(packet.buffer(), peer_ip, nat_info, count < 2) {
             log::warn!("{:?}", e)
         }
     }
@@ -151,16 +172,27 @@ fn punch_request(
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
     client_cipher: Cipher,
     count: usize,
+    punch_record: Arc<Mutex<HashMap<Ipv4Addr, usize>>>,
+    mut last_punch_record: HashMap<Ipv4Addr, usize>,
 ) {
     let curr = current_device.load();
     let secs = if curr.status.online() {
-        if let Err(e) = punch0(&context, &nat_test, &device_list, curr, &client_cipher) {
+        if let Err(e) = punch0(
+            &context,
+            &nat_test,
+            &device_list,
+            curr,
+            &client_cipher,
+            &punch_record,
+            &mut last_punch_record,
+            count,
+        ) {
             log::warn!("{:?}", e)
         }
-        let sleep_time = [3, 5, 7, 11, 13, 17, 19, 23, 29];
+        let sleep_time = [5, 6, 7];
         Duration::from_secs(sleep_time[count % sleep_time.len()])
     } else {
-        Duration::from_secs(3)
+        Duration::from_secs(5)
     };
     let rs = scheduler.timeout(secs, move |s| {
         punch_request(
@@ -171,6 +203,8 @@ fn punch_request(
             current_device,
             client_cipher,
             count + 1,
+            punch_record,
+            last_punch_record,
         );
     });
     if !rs {
@@ -185,8 +219,19 @@ fn punch0(
     device_list: &Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
     current_device: CurrentDeviceInfo,
     client_cipher: &Cipher,
+    punch_record: &Mutex<HashMap<Ipv4Addr, usize>>,
+    last_punch_record: &mut HashMap<Ipv4Addr, usize>,
+    total_count: usize,
 ) -> io::Result<()> {
     let nat_info = nat_test.nat_info();
+    if total_count < 10
+        && (nat_info.public_ips.is_empty()
+            || nat_info.public_ports.is_empty()
+            || nat_info.public_ports[0] == 0)
+    {
+        log::info!("公网地址为空，暂时放弃打洞,第{}轮", total_count);
+        return Ok(());
+    }
     let current_ip = current_device.virtual_ip;
     let mut list: Vec<PeerDeviceInfo> = device_list
         .lock()
@@ -196,43 +241,40 @@ fn punch0(
         .cloned()
         .collect();
     list.shuffle(&mut rand::thread_rng());
-    let mut count = 0;
-    // // 优先没打洞的 need_punch会过滤掉已经打洞成功的
-    // 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;
-        }
-        if info.virtual_ip <= current_device.virtual_ip {
-            continue;
-        }
         if !context.route_table.need_punch(&info.virtual_ip) {
+            punch_record.lock().remove(&info.virtual_ip);
             continue;
         }
-        count += 1;
-        if count > 2 {
+        // 能发起打洞的前提是自己空闲，这里会间隔5秒以上发起一次打洞，所以假定上一轮打洞已结束
+        let punch_count = punch_record
+            .lock()
+            .get(&info.virtual_ip)
+            .cloned()
+            .unwrap_or(0);
+        let last_punch = last_punch_record
+            .get(&info.virtual_ip)
+            .cloned()
+            .unwrap_or(0);
+        // 梯度减少打洞频率
+        if total_count > last_punch + punch_count.min(35) {
+            last_punch_record.insert(info.virtual_ip, total_count);
+            let packet = punch_packet(
+                client_cipher,
+                current_device.virtual_ip(),
+                &nat_info,
+                info.virtual_ip,
+            )?;
+            log::info!(
+                "目标:{:?},当前nat:{:?} 第{}次发起打洞协商请求， 第:{}轮",
+                info.virtual_ip,
+                nat_info,
+                punch_count,
+                total_count,
+            );
+            context.send_default(packet.buffer(), current_device.connect_server)?;
             break;
         }
-        let packet = punch_packet(
-            client_cipher,
-            current_device.virtual_ip(),
-            &nat_info,
-            info.virtual_ip,
-        )?;
-        log::info!(
-            "发起打洞协商请求，目标:{:?},{:?}",
-            info.virtual_ip,
-            nat_info
-        );
-        context.send_default(packet.buffer(), current_device.connect_server)?;
     }
     Ok(())
 }

vnt/src/handle/recv_data/server.rs

@@ -237,6 +237,10 @@ impl<Call: VntCallback> ServerPacketHandler<Call> {
                     context
                         .route_table
                         .add_route_if_absent(virtual_gateway, route);
+                    let public_ip = response.public_ip.into();
+                    let public_port = response.public_port as u16;
+                    self.nat_test
+                        .update_addr(route_key.index(), public_ip, public_port);
                     let old = current_device;
                     let mut cur = *current_device;
                     loop {
@@ -245,7 +249,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;
+                        new_current_device.status = ConnectStatus::Connected;
                         if let Err(c) = self
                             .current_device
                             .compare_exchange(cur, new_current_device)
@@ -256,10 +260,6 @@ impl<Call: VntCallback> ServerPacketHandler<Call> {
                         }
                     }
 
-                    let public_ip = response.public_ip.into();
-                    let public_port = response.public_port as u16;
-                    self.nat_test
-                        .update_addr(route_key.index(), public_ip, public_port);
                     if old.virtual_ip != virtual_ip
                         || old.virtual_gateway != virtual_gateway
                         || old.virtual_netmask != virtual_netmask

vnt/src/handle/tun_tap/mod.rs

@@ -1,13 +1,15 @@
 use std::io;
 use std::net::Ipv4Addr;
 
-use crate::channel::context::Context;
+use parking_lot::Mutex;
+
 use packet::ip::ipv4::packet::IpV4Packet;
 use packet::ip::ipv4::protocol::Protocol;
 
+use crate::channel::context::Context;
 use crate::cipher::Cipher;
 use crate::external_route::ExternalRoute;
-use crate::handle::{check_dest, CurrentDeviceInfo};
+use crate::handle::{check_dest, CurrentDeviceInfo, PeerDeviceInfo};
 #[cfg(feature = "ip_proxy")]
 use crate::ip_proxy::{IpProxyMap, ProxyHandler};
 use crate::protocol;
@@ -23,57 +25,82 @@ fn broadcast(
     sender: &Context,
     net_packet: &mut NetPacket<&mut [u8]>,
     current_device: &CurrentDeviceInfo,
+    device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
 ) -> io::Result<()> {
-    let mut peer_ips = Vec::with_capacity(8);
-    let vec = sender.route_table.route_table_one();
-    let mut relay_count = 0;
+    let list: Vec<Ipv4Addr> = device_list
+        .lock()
+        .1
+        .iter()
+        .filter(|info| info.status.is_online())
+        .map(|info| info.virtual_ip)
+        .collect();
     const MAX_COUNT: usize = 8;
-    for (peer_ip, route) in vec {
-        if peer_ip == current_device.virtual_gateway {
-            continue;
-        }
-        if peer_ips.len() == MAX_COUNT {
-            relay_count += 1;
+    let mut p2p_ips = Vec::with_capacity(8);
+    let mut relay_ips = Vec::with_capacity(8);
+    let mut overflow = false;
+    for (index, peer_ip) in list.into_iter().enumerate() {
+        if index > MAX_COUNT {
+            overflow = true;
             break;
         }
-        if route.is_p2p()
-            && sender
+        if let Some(route) = sender.route_table.route_one_p2p(&peer_ip) {
+            if sender
                 .send_by_key(net_packet.buffer(), route.route_key())
                 .is_ok()
-        {
-            peer_ips.push(peer_ip);
-        } else {
-            relay_count += 1;
+            {
+                p2p_ips.push(peer_ip);
+                continue;
+            }
         }
+        relay_ips.push(peer_ip);
     }
-    if (relay_count == 0 && !peer_ips.is_empty()) || current_device.status.offline() {
-        //不需要转发
+    if !overflow && relay_ips.is_empty() {
+        //全部p2p,不需要服务器中转
         return Ok(());
     }
-    //转发到服务端的可选择广播，还要进行服务端加密
-    if peer_ips.is_empty() {
-        sender.send_default(net_packet.buffer(), current_device.connect_server)?;
-    } else {
-        let buf =
-            vec![0u8; 12 + 1 + peer_ips.len() * 4 + net_packet.data_len() + ENCRYPTION_RESERVED];
-        //剩余的发送到服务端，需要告知哪些已发送过
-        let mut server_packet = NetPacket::new_encrypt(buf)?;
-        server_packet.set_version(Version::V1);
-        server_packet.set_gateway_flag(true);
-        server_packet.first_set_ttl(MAX_TTL);
-        server_packet.set_source(net_packet.source());
-        //使用对应的目的地址
-        server_packet.set_destination(net_packet.destination());
-        server_packet.set_protocol(protocol::Protocol::IpTurn);
-        server_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4Broadcast.into());
 
-        let mut broadcast = BroadcastPacket::unchecked(server_packet.payload_mut());
-        broadcast.set_address(&peer_ips)?;
-        broadcast.set_data(net_packet.buffer())?;
-        server_cipher.encrypt_ipv4(&mut server_packet)?;
-        sender.send_default(server_packet.buffer(), current_device.connect_server)?;
+    if p2p_ips.is_empty() {
+        //都没有p2p则直接由服务器转发
+        if current_device.status.online() {
+            sender.send_default(net_packet.buffer(), current_device.connect_server)?;
+        }
+        return Ok(());
     }
-    Ok(())
+    if !overflow && relay_ips.len() == 2 {
+        // 如果转发的ip数不多就直接发
+        for peer_ip in relay_ips {
+            //非直连的广播要改变目的地址，不然服务端收到了会再次广播
+            net_packet.set_destination(peer_ip);
+            sender.send_ipv4_by_id(
+                net_packet.buffer(),
+                &peer_ip,
+                current_device.connect_server,
+                current_device.status.online(),
+            )?;
+        }
+        return Ok(());
+    }
+    if current_device.status.offline() {
+        //离线的不再转发
+        return Ok(());
+    }
+    let buf = vec![0u8; 12 + 1 + p2p_ips.len() * 4 + net_packet.data_len() + ENCRYPTION_RESERVED];
+    //剩余的发送到服务端，需要告知哪些已发送过
+    let mut server_packet = NetPacket::new_encrypt(buf)?;
+    server_packet.set_version(Version::V1);
+    server_packet.set_gateway_flag(true);
+    server_packet.first_set_ttl(MAX_TTL);
+    server_packet.set_source(net_packet.source());
+    //使用对应的目的地址
+    server_packet.set_destination(net_packet.destination());
+    server_packet.set_protocol(protocol::Protocol::IpTurn);
+    server_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4Broadcast.into());
+
+    let mut broadcast = BroadcastPacket::unchecked(server_packet.payload_mut());
+    broadcast.set_address(&p2p_ips)?;
+    broadcast.set_data(net_packet.buffer())?;
+    server_cipher.encrypt_ipv4(&mut server_packet)?;
+    sender.send_default(server_packet.buffer(), current_device.connect_server)
 }
 
 /// 实现一个原地发送，必须保证是如下结构
@@ -89,6 +116,7 @@ pub fn base_handle(
     #[cfg(feature = "ip_proxy")] proxy_map: &Option<IpProxyMap>,
     client_cipher: &Cipher,
     server_cipher: &Cipher,
+    device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
 ) -> io::Result<()> {
     let ipv4_packet = IpV4Packet::new(&buf[12..data_len])?;
     let protocol = ipv4_packet.protocol();
@@ -118,7 +146,13 @@ pub fn base_handle(
     if dest_ip.is_broadcast() || current_device.broadcast_ip == dest_ip {
         // 广播 发送到直连目标
         client_cipher.encrypt_ipv4(&mut net_packet)?;
-        broadcast(server_cipher, context, &mut net_packet, &current_device)?;
+        broadcast(
+            server_cipher,
+            context,
+            &mut net_packet,
+            &current_device,
+            device_list,
+        )?;
         return Ok(());
     }
     if !check_dest(

vnt/src/handle/tun_tap/tun_handler.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::{io, thread};
 
 use crossbeam_utils::atomic::AtomicCell;
+use parking_lot::Mutex;
 
 use packet::icmp::icmp::IcmpPacket;
 use packet::icmp::Kind;
@@ -14,7 +15,7 @@ use crate::channel::context::Context;
 use crate::cipher::Cipher;
 use crate::external_route::ExternalRoute;
 use crate::handle::tun_tap::channel_group::{channel_group, GroupSyncSender};
-use crate::handle::CurrentDeviceInfo;
+use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
 #[cfg(feature = "ip_proxy")]
 use crate::ip_proxy::IpProxyMap;
 use crate::util::{SingleU64Adder, StopManager};
@@ -46,6 +47,7 @@ fn handle(
     #[cfg(feature = "ip_proxy")] proxy_map: &Option<IpProxyMap>,
     client_cipher: &Cipher,
     server_cipher: &Cipher,
+    device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
 ) -> io::Result<()> {
     //忽略掉结构不对的情况（ipv6数据、win tap会读到空数据），不然日志打印太多了
     let ipv4_packet = match IpV4Packet::new(&mut data[12..len]) {
@@ -67,6 +69,7 @@ fn handle(
         proxy_map,
         client_cipher,
         server_cipher,
+        device_list,
     );
 }
 
@@ -81,6 +84,7 @@ pub fn start(
     server_cipher: Cipher,
     parallel: usize,
     mut up_counter: SingleU64Adder,
+    device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
 ) -> io::Result<()> {
     let worker = {
         #[cfg(target_os = "macos")]
@@ -110,6 +114,7 @@ pub fn start(
             let ip_proxy_map = ip_proxy_map.clone();
             let client_cipher = client_cipher.clone();
             let server_cipher = server_cipher.clone();
+            let device_list = device_list.clone();
             thread::Builder::new()
                 .name(format!("tunHandler-{}", index))
                 .spawn(move || {
@@ -129,6 +134,7 @@ pub fn start(
                             &ip_proxy_map,
                             &client_cipher,
                             &server_cipher,
+                            &device_list,
                         ) {
                             Ok(_) => {}
                             Err(e) => {
@@ -161,6 +167,7 @@ pub fn start(
                     client_cipher,
                     server_cipher,
                     &mut up_counter,
+                    device_list,
                 ) {
                     log::warn!("stop:{}", e);
                 }
@@ -180,6 +187,7 @@ fn start_simple(
     client_cipher: Cipher,
     server_cipher: Cipher,
     up_counter: &mut SingleU64Adder,
+    device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
 ) -> io::Result<()> {
     let mut buf = [0; 1024 * 16];
     loop {
@@ -204,6 +212,7 @@ fn start_simple(
             &ip_proxy_map,
             &client_cipher,
             &server_cipher,
+            &device_list,
         ) {
             Ok(_) => {}
             Err(e) => {

vnt/src/nat/mod.rs

@@ -78,7 +78,7 @@ impl Into<NatType> for PunchNatType {
 
 impl NatTest {
     pub fn new(
-        channel_num: usize,
+        _channel_num: usize,
         mut stun_server: Vec<String>,
         local_ipv4: Option<Ipv4Addr>,
         ipv6: Option<Ipv6Addr>,
@@ -87,8 +87,7 @@ impl NatTest {
     ) -> NatTest {
         let server = stun_server[0].clone();
         stun_server.resize(3, server);
-        let mut ports = udp_ports.clone();
-        ports.resize(channel_num, 0);
+        let ports = vec![0; udp_ports.len()];
         let nat_info = NatInfo::new(
             Vec::new(),
             ports,