修改rand版本

优化从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.8.5"
+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-jni/java/top/wherewego/vnt/jni/CallBack.java

@@ -8,6 +8,10 @@ import top.wherewego.vnt.jni.param.*;
  * @author https://github.com/lbl8603/vnt
  */
 public interface CallBack {
+    /**
+     * 连接成功的回调
+     */
+    void success();
     /**
      * 创建虚拟网卡成功的回调方法
      *

vnt-jni/java/top/wherewego/vnt/jni/Config.java

@@ -90,6 +90,18 @@ public class Config {
      * 虚拟网卡fd 仅在android上支持
      */
     private int deviceFd;
+    /**
+     * enum: relay/p2p/all
+     */
+    private String useChannel;
+    /**
+     * 模拟丢包率，取0~1之间的数，为null表示不丢包，1表示全部丢包
+     */
+    private Double packetLossRate;
+    /**
+     * 模拟延迟 单位毫秒(ms)
+     */
+    private Integer packetDelay;
 
     public Config() {
     }
@@ -261,4 +273,28 @@ public class Config {
     public void setDeviceFd(int deviceFd) {
         this.deviceFd = deviceFd;
     }
+
+    public String getUseChannel() {
+        return useChannel;
+    }
+
+    public void setUseChannel(String useChannel) {
+        this.useChannel = useChannel;
+    }
+
+    public Double getPacketLossRate() {
+        return packetLossRate;
+    }
+
+    public void setPacketLossRate(Double packetLossRate) {
+        this.packetLossRate = packetLossRate;
+    }
+
+    public Integer getPacketDelay() {
+        return packetDelay;
+    }
+
+    public void setPacketDelay(Integer packetDelay) {
+        this.packetDelay = packetDelay;
+    }
 }

vnt-jni/java/top/wherewego/vnt/jni/IpUtils.java

@@ -1,8 +1,9 @@
 package top.wherewego.vnt.jni;
 
 /**
- * @author lubeilin
- * @date: 2024/02/27 18:31
+ * ip转换
+ *
+ * @author https://github.com/lbl8603/vnt
  */
 public class IpUtils {
     public static String intToIpAddress(int ipAddress) {

vnt-jni/java/top/wherewego/vnt/jni/Vnt.java

@@ -11,11 +11,8 @@ import java.io.IOException;
 public class Vnt implements Closeable {
     private final long raw;
 
-    public Vnt(Config config, CallBack callBack) {
+    public Vnt(Config config, CallBack callBack) throws Exception{
         this.raw = new0(config, callBack);
-        if(this.raw == 0){
-            throw new RuntimeException();
-        }
     }
 
     public void stop() {
@@ -30,7 +27,7 @@ public class Vnt implements Closeable {
         return list0(raw);
     }
 
-    private native long new0(Config config, CallBack callBack);
+    private native long new0(Config config, CallBack callBack) throws Exception;
 
     private native void stop0(long raw);
 

vnt-jni/src/callback.rs

@@ -26,6 +26,11 @@ impl CallBack {
 }
 
 impl CallBack {
+    fn success0(&self) -> jni::errors::Result<()> {
+        let env = &mut self.jvm.attach_current_thread()? as &mut JNIEnv;
+        env.call_method(&self.this, "success", "()V", &[])?;
+        Ok(())
+    }
     fn create_tun0(&self, info: DeviceInfo) -> jni::errors::Result<()> {
         let env = &mut self.jvm.attach_current_thread()? as &mut JNIEnv;
         let param = env.new_object(
@@ -139,13 +144,16 @@ impl CallBack {
     }
     fn stop0(&self) -> jni::errors::Result<()> {
         let env = &mut self.jvm.attach_current_thread()? as &mut JNIEnv;
-        env.call_method(&self.this, "error", "()V", &[])?;
+        env.call_method(&self.this, "stop", "()V", &[])?;
         Ok(())
     }
 }
 
 impl VntCallback for CallBack {
     fn success(&self) {
+        if let Err(e) = self.success0() {
+            log::warn!("success {:?}", e);
+        }
     }
     fn create_tun(&self, info: DeviceInfo) {
         if let Err(e) = self.create_tun0(info) {

vnt-jni/src/config.rs

@@ -29,6 +29,11 @@ pub fn new_config(env: &mut JNIEnv, config: JObject) -> Result<Config, Error> {
     let use_channel = to_string(env, &config, "useChannel")?;
     let finger = env.get_field(&config, "finger", "Z")?.z()?;
     let first_latency = env.get_field(&config, "firstLatency", "Z")?.z()?;
+    let packet_delay = to_integer(env, &config, "packetDelay")?
+        .map(|v| v as u32)
+        .unwrap_or_default();
+    let packet_loss_rate = to_double(env, &config, "packetLossRate")?;
+
     let in_ips = to_string_array(env, &config, "inIps")?;
     let out_ips = to_string_array(env, &config, "outIps")?;
     let ports =
@@ -132,8 +137,8 @@ pub fn new_config(env: &mut JNIEnv, config: JObject) -> Result<Config, Error> {
         #[cfg(target_os = "android")]
         device_fd,
         UseChannelType::from_str(&use_channel.unwrap_or_default()).unwrap_or_default(),
-        None,
-        0,
+        packet_loss_rate,
+        packet_delay,
     ) {
         Ok(config) => config,
         Err(e) => {

vnt-jni/src/utils.rs

@@ -119,3 +119,13 @@ pub fn to_integer(env: &mut JNIEnv, config: &JObject, name: &str) -> Result<Opti
         env.call_method(value, "intValue", "()I", &[])?.i()? as _
     ));
 }
+pub fn to_double(env: &mut JNIEnv, config: &JObject, name: &str) -> Result<Option<f64>, Error> {
+    let value = env.get_field(config, name, "Ljava/lang/Double;")?.l()?;
+    if value.is_null() {
+        return Ok(None);
+    }
+    // 调用 intValue
+    return Ok(Some(
+        env.call_method(value, "doubleValue", "()D", &[])?.d()? as _,
+    ));
+}

vnt/proto/message.proto

@@ -1,66 +1,78 @@
 syntax = "proto3";
-message HandshakeRequest{
-  string version = 1;
-  bool secret = 2;
+
+message HandshakeRequest {
+    string version = 1;
+    bool secret = 2;
 }
-message HandshakeResponse{
-  string version = 1;
-  bool secret = 2;
-  bytes public_key = 3;
-  string key_finger = 4;
+message HandshakeResponse {
+    string version = 1;
+    bool secret = 2;
+    bytes public_key = 3;
+    string key_finger = 4;
 }
-message SecretHandshakeRequest{
-  string token = 1;
-  bytes key = 2;
+message SecretHandshakeRequest {
+    string token = 1;
+    bytes key = 2;
 }
-message RegistrationRequest{
-  string token = 1;
-  string device_id = 2;
-  string name = 3;
-  bool is_fast = 4;
-  string version = 5;
-  fixed32 virtual_ip = 6;
-  bool allow_ip_change = 7;
-  bool client_secret = 8;
+message RegistrationRequest {
+    string token = 1;
+    string device_id = 2;
+    string name = 3;
+    bool is_fast = 4;
+    string version = 5;
+    fixed32 virtual_ip = 6;
+    bool allow_ip_change = 7;
+    bool client_secret = 8;
 }
 
-message RegistrationResponse{
-  fixed32 virtual_ip = 1;
-  fixed32 virtual_gateway = 2;
-  fixed32 virtual_netmask = 3;
-  uint32 epoch = 4;
-  repeated DeviceInfo device_info_list = 5;
-  fixed32 public_ip = 6;
-  uint32 public_port = 7;
-  bytes public_ipv6 = 8;
+message RegistrationResponse {
+    fixed32 virtual_ip = 1;
+    fixed32 virtual_gateway = 2;
+    fixed32 virtual_netmask = 3;
+    uint32 epoch = 4;
+    repeated DeviceInfo device_info_list = 5;
+    fixed32 public_ip = 6;
+    uint32 public_port = 7;
+    bytes public_ipv6 = 8;
 }
-message DeviceInfo{
-  string name = 1;
-  fixed32 virtual_ip = 2;
-  uint32 device_status = 3;
-  bool client_secret = 4;
+message DeviceInfo {
+    string name = 1;
+    fixed32 virtual_ip = 2;
+    uint32 device_status = 3;
+    bool client_secret = 4;
 }
 
-message DeviceList{
-  uint32 epoch = 1;
-  repeated DeviceInfo device_info_list = 2;
+message DeviceList {
+    uint32 epoch = 1;
+    repeated DeviceInfo device_info_list = 2;
 }
 
-message PunchInfo{
-  repeated fixed32 public_ip_list = 2;
-  uint32 public_port = 3;
-  uint32 public_port_range = 4;
-  PunchNatType nat_type = 5;
-  bool reply = 6;
-  fixed32 local_ip = 7;
-  uint32 local_port = 8;
-  bytes ipv6 = 9;
-  uint32 ipv6_port = 10;
-  uint32 tcp_port = 11;
-  repeated uint32 udp_ports = 12;
-  repeated uint32 public_ports = 13;
+message PunchInfo {
+    repeated fixed32 public_ip_list = 2;
+    uint32 public_port = 3;
+    uint32 public_port_range = 4;
+    PunchNatType nat_type = 5;
+    bool reply = 6;
+    fixed32 local_ip = 7;
+    uint32 local_port = 8;
+    bytes ipv6 = 9;
+    uint32 ipv6_port = 10;
+    uint32 tcp_port = 11;
+    repeated uint32 udp_ports = 12;
+    repeated uint32 public_ports = 13;
 }
-enum PunchNatType{
-  Symmetric = 0;
-  Cone = 1;
+enum PunchNatType {
+    Symmetric = 0;
+    Cone = 1;
+}
+/// 向服务器上报客户端状态信息
+message ClientStatusInfo {
+    fixed32 source = 1;
+    repeated RouteItem p2p_list = 2;
+    uint64 up_stream = 3;
+    uint64 down_stream = 4;
+    PunchNatType nat_type = 5;
+}
+message RouteItem {
+    fixed32 next_ip = 1;
 }

vnt/src/channel/context.rs

@@ -225,6 +225,7 @@ impl ContextInner {
         buf: &[u8],
         id: &Ipv4Addr,
         server_addr: SocketAddr,
+        send_default: bool,
     ) -> io::Result<()> {
         if self.packet_loss_rate > 0 {
             if rand::thread_rng().gen_ratio(self.packet_loss_rate, PACKET_LOSS_RATE_DENOMINATOR) {
@@ -239,7 +240,7 @@ impl ContextInner {
             if e.kind() != io::ErrorKind::NotFound {
                 log::warn!("{}:{:?}", id, e);
             }
-            if !self.route_table.use_channel_type.is_only_p2p() {
+            if !self.route_table.use_channel_type.is_only_p2p() && send_default {
                 //符合条件再发到服务器转发
                 self.send_default(buf, server_addr)?;
             }
@@ -288,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)
     }
 }
 
@@ -460,14 +453,19 @@ impl RouteTable {
         }
         None
     }
-    pub fn need_punch(&self, id: &Ipv4Addr) -> bool {
+    pub fn no_need_punch(&self, id: &Ipv4Addr) -> bool {
         if let Some((_, v)) = self.route_table.read().get(id) {
-            //存在p2p的通道则不再打洞
-            if v.iter().filter(|(k, _)| k.is_p2p()).count() >= 1 {
-                return false;
-            }
+            //p2p的通道数符合要求
+            return v.iter().filter(|(k, _)| k.is_p2p()).count() >= self.channel_num;
+        }
+        false
+    }
+    pub fn p2p_num(&self, id: &Ipv4Addr) -> usize {
+        if let Some((_, v)) = self.route_table.read().get(id) {
+            v.iter().filter(|(k, _)| k.is_p2p()).count()
+        } else {
+            0
         }
-        true
     }
     /// 返回所有路由
     pub fn route_table(&self) -> Vec<(Ipv4Addr, Vec<Route>)> {
@@ -481,9 +479,10 @@ impl RouteTable {
         let table = self.route_table.read();
         let mut list = Vec::with_capacity(8);
         for (ip, (_, routes)) in table.iter() {
-            if let Some((route, _)) = routes.first() {
+            for (route, _) in routes.iter() {
                 if route.is_p2p() {
                     list.push((*ip, *route));
+                    break;
                 }
             }
         }
@@ -499,18 +498,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<()> {
-        if !self.context.route_table.need_punch(&id) {
+    pub fn punch(
+        &mut self,
+        buf: &[u8],
+        id: Ipv4Addr,
+        nat_info: NatInfo,
+        punch_tcp: bool,
+    ) -> io::Result<()> {
+        if self.context.route_table.no_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();
@@ -188,8 +255,14 @@ where
                 NOTIFY => return Ok(()),
                 Token(index) => index - 1,
             };
+            let udp = if let Some(udp) = udps.get(index) {
+                udp
+            } else {
+                log::error!("{:?}", x);
+                continue;
+            };
             loop {
-                match udps[index].recv_from(&mut buf) {
+                match udp.recv_from(&mut buf) {
                     Ok((len, addr)) => {
                         recv_handler.handle(
                             &mut buf[..len],

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,
@@ -205,6 +206,8 @@ impl Vnt {
             let context = context.clone();
             let nat_test = nat_test.clone();
             let device_list = device_list.clone();
+            let down_count_watcher = down_count_watcher.clone();
+            let up_count_watcher = up_count_watcher.clone();
             let current_device = current_device.clone();
             if !config.use_channel_type.is_only_relay() {
                 // 定时nat探测
@@ -229,6 +232,8 @@ impl Vnt {
                     config_info,
                     punch,
                     callback,
+                    down_count_watcher,
+                    up_count_watcher,
                 );
             });
         }
@@ -259,6 +264,8 @@ pub fn start<Call: VntCallback>(
     config_info: BaseConfigInfo,
     punch: Punch,
     callback: Call,
+    down_count_watcher: WatchU64Adder,
+    up_count_watcher: WatchSingleU64Adder,
 ) {
     // 定时心跳
     maintain::heartbeat(
@@ -310,6 +317,13 @@ pub fn start<Call: VntCallback>(
             punch,
         );
     }
+    maintain::up_status(
+        scheduler,
+        context.clone(),
+        current_device.clone(),
+        down_count_watcher,
+        up_count_watcher,
+    )
 }
 
 impl Vnt {

vnt/src/handle/handshaker.rs

@@ -37,7 +37,7 @@ impl Handshake {
     pub fn send(&self, context: &Context, secret: bool, addr: SocketAddr) -> io::Result<()> {
         let last = self.time.load();
         //短时间不重复发送
-        if last.elapsed() < Duration::from_secs(5) {
+        if last.elapsed() < Duration::from_secs(3) {
             return Ok(());
         }
         let request_packet = handshake_request_packet(secret)?;

vnt/src/handle/maintain/addr_request.rs

@@ -1,4 +1,3 @@
-use std::net::ToSocketAddrs;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -16,12 +15,25 @@ pub fn addr_request(
     context: Context,
     current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
     server_cipher: Cipher,
-    config: BaseConfigInfo,
+    _config: BaseConfigInfo,
 ) {
-    addr_request0(&context, &current_device_info, &server_cipher, &config);
-    // 17秒发送一次
-    let rs = scheduler.timeout(Duration::from_secs(17), |s| {
-        addr_request(s, context, current_device_info, server_cipher, config)
+    pub_address_request(
+        scheduler,
+        context,
+        current_device_info.clone(),
+        server_cipher,
+    );
+}
+pub fn pub_address_request(
+    scheduler: &Scheduler,
+    context: Context,
+    current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
+    server_cipher: Cipher,
+) {
+    addr_request0(&context, &current_device_info, &server_cipher);
+    // 9秒发送一次
+    let rs = scheduler.timeout(Duration::from_secs(9), |s| {
+        pub_address_request(s, context, current_device_info, server_cipher)
     });
     if !rs {
         log::info!("定时任务停止");
@@ -32,27 +44,9 @@ pub fn addr_request0(
     context: &Context,
     current_device: &AtomicCell<CurrentDeviceInfo>,
     server_cipher: &Cipher,
-    config: &BaseConfigInfo,
 ) {
-    let mut current_dev = current_device.load();
-    // 探测服务端地址变化
-    if let Ok(mut addr) = config.server_addr.to_socket_addrs() {
-        if let Some(addr) = addr.next() {
-            if addr != current_dev.connect_server {
-                let mut tmp = current_dev.clone();
-                tmp.connect_server = addr;
-                let rs = current_device.compare_exchange(current_dev, tmp);
-                current_dev.connect_server = addr;
-                log::info!(
-                    "服务端地址变化,旧地址:{}，新地址:{},替换结果:{}",
-                    current_dev.connect_server,
-                    addr,
-                    rs.is_ok()
-                );
-            }
-        }
-    }
-    if current_dev.connect_server.is_ipv4() {
+    let current_dev = current_device.load();
+    if current_dev.connect_server.is_ipv4() && current_dev.status.online() {
         // 如果连接的是ipv4服务，则探测公网端口
         let gateway_ip = current_dev.virtual_gateway;
         let src_ip = current_dev.virtual_ip;

vnt/src/handle/maintain/idle.rs

@@ -1,3 +1,11 @@
+use std::io;
+use std::net::{SocketAddr, ToSocketAddrs};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use crossbeam_utils::atomic::AtomicCell;
+use mio::net::TcpStream;
+
 use crate::channel::context::Context;
 use crate::channel::idle::{Idle, IdleType};
 use crate::channel::sender::AcceptSocketSender;
@@ -6,12 +14,6 @@ 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;
-use mio::net::TcpStream;
-use std::io;
-use std::net::SocketAddr;
-use std::sync::Arc;
-use std::time::Duration;
 
 pub fn idle_route<Call: VntCallback>(
     scheduler: &Scheduler,
@@ -29,6 +31,29 @@ pub fn idle_route<Call: VntCallback>(
     }
 }
 pub fn idle_gateway<Call: VntCallback>(
+    scheduler: &Scheduler,
+    context: Context,
+    current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
+    config: BaseConfigInfo,
+    tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
+    call: Call,
+    connect_count: usize,
+    handshake: Handshake,
+) {
+    let time = Instant::now();
+    idle_gateway_(
+        scheduler,
+        context,
+        current_device_info,
+        config,
+        tcp_socket_sender,
+        call,
+        connect_count,
+        handshake,
+        time,
+    );
+}
+pub fn idle_gateway_<Call: VntCallback>(
     scheduler: &Scheduler,
     context: Context,
     current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
@@ -37,6 +62,7 @@ pub fn idle_gateway<Call: VntCallback>(
     call: Call,
     mut connect_count: usize,
     handshake: Handshake,
+    mut time: Instant,
 ) {
     idle_gateway0(
         &context,
@@ -46,9 +72,10 @@ pub fn idle_gateway<Call: VntCallback>(
         &call,
         &mut connect_count,
         &handshake,
+        &mut time,
     );
     let rs = scheduler.timeout(Duration::from_secs(5), move |s| {
-        idle_gateway(
+        idle_gateway_(
             s,
             context,
             current_device_info,
@@ -57,6 +84,7 @@ pub fn idle_gateway<Call: VntCallback>(
             call,
             connect_count,
             handshake,
+            time,
         )
     });
     if !rs {
@@ -71,6 +99,7 @@ fn idle_gateway0<Call: VntCallback>(
     call: &Call,
     connect_count: &mut usize,
     handshake: &Handshake,
+    time: &mut Instant,
 ) {
     if let Err(e) = check_gateway_channel(
         context,
@@ -80,6 +109,7 @@ fn idle_gateway0<Call: VntCallback>(
         call,
         connect_count,
         handshake,
+        time,
     ) {
         let cur = current_device.load();
         call.error(ErrorInfo::new_msg(
@@ -113,16 +143,22 @@ fn idle_route0<Call: VntCallback>(
 
 fn check_gateway_channel<Call: VntCallback>(
     context: &Context,
-    current_device: &AtomicCell<CurrentDeviceInfo>,
+    current_device_info: &AtomicCell<CurrentDeviceInfo>,
     config: &BaseConfigInfo,
     tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
     call: &Call,
     count: &mut usize,
     handshake: &Handshake,
+    time: &mut Instant,
 ) -> io::Result<()> {
-    let current_device = current_device.load();
+    let mut current_device = current_device_info.load();
     if current_device.status.offline() {
         *count += 1;
+        if time.elapsed() < Duration::from_secs(6 * 60) {
+            // 探测服务器地址
+            current_device = domain_request0(current_device_info, config);
+            *time = Instant::now()
+        }
         //需要重连
         call.connect(ConnectInfo::new(*count, current_device.connect_server));
         log::info!("发送握手请求,{:?}", config);
@@ -149,3 +185,27 @@ fn check_gateway_channel<Call: VntCallback>(
     }
     Ok(())
 }
+pub fn domain_request0(
+    current_device: &AtomicCell<CurrentDeviceInfo>,
+    config: &BaseConfigInfo,
+) -> CurrentDeviceInfo {
+    let mut current_dev = current_device.load();
+    // 探测服务端地址变化
+    if let Ok(mut addr) = config.server_addr.to_socket_addrs() {
+        if let Some(addr) = addr.next() {
+            if addr != current_dev.connect_server {
+                let mut tmp = current_dev.clone();
+                tmp.connect_server = addr;
+                let rs = current_device.compare_exchange(current_dev, tmp);
+                current_dev.connect_server = addr;
+                log::info!(
+                    "服务端地址变化,旧地址:{}，新地址:{},替换结果:{}",
+                    current_dev.connect_server,
+                    addr,
+                    rs.is_ok()
+                );
+            }
+        }
+    }
+    current_dev
+}

vnt/src/handle/maintain/mod.rs

@@ -14,3 +14,6 @@ pub use punch::*;
 mod idle;
 pub use idle::idle_gateway;
 pub use idle::idle_route;
+
+mod up_status;
+pub use up_status::*;

vnt/src/handle/maintain/punch.rs

@@ -1,4 +1,4 @@
-use std::cmp::Ordering;
+use std::collections::HashMap;
 use std::net::Ipv4Addr;
 use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
 use std::sync::Arc;
@@ -27,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!(
@@ -54,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,
@@ -91,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,
@@ -99,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");
     };
@@ -123,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();
@@ -132,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)
         }
     }
@@ -152,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(
@@ -172,6 +203,8 @@ fn punch_request(
             current_device,
             client_cipher,
             count + 1,
+            punch_record,
+            last_punch_record,
         );
     });
     if !rs {
@@ -186,8 +219,21 @@ 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
+            || nat_info.public_ports.iter().filter(|&&v| v == 0).count()
+                > nat_info.public_ports.len() / 2)
+    {
+        log::info!("公网地址为空，暂时放弃打洞,第{}轮", total_count);
+        return Ok(());
+    }
     let current_ip = current_device.virtual_ip;
     let mut list: Vec<PeerDeviceInfo> = device_list
         .lock()
@@ -197,43 +243,51 @@ fn punch0(
         .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;
+        let punch_count = punch_record
+            .lock()
+            .get(&info.virtual_ip)
+            .cloned()
+            .unwrap_or(0);
+        let p2p_num = context.route_table.p2p_num(&info.virtual_ip);
+        let mut max_punch_interval = 70;
+        if p2p_num > 0 {
+            if punch_count == 0 {
+                continue;
+            }
+            if p2p_num >= context.channel_num() {
+                //通道数满足要求，不再打洞
+                punch_record.lock().remove(&info.virtual_ip);
+                continue;
+            }
+            //有p2p通道，但是通道数量不够，则继续打洞
+            // 提高等待上限
+            max_punch_interval = 300;
         }
-        if info.virtual_ip <= current_device.virtual_ip {
-            continue;
-        }
-        if !context.route_table.need_punch(&info.virtual_ip) {
-            continue;
-        }
-        count += 1;
-        if count > 2 {
+        // 能发起打洞的前提是自己空闲，这里会间隔5秒以上发起一次打洞，所以假定上一轮打洞已结束
+        let last_punch = last_punch_record
+            .get(&info.virtual_ip)
+            .cloned()
+            .unwrap_or(0);
+        // 梯度增加打洞时间间隔
+        if total_count > last_punch + punch_count.min(max_punch_interval) {
+            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/maintain/up_status.rs

@@ -0,0 +1,104 @@
+use crate::channel::context::Context;
+use crate::handle::CurrentDeviceInfo;
+use crate::proto::message::{ClientStatusInfo, PunchNatType, RouteItem};
+use crate::protocol::body::ENCRYPTION_RESERVED;
+use crate::protocol::{service_packet, NetPacket, Protocol, Version, HEAD_LEN, MAX_TTL};
+use crate::util::{Scheduler, WatchSingleU64Adder, WatchU64Adder};
+use crossbeam_utils::atomic::AtomicCell;
+use protobuf::Message;
+use std::io;
+use std::sync::Arc;
+use std::time::Duration;
+
+/// 上报状态给服务器
+pub fn up_status(
+    scheduler: &Scheduler,
+    context: Context,
+    current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
+    down_count_watcher: WatchU64Adder,
+    up_count_watcher: WatchSingleU64Adder,
+) {
+    let _ = scheduler.timeout(Duration::from_secs(60), move |x| {
+        up_status0(
+            x,
+            context,
+            current_device_info,
+            down_count_watcher,
+            up_count_watcher,
+        )
+    });
+}
+
+fn up_status0(
+    scheduler: &Scheduler,
+    context: Context,
+    current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
+    down_count_watcher: WatchU64Adder,
+    up_count_watcher: WatchSingleU64Adder,
+) {
+    if let Err(e) = send_up_status_packet(
+        &context,
+        &current_device_info,
+        &down_count_watcher,
+        &up_count_watcher,
+    ) {
+        log::warn!("{:?}", e)
+    }
+    let rs = scheduler.timeout(Duration::from_secs(10 * 60), move |x| {
+        up_status0(
+            x,
+            context,
+            current_device_info,
+            down_count_watcher,
+            up_count_watcher,
+        )
+    });
+    if !rs {
+        log::info!("定时任务停止");
+    }
+}
+
+fn send_up_status_packet(
+    context: &Context,
+    current_device_info: &AtomicCell<CurrentDeviceInfo>,
+    down_count_watcher: &WatchU64Adder,
+    up_count_watcher: &WatchSingleU64Adder,
+) -> io::Result<()> {
+    let device_info = current_device_info.load();
+    if device_info.status.offline() {
+        return Ok(());
+    }
+    let routes = context.route_table.route_table_p2p();
+    if routes.is_empty() {
+        return Ok(());
+    }
+    let mut message = ClientStatusInfo::new();
+    message.source = device_info.virtual_ip.into();
+    for (ip, _) in routes {
+        let mut item = RouteItem::new();
+        item.next_ip = ip.into();
+        message.p2p_list.push(item);
+    }
+    message.up_stream = up_count_watcher.get();
+    message.down_stream = down_count_watcher.get();
+    message.nat_type = protobuf::EnumOrUnknown::new(if context.is_cone() {
+        PunchNatType::Cone
+    } else {
+        PunchNatType::Symmetric
+    });
+    let buf = message
+        .write_to_bytes()
+        .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("up_status_packet {:?}", e)))?;
+    let mut net_packet =
+        NetPacket::new_encrypt(vec![0; HEAD_LEN + buf.len() + ENCRYPTION_RESERVED])?;
+    net_packet.set_version(Version::V1);
+    net_packet.set_gateway_flag(true);
+    net_packet.set_protocol(Protocol::Service);
+    net_packet.set_transport_protocol_into(service_packet::Protocol::ClientStatusInfo);
+    net_packet.first_set_ttl(MAX_TTL);
+    net_packet.set_source(device_info.virtual_ip);
+    net_packet.set_destination(device_info.virtual_gateway);
+    net_packet.set_payload(&buf)?;
+    context.send_default(net_packet.buffer(), device_info.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
@@ -267,79 +267,82 @@ impl<Call: VntCallback> ServerPacketHandler<Call> {
                         if old.virtual_ip != Ipv4Addr::UNSPECIFIED {
                             log::info!("ip发生变化,old:{:?},response={:?}", old, response);
                         }
-                        if let Err(e) = self.device.set_ip(virtual_ip, virtual_netmask) {
-                            log::error!("LocalIpExists {:?}", e);
-                            self.callback.error(ErrorInfo::new_msg(
-                                ErrorType::LocalIpExists,
-                                format!("set_ip {:?}", e),
-                            ));
-                            return Ok(());
-                        }
-                        let mut guard = self.route_record.lock();
-                        for (dest, mask) in guard.drain(..) {
-                            if let Err(e) = self.device.delete_route(dest, mask) {
-                                log::warn!("删除路由失败 ={:?}", e);
-                            }
-                        }
-                        if let Err(e) = self.device.add_route(virtual_network, virtual_netmask, 1) {
-                            log::warn!("添加默认路由失败 ={:?}", e);
-                        } else {
-                            guard.push((virtual_network, virtual_netmask));
-                        }
-                        if let Err(e) =
-                            self.device
-                                .add_route(Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST, 1)
+                        #[cfg(not(target_os = "android"))]
                         {
-                            log::warn!("添加广播路由失败 ={:?}", e);
-                        } else {
-                            guard.push((Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST));
-                        }
+                            if let Err(e) = self.device.set_ip(virtual_ip, virtual_netmask) {
+                                log::error!("LocalIpExists {:?}", e);
+                                self.callback.error(ErrorInfo::new_msg(
+                                    ErrorType::LocalIpExists,
+                                    format!("set_ip {:?}", e),
+                                ));
+                                return Ok(());
+                            }
+                            let mut guard = self.route_record.lock();
+                            for (dest, mask) in guard.drain(..) {
+                                if let Err(e) = self.device.delete_route(dest, mask) {
+                                    log::warn!("删除路由失败 ={:?}", e);
+                                }
+                            }
+                            if let Err(e) =
+                                self.device.add_route(virtual_network, virtual_netmask, 1)
+                            {
+                                log::warn!("添加默认路由失败 ={:?}", e);
+                            } else {
+                                guard.push((virtual_network, virtual_netmask));
+                            }
+                            if let Err(e) =
+                                self.device
+                                    .add_route(Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST, 1)
+                            {
+                                log::warn!("添加广播路由失败 ={:?}", e);
+                            } else {
+                                guard.push((Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST));
+                            }
 
-                        if let Err(e) = self.device.add_route(
-                            Ipv4Addr::from([224, 0, 0, 0]),
-                            Ipv4Addr::from([240, 0, 0, 0]),
-                            1,
-                        ) {
-                            log::warn!("添加组播路由失败 ={:?}", e);
-                        } else {
-                            guard.push((
+                            if let Err(e) = self.device.add_route(
                                 Ipv4Addr::from([224, 0, 0, 0]),
                                 Ipv4Addr::from([240, 0, 0, 0]),
-                            ));
-                        }
-
-                        for (dest, mask) in self.external_route.to_route() {
-                            if let Err(e) = self.device.add_route(dest, mask, 1) {
-                                log::warn!("添加路由失败 ={:?}", e);
+                                1,
+                            ) {
+                                log::warn!("添加组播路由失败 ={:?}", e);
                             } else {
-                                guard.push((dest, mask));
+                                guard.push((
+                                    Ipv4Addr::from([224, 0, 0, 0]),
+                                    Ipv4Addr::from([240, 0, 0, 0]),
+                                ));
+                            }
+
+                            for (dest, mask) in self.external_route.to_route() {
+                                if let Err(e) = self.device.add_route(dest, mask, 1) {
+                                    log::warn!("添加路由失败 ={:?}", e);
+                                } else {
+                                    guard.push((dest, mask));
+                                }
                             }
                         }
-                        self.callback.success();
                     }
                     self.set_device_info_list(response.device_info_list, response.epoch as _);
+                    if old.status.offline() {
+                        self.callback.success();
+                    }
                 }
             }
-            service_packet::Protocol::RegistrationRequest => {
-                //不处理注册包
-            }
-            service_packet::Protocol::PollDeviceList => {}
             service_packet::Protocol::PushDeviceList => {
                 let response = DeviceList::parse_from_bytes(net_packet.payload()).map_err(|e| {
                     io::Error::new(io::ErrorKind::Other, format!("PushDeviceList {:?}", e))
                 })?;
                 self.set_device_info_list(response.device_info_list, response.epoch as _);
             }
-            service_packet::Protocol::HandshakeRequest => {}
-            service_packet::Protocol::HandshakeResponse => {}
-            service_packet::Protocol::SecretHandshakeRequest => {}
             service_packet::Protocol::SecretHandshakeResponse => {
                 log::info!("SecretHandshakeResponse");
                 //加密握手结束，发送注册数据
                 self.register(current_device, context)?;
             }
-            service_packet::Protocol::Unknown(e) => {
-                log::warn!("service_packet::Protocol::Unknown = {}", e);
+            _ => {
+                log::warn!(
+                    "service_packet::Protocol::Unknown = {:?}",
+                    net_packet.head()
+                );
             }
         }
         Ok(())
@@ -370,7 +373,10 @@ impl<Call: VntCallback> ServerPacketHandler<Call> {
         let device_id = self.config_info.device_id.clone();
         let name = self.config_info.name.clone();
         let client_secret = self.config_info.client_secret;
-        let ip = self.config_info.ip;
+        let mut ip = self.config_info.ip;
+        if ip.is_none() {
+            ip = Some(current_device.virtual_ip)
+        }
         let response = registrar::registration_request_packet(
             &self.server_cipher,
             token,

vnt/src/handle/recv_data/turn.rs

@@ -29,12 +29,7 @@ impl PacketHandler for TurnPacketHandler {
             if let Some(route) = context.route_table.route_one(&destination) {
                 if route.addr == route_key.addr {
                     //防止环路
-                    log::warn!(
-                        "来源和目标相同 {:?},{},{}",
-                        route_key,
-                        net_packet.source(),
-                        net_packet.destination()
-                    );
+                    log::warn!("来源和目标相同 {:?},{:?}", route_key, net_packet.head());
                     return Ok(());
                 }
                 if route.metric <= ttl {

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,56 +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 {
+    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() && peer_ips.len() != MAX_COUNT {
-        //不需要转发
+    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)
 }
 
 /// 实现一个原地发送，必须保证是如下结构
@@ -88,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();
@@ -117,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(
@@ -143,5 +178,10 @@ pub fn base_handle(
         proxy_map.send_handle(&mut ipv4_packet)?;
     }
     client_cipher.encrypt_ipv4(&mut net_packet)?;
-    context.send_ipv4_by_id(net_packet.buffer(), &dest_ip, current_device.connect_server)
+    context.send_ipv4_by_id(
+        net_packet.buffer(),
+        &dest_ip,
+        current_device.connect_server,
+        current_device.status.online(),
+    )
 }

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/ip_proxy/icmp_proxy.rs

@@ -188,6 +188,7 @@ fn recv_handle(
                             net_packet.buffer(),
                             &dest_ip,
                             current_device.connect_server,
+                            current_device.status.online(),
                         ) {
                             log::warn!("发送到目标失败:{}", 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,

vnt/src/proto/message.rs

@@ -1622,6 +1622,323 @@ impl ::protobuf::reflect::ProtobufValue for PunchInfo {
     type RuntimeType = ::protobuf::reflect::rt::RuntimeTypeMessage<Self>;
 }
 
+#[derive(PartialEq,Clone,Default,Debug)]
+// @@protoc_insertion_point(message:ClientStatusInfo)
+pub struct ClientStatusInfo {
+    // message fields
+    // @@protoc_insertion_point(field:ClientStatusInfo.source)
+    pub source: u32,
+    // @@protoc_insertion_point(field:ClientStatusInfo.p2p_list)
+    pub p2p_list: ::std::vec::Vec<RouteItem>,
+    // @@protoc_insertion_point(field:ClientStatusInfo.up_stream)
+    pub up_stream: u64,
+    // @@protoc_insertion_point(field:ClientStatusInfo.down_stream)
+    pub down_stream: u64,
+    // @@protoc_insertion_point(field:ClientStatusInfo.nat_type)
+    pub nat_type: ::protobuf::EnumOrUnknown<PunchNatType>,
+    // special fields
+    // @@protoc_insertion_point(special_field:ClientStatusInfo.special_fields)
+    pub special_fields: ::protobuf::SpecialFields,
+}
+
+impl<'a> ::std::default::Default for &'a ClientStatusInfo {
+    fn default() -> &'a ClientStatusInfo {
+        <ClientStatusInfo as ::protobuf::Message>::default_instance()
+    }
+}
+
+impl ClientStatusInfo {
+    pub fn new() -> ClientStatusInfo {
+        ::std::default::Default::default()
+    }
+
+    fn generated_message_descriptor_data() -> ::protobuf::reflect::GeneratedMessageDescriptorData {
+        let mut fields = ::std::vec::Vec::with_capacity(5);
+        let mut oneofs = ::std::vec::Vec::with_capacity(0);
+        fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
+            "source",
+            |m: &ClientStatusInfo| { &m.source },
+            |m: &mut ClientStatusInfo| { &mut m.source },
+        ));
+        fields.push(::protobuf::reflect::rt::v2::make_vec_simpler_accessor::<_, _>(
+            "p2p_list",
+            |m: &ClientStatusInfo| { &m.p2p_list },
+            |m: &mut ClientStatusInfo| { &mut m.p2p_list },
+        ));
+        fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
+            "up_stream",
+            |m: &ClientStatusInfo| { &m.up_stream },
+            |m: &mut ClientStatusInfo| { &mut m.up_stream },
+        ));
+        fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
+            "down_stream",
+            |m: &ClientStatusInfo| { &m.down_stream },
+            |m: &mut ClientStatusInfo| { &mut m.down_stream },
+        ));
+        fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
+            "nat_type",
+            |m: &ClientStatusInfo| { &m.nat_type },
+            |m: &mut ClientStatusInfo| { &mut m.nat_type },
+        ));
+        ::protobuf::reflect::GeneratedMessageDescriptorData::new_2::<ClientStatusInfo>(
+            "ClientStatusInfo",
+            fields,
+            oneofs,
+        )
+    }
+}
+
+impl ::protobuf::Message for ClientStatusInfo {
+    const NAME: &'static str = "ClientStatusInfo";
+
+    fn is_initialized(&self) -> bool {
+        true
+    }
+
+    fn merge_from(&mut self, is: &mut ::protobuf::CodedInputStream<'_>) -> ::protobuf::Result<()> {
+        while let Some(tag) = is.read_raw_tag_or_eof()? {
+            match tag {
+                13 => {
+                    self.source = is.read_fixed32()?;
+                },
+                18 => {
+                    self.p2p_list.push(is.read_message()?);
+                },
+                24 => {
+                    self.up_stream = is.read_uint64()?;
+                },
+                32 => {
+                    self.down_stream = is.read_uint64()?;
+                },
+                40 => {
+                    self.nat_type = is.read_enum_or_unknown()?;
+                },
+                tag => {
+                    ::protobuf::rt::read_unknown_or_skip_group(tag, is, self.special_fields.mut_unknown_fields())?;
+                },
+            };
+        }
+        ::std::result::Result::Ok(())
+    }
+
+    // Compute sizes of nested messages
+    #[allow(unused_variables)]
+    fn compute_size(&self) -> u64 {
+        let mut my_size = 0;
+        if self.source != 0 {
+            my_size += 1 + 4;
+        }
+        for value in &self.p2p_list {
+            let len = value.compute_size();
+            my_size += 1 + ::protobuf::rt::compute_raw_varint64_size(len) + len;
+        };
+        if self.up_stream != 0 {
+            my_size += ::protobuf::rt::uint64_size(3, self.up_stream);
+        }
+        if self.down_stream != 0 {
+            my_size += ::protobuf::rt::uint64_size(4, self.down_stream);
+        }
+        if self.nat_type != ::protobuf::EnumOrUnknown::new(PunchNatType::Symmetric) {
+            my_size += ::protobuf::rt::int32_size(5, self.nat_type.value());
+        }
+        my_size += ::protobuf::rt::unknown_fields_size(self.special_fields.unknown_fields());
+        self.special_fields.cached_size().set(my_size as u32);
+        my_size
+    }
+
+    fn write_to_with_cached_sizes(&self, os: &mut ::protobuf::CodedOutputStream<'_>) -> ::protobuf::Result<()> {
+        if self.source != 0 {
+            os.write_fixed32(1, self.source)?;
+        }
+        for v in &self.p2p_list {
+            ::protobuf::rt::write_message_field_with_cached_size(2, v, os)?;
+        };
+        if self.up_stream != 0 {
+            os.write_uint64(3, self.up_stream)?;
+        }
+        if self.down_stream != 0 {
+            os.write_uint64(4, self.down_stream)?;
+        }
+        if self.nat_type != ::protobuf::EnumOrUnknown::new(PunchNatType::Symmetric) {
+            os.write_enum(5, ::protobuf::EnumOrUnknown::value(&self.nat_type))?;
+        }
+        os.write_unknown_fields(self.special_fields.unknown_fields())?;
+        ::std::result::Result::Ok(())
+    }
+
+    fn special_fields(&self) -> &::protobuf::SpecialFields {
+        &self.special_fields
+    }
+
+    fn mut_special_fields(&mut self) -> &mut ::protobuf::SpecialFields {
+        &mut self.special_fields
+    }
+
+    fn new() -> ClientStatusInfo {
+        ClientStatusInfo::new()
+    }
+
+    fn clear(&mut self) {
+        self.source = 0;
+        self.p2p_list.clear();
+        self.up_stream = 0;
+        self.down_stream = 0;
+        self.nat_type = ::protobuf::EnumOrUnknown::new(PunchNatType::Symmetric);
+        self.special_fields.clear();
+    }
+
+    fn default_instance() -> &'static ClientStatusInfo {
+        static instance: ClientStatusInfo = ClientStatusInfo {
+            source: 0,
+            p2p_list: ::std::vec::Vec::new(),
+            up_stream: 0,
+            down_stream: 0,
+            nat_type: ::protobuf::EnumOrUnknown::from_i32(0),
+            special_fields: ::protobuf::SpecialFields::new(),
+        };
+        &instance
+    }
+}
+
+impl ::protobuf::MessageFull for ClientStatusInfo {
+    fn descriptor() -> ::protobuf::reflect::MessageDescriptor {
+        static descriptor: ::protobuf::rt::Lazy<::protobuf::reflect::MessageDescriptor> = ::protobuf::rt::Lazy::new();
+        descriptor.get(|| file_descriptor().message_by_package_relative_name("ClientStatusInfo").unwrap()).clone()
+    }
+}
+
+impl ::std::fmt::Display for ClientStatusInfo {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
+        ::protobuf::text_format::fmt(self, f)
+    }
+}
+
+impl ::protobuf::reflect::ProtobufValue for ClientStatusInfo {
+    type RuntimeType = ::protobuf::reflect::rt::RuntimeTypeMessage<Self>;
+}
+
+#[derive(PartialEq,Clone,Default,Debug)]
+// @@protoc_insertion_point(message:RouteItem)
+pub struct RouteItem {
+    // message fields
+    // @@protoc_insertion_point(field:RouteItem.next_ip)
+    pub next_ip: u32,
+    // special fields
+    // @@protoc_insertion_point(special_field:RouteItem.special_fields)
+    pub special_fields: ::protobuf::SpecialFields,
+}
+
+impl<'a> ::std::default::Default for &'a RouteItem {
+    fn default() -> &'a RouteItem {
+        <RouteItem as ::protobuf::Message>::default_instance()
+    }
+}
+
+impl RouteItem {
+    pub fn new() -> RouteItem {
+        ::std::default::Default::default()
+    }
+
+    fn generated_message_descriptor_data() -> ::protobuf::reflect::GeneratedMessageDescriptorData {
+        let mut fields = ::std::vec::Vec::with_capacity(1);
+        let mut oneofs = ::std::vec::Vec::with_capacity(0);
+        fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
+            "next_ip",
+            |m: &RouteItem| { &m.next_ip },
+            |m: &mut RouteItem| { &mut m.next_ip },
+        ));
+        ::protobuf::reflect::GeneratedMessageDescriptorData::new_2::<RouteItem>(
+            "RouteItem",
+            fields,
+            oneofs,
+        )
+    }
+}
+
+impl ::protobuf::Message for RouteItem {
+    const NAME: &'static str = "RouteItem";
+
+    fn is_initialized(&self) -> bool {
+        true
+    }
+
+    fn merge_from(&mut self, is: &mut ::protobuf::CodedInputStream<'_>) -> ::protobuf::Result<()> {
+        while let Some(tag) = is.read_raw_tag_or_eof()? {
+            match tag {
+                13 => {
+                    self.next_ip = is.read_fixed32()?;
+                },
+                tag => {
+                    ::protobuf::rt::read_unknown_or_skip_group(tag, is, self.special_fields.mut_unknown_fields())?;
+                },
+            };
+        }
+        ::std::result::Result::Ok(())
+    }
+
+    // Compute sizes of nested messages
+    #[allow(unused_variables)]
+    fn compute_size(&self) -> u64 {
+        let mut my_size = 0;
+        if self.next_ip != 0 {
+            my_size += 1 + 4;
+        }
+        my_size += ::protobuf::rt::unknown_fields_size(self.special_fields.unknown_fields());
+        self.special_fields.cached_size().set(my_size as u32);
+        my_size
+    }
+
+    fn write_to_with_cached_sizes(&self, os: &mut ::protobuf::CodedOutputStream<'_>) -> ::protobuf::Result<()> {
+        if self.next_ip != 0 {
+            os.write_fixed32(1, self.next_ip)?;
+        }
+        os.write_unknown_fields(self.special_fields.unknown_fields())?;
+        ::std::result::Result::Ok(())
+    }
+
+    fn special_fields(&self) -> &::protobuf::SpecialFields {
+        &self.special_fields
+    }
+
+    fn mut_special_fields(&mut self) -> &mut ::protobuf::SpecialFields {
+        &mut self.special_fields
+    }
+
+    fn new() -> RouteItem {
+        RouteItem::new()
+    }
+
+    fn clear(&mut self) {
+        self.next_ip = 0;
+        self.special_fields.clear();
+    }
+
+    fn default_instance() -> &'static RouteItem {
+        static instance: RouteItem = RouteItem {
+            next_ip: 0,
+            special_fields: ::protobuf::SpecialFields::new(),
+        };
+        &instance
+    }
+}
+
+impl ::protobuf::MessageFull for RouteItem {
+    fn descriptor() -> ::protobuf::reflect::MessageDescriptor {
+        static descriptor: ::protobuf::rt::Lazy<::protobuf::reflect::MessageDescriptor> = ::protobuf::rt::Lazy::new();
+        descriptor.get(|| file_descriptor().message_by_package_relative_name("RouteItem").unwrap()).clone()
+    }
+}
+
+impl ::std::fmt::Display for RouteItem {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
+        ::protobuf::text_format::fmt(self, f)
+    }
+}
+
+impl ::protobuf::reflect::ProtobufValue for RouteItem {
+    type RuntimeType = ::protobuf::reflect::rt::RuntimeTypeMessage<Self>;
+}
+
 #[derive(Clone,Copy,PartialEq,Eq,Debug,Hash)]
 // @@protoc_insertion_point(enum:PunchNatType)
 pub enum PunchNatType {
@@ -1713,8 +2030,14 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     6\x18\t\x20\x01(\x0cR\x04ipv6\x12\x1b\n\tipv6_port\x18\n\x20\x01(\rR\x08\
     ipv6Port\x12\x19\n\x08tcp_port\x18\x0b\x20\x01(\rR\x07tcpPort\x12\x1b\n\
     \tudp_ports\x18\x0c\x20\x03(\rR\x08udpPorts\x12!\n\x0cpublic_ports\x18\r\
-    \x20\x03(\rR\x0bpublicPorts*'\n\x0cPunchNatType\x12\r\n\tSymmetric\x10\0\
-    \x12\x08\n\x04Cone\x10\x01b\x06proto3\
+    \x20\x03(\rR\x0bpublicPorts\"\xb9\x01\n\x10ClientStatusInfo\x12\x16\n\
+    \x06source\x18\x01\x20\x01(\x07R\x06source\x12%\n\x08p2p_list\x18\x02\
+    \x20\x03(\x0b2\n.RouteItemR\x07p2pList\x12\x1b\n\tup_stream\x18\x03\x20\
+    \x01(\x04R\x08upStream\x12\x1f\n\x0bdown_stream\x18\x04\x20\x01(\x04R\nd\
+    ownStream\x12(\n\x08nat_type\x18\x05\x20\x01(\x0e2\r.PunchNatTypeR\x07na\
+    tType\"$\n\tRouteItem\x12\x17\n\x07next_ip\x18\x01\x20\x01(\x07R\x06next\
+    Ip*'\n\x0cPunchNatType\x12\r\n\tSymmetric\x10\0\x12\x08\n\x04Cone\x10\
+    \x01b\x06proto3\
 ";
 
 /// `FileDescriptorProto` object which was a source for this generated file
@@ -1732,7 +2055,7 @@ pub fn file_descriptor() -> &'static ::protobuf::reflect::FileDescriptor {
     file_descriptor.get(|| {
         let generated_file_descriptor = generated_file_descriptor_lazy.get(|| {
             let mut deps = ::std::vec::Vec::with_capacity(0);
-            let mut messages = ::std::vec::Vec::with_capacity(8);
+            let mut messages = ::std::vec::Vec::with_capacity(10);
             messages.push(HandshakeRequest::generated_message_descriptor_data());
             messages.push(HandshakeResponse::generated_message_descriptor_data());
             messages.push(SecretHandshakeRequest::generated_message_descriptor_data());
@@ -1741,6 +2064,8 @@ pub fn file_descriptor() -> &'static ::protobuf::reflect::FileDescriptor {
             messages.push(DeviceInfo::generated_message_descriptor_data());
             messages.push(DeviceList::generated_message_descriptor_data());
             messages.push(PunchInfo::generated_message_descriptor_data());
+            messages.push(ClientStatusInfo::generated_message_descriptor_data());
+            messages.push(RouteItem::generated_message_descriptor_data());
             let mut enums = ::std::vec::Vec::with_capacity(1);
             enums.push(PunchNatType::generated_enum_descriptor_data());
             ::protobuf::reflect::GeneratedFileDescriptor::new_generated(

vnt/src/protocol/mod.rs

@@ -183,6 +183,9 @@ impl<B: AsRef<[u8]>> NetPacket<B> {
     pub fn payload(&self) -> &[u8] {
         &self.buffer.as_ref()[12..self.data_len]
     }
+    pub fn head(&self) -> &[u8] {
+        &self.buffer.as_ref()[..12]
+    }
 }
 
 impl<B: AsRef<[u8]> + AsMut<[u8]>> NetPacket<B> {
@@ -214,6 +217,9 @@ impl<B: AsRef<[u8]> + AsMut<[u8]>> NetPacket<B> {
     pub fn set_transport_protocol(&mut self, transport_protocol: u8) {
         self.buffer.as_mut()[2] = transport_protocol;
     }
+    pub fn set_transport_protocol_into<P: Into<u8>>(&mut self, transport_protocol: P) {
+        self.buffer.as_mut()[2] = transport_protocol.into();
+    }
     pub fn first_set_ttl(&mut self, ttl: u8) {
         self.buffer.as_mut()[3] = ttl << 4 | ttl;
     }

vnt/src/protocol/service_packet.rs

@@ -13,6 +13,8 @@ pub enum Protocol {
     HandshakeResponse,
     SecretHandshakeRequest,
     SecretHandshakeResponse,
+    /// 客户端上报状态
+    ClientStatusInfo,
     Unknown(u8),
 }
 
@@ -27,6 +29,7 @@ impl From<u8> for Protocol {
             6 => Self::HandshakeResponse,
             7 => Self::SecretHandshakeRequest,
             8 => Self::SecretHandshakeResponse,
+            9 => Self::ClientStatusInfo,
             val => Self::Unknown(val),
         }
     }
@@ -43,6 +46,7 @@ impl Into<u8> for Protocol {
             Self::HandshakeResponse => 6,
             Self::SecretHandshakeRequest => 7,
             Self::SecretHandshakeResponse => 8,
+            Self::ClientStatusInfo => 9,
             Self::Unknown(val) => val,
         }
     }