去除udp通道的arc包装

Update README.md

Cargo.toml

@@ -6,7 +6,6 @@ opt-level = 'z'
 debug = 0
 debug-assertions = false
 strip= "debuginfo"
-overflow-checks = true
 lto = true
 panic = 'abort'
 incremental = false

README.md

@@ -111,6 +111,37 @@ sudo iptables -t nat -A POSTROUTING  -o eth0 -s 10.26.0.0/24 -j MASQUERADE
 # 查看设置
 iptables -vnL -t nat
 ```
+
+### Arch Linux
+
+[![Packaging status](https://repology.org/badge/vertical-allrepos/vnt.svg)](https://repology.org/project/vnt/versions)
+
+- 通过 AUR 安装 [vnt-git](https://aur.archlinux.org/packages/vnt-git)
+
+```bash
+yay -Syu vnt
+```
+
+- 通过 `systemd` 设置开机自启及配置
+
+```bash
+sudo systemctl enable --now vnt-cli@
+sudo systemctl status vnt-cli@
+```
+
+- 启用内置 `IPv4` 转发规则
+
+```bash
+sudo sysctl --system
+```
+
+- 通过内置防火墙文件配置防火墙转发规则
+
+```bash
+sudo cat /etc/vnt/iptables-vnt.rules >> /etc/iptables/iptables.rules
+sudo iptables-restore iptables.rules
+```
+
 ### macos
 ```shell
 # 开启ip转发
@@ -127,6 +158,7 @@ sudo pfctl -f /etc/pf.conf -e
 
 - Mac
 - Linux
+    - Arch Linux `yay -Syu vnt`
 - Windows
     - 默认使用tun网卡 依赖wintun.dll([win-tun](https://www.wintun.net/))(将dll放到同目录下，建议使用版本0.14.1)
     - 使用tap网卡 依赖tap-windows([win-tap](https://build.openvpn.net/downloads/releases/))(建议使用版本9.24.7)
@@ -226,10 +258,16 @@ vnt默认使用10.26.0.0/24网段，和本地网络适配器的ip冲突
 
 ### 交流群
 
-QQ:1034868233
+QQ: 1034868233
 
 ### 其他
 
 可使用社区小伙伴搭建的中继服务器
 
 1. -s vnt.8443.eu.org:29871
+
+### 参与贡献
+
+<a href="https://github.com/lbl8603/vnt/graphs/contributors">
+  <img src="https://contrib.rocks/image?repo=lbl8603/vnt" />
+</a>

common/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "common"
-version = "1.2.6"
+version = "1.2.8"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

vnt-cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vnt-cli"
-version = "1.2.6"
+version = "1.2.8"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

vnt-cli/README.md

@@ -82,6 +82,8 @@
 取值0~65535，指定本地监听的端口，默认取随机端口
 ### --cmd
 开启交互式命令，开启后可以直接在窗口下输入命令，如需后台运行请勿开启
+### --first_latency
+优先使用低延迟通道，默认情况下优先使用p2p通道，某些情况下可能p2p比客户端中继延迟更高，可使用此参数进行优化传输
 ### --no-proxy
 关闭内置的ip代理，内置的代理较为简单，而且一般来说直接使用网卡NAT转发性能会更高，
 有需要可以自行配置NAT转发，[可参考‘编译’小节中的NAT配置](https://github.com/lbl8603/vnt#%E7%BC%96%E8%AF%91)
@@ -112,9 +114,11 @@ server_encrypt: true #服务端加密
 parallel: 1 #任务并行度
 cipher_model: aes_gcm #客户端加密算法
 finger: false #关闭数据指纹
-punch_model: ipv4 #打洞模式
+punch_model: ipv4 #打洞模式 
 port: 0 #使用随机端口
 cmd: false #关闭控制台输入
+no_proxy: false #是否关闭内置代理，true为关闭
+first_latency: false #是否优先低延迟通道，默认为false，表示优先使用p2p通道
 ```
 
 或者需要哪个配置就加哪个，当然token是必须的

vnt-cli/src/config/mod.rs

@@ -33,6 +33,7 @@ pub struct FileConfig {
     pub punch_model: String,
     pub port: u16,
     pub cmd: bool,
+    pub first_latency: bool,
 }
 
 impl Default for FileConfig {
@@ -64,6 +65,7 @@ impl Default for FileConfig {
             punch_model: "".to_string(),
             port: 0,
             cmd: false,
+            first_latency: false,
         }
     }
 }
@@ -155,7 +157,9 @@ pub fn read_config(file_path: &str) -> io::Result<(Config, bool)> {
         file_conf.finger,
         punch_model,
         file_conf.port,
-    );
+        file_conf.first_latency,
+    )
+    .unwrap();
     Ok((config, file_conf.cmd))
 }
 

vnt-cli/src/main.rs

@@ -59,6 +59,7 @@ fn main() {
     opts.optopt("", "port", "监听的端口", "<port>");
     opts.optflag("", "cmd", "开启窗口输入");
     opts.optflag("", "no-proxy", "关闭内置代理");
+    opts.optflag("", "first-latency", "优先延迟");
     opts.optopt("f", "", "配置文件", "<conf>");
     //"后台运行时,查看其他设备列表"
     opts.optflag("", "list", "后台运行时,查看其他设备列表");
@@ -261,6 +262,7 @@ fn main() {
         let cmd = matches.opt_present("cmd");
         #[cfg(feature = "ip_proxy")]
         let no_proxy = matches.opt_present("no-proxy");
+        let first_latency = matches.opt_present("first-latency");
         let config = Config::new(
             tap,
             token,
@@ -285,7 +287,9 @@ fn main() {
             finger,
             punch_model,
             port,
-        );
+            first_latency,
+        )
+        .unwrap();
         (config, cmd)
     };
     println!("version {}", vnt::VNT_VERSION);
@@ -541,6 +545,7 @@ fn print_usage(program: &str, _opts: Options) {
     println!("  --cmd               开启交互式命令，使用此参数开启控制台输入");
     #[cfg(feature = "ip_proxy")]
     println!("  --no-proxy          关闭内置代理，如需点对网则需要配置网卡NAT转发");
+    println!("  --first-latency     优先低延迟的通道，默认情况优先使用p2p通道");
 
     println!();
     println!(

vnt-jni/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vnt-jni"
-version = "1.2.6"
+version = "1.2.8"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

vnt-jni/src/vnt_util.rs

@@ -66,6 +66,7 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result<VntUtilSync, Error> {
     let cipher_model = to_string_not_null(env, &config, "cipherModel")?;
     let tcp = env.get_field(&config, "tcp", "Z")?.z()?;
     let finger = env.get_field(&config, "finger", "Z")?.z()?;
+    let first_latency = env.get_field(&config, "firstLatency", "Z")?.z()?;
     let in_ips = to_string(env, &config, "inIps")?;
     let out_ips = to_string(env, &config, "outIps")?;
     let port = env.get_field(&config, "port", "I")?.i()? as u16;
@@ -129,7 +130,7 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result<VntUtilSync, Error> {
     for addr in stun_server_str.split(",") {
         stun_server.push(addr.trim().to_string());
     }
-    let config = Config::new(
+    let config = match Config::new(
         false,
         token,
         device_id,
@@ -152,7 +153,18 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result<VntUtilSync, Error> {
         finger,
         PunchModel::All,
         port,
-    );
+        first_latency,
+    ) {
+        Ok(config) => config,
+        Err(e) => {
+            env.throw_new(
+                "java/lang/RuntimeException",
+                format!("vnt start error {}", e),
+            )
+            .expect("throw");
+            return Err(Error::JavaException);
+        }
+    };
     match VntUtilSync::new(config) {
         Ok(vnt_util) => Ok(vnt_util),
         Err(e) => {

vnt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vnt"
-version = "1.2.6"
+version = "1.2.8"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

vnt/src/channel/channel.rs

@@ -1,36 +1,35 @@
 use std::collections::HashMap;
 use std::io::{Read, Write};
-use std::net::TcpStream;
 use std::net::UdpSocket as StdUdpSocket;
-use std::net::{Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr};
+use std::net::{Ipv4Addr, Shutdown, SocketAddr};
-use std::sync::atomic::Ordering;
+use std::net::{SocketAddrV6, TcpStream};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use std::{io, thread};
 
-use crossbeam_epoch::{Atomic, Owned};
 use crossbeam_utils::atomic::AtomicCell;
+use parking_lot::{Mutex, RwLock};
 use tokio::net::UdpSocket;
 use tokio::sync::watch::{channel, Receiver, Sender};
 
 use crate::channel::punch::NatType;
-use crate::channel::{Route, RouteKey, Status, TCP_ID, UDP_ID, UDP_V6_ID};
+use crate::channel::{Route, RouteKey, Status, TCP_ID, UDP_ID};
 use crate::core::status::VntWorker;
 use crate::handle::recv_handler::ChannelDataHandler;
 use crate::handle::CurrentDeviceInfo;
 
 pub struct ContextInner {
     //udp用于打洞、服务端通信(可选)
-    pub(crate) main_channel: Arc<StdUdpSocket>,
+    pub(crate) main_channel: StdUdpSocket,
-    pub(crate) main_channel_ipv6: Option<Arc<StdUdpSocket>>,
     //在udp的基础上，可以选择使用tcp和服务端通信
-    pub(crate) main_tcp_channel: Option<std::sync::mpsc::SyncSender<Vec<u8>>>,
+    pub(crate) main_tcp_channel: Option<Mutex<TcpStream>>,
-    pub(crate) route_table: Atomic<HashMap<Ipv4Addr, Vec<(Route, Arc<AtomicCell<Instant>>)>>>,
+    pub(crate) route_table: RwLock<HashMap<Ipv4Addr, Vec<(Route, AtomicCell<Instant>)>>>,
     pub(crate) status_receiver: Receiver<Status>,
     pub(crate) status_sender: Sender<Status>,
-    pub(crate) udp_map: Atomic<HashMap<usize, Arc<UdpSocket>>>,
+    pub(crate) udp_map: RwLock<HashMap<usize, Arc<UdpSocket>>>,
     pub(crate) channel_num: usize,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
+    first_latency: bool,
 }
 
 #[derive(Clone)]
@@ -40,25 +39,26 @@ pub struct Context {
 
 impl Context {
     pub fn new(
-        main_channel: Arc<StdUdpSocket>,
+        main_channel: StdUdpSocket,
-        main_channel_ipv6: Option<Arc<StdUdpSocket>>,
+        main_tcp_channel: Option<TcpStream>,
-        main_tcp_channel: Option<std::sync::mpsc::SyncSender<Vec<u8>>>,
         current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
         _channel_num: usize,
+        first_latency: bool,
     ) -> Self {
         //当前版本只支持一个通道
         let channel_num = 1;
         let (status_sender, status_receiver) = channel(Status::Cone);
+        let main_tcp_channel = main_tcp_channel.map(|e| Mutex::new(e));
         let inner = Arc::new(ContextInner {
             main_channel,
-            main_channel_ipv6,
             main_tcp_channel,
-            route_table: Atomic::new(HashMap::with_capacity(16)),
+            route_table: RwLock::new(HashMap::with_capacity(16)),
             status_receiver,
             status_sender,
-            udp_map: Atomic::new(HashMap::with_capacity(16)),
+            udp_map: RwLock::new(HashMap::with_capacity(16)),
             channel_num,
             current_device,
+            first_latency,
         });
         Self { inner }
     }
@@ -73,26 +73,23 @@ impl Context {
     }
     pub fn close(&self) -> io::Result<()> {
         let _ = self.inner.status_sender.send(Status::Close);
-        if let Ok(port) = self.main_local_ipv4_port() {
+        if let Ok(port) = self.main_local_udp_port() {
             let _ = StdUdpSocket::bind("127.0.0.1:0")?.send_to(
                 b"stop",
                 SocketAddr::V4(std::net::SocketAddrV4::new(Ipv4Addr::LOCALHOST, port)),
             );
         }
-        if let Ok(port) = self.main_local_ipv6_port() {
-            let _ = StdUdpSocket::bind("[::]:0")?.send_to(
-                b"stop",
-                SocketAddr::V6(std::net::SocketAddrV6::new(Ipv6Addr::LOCALHOST, port, 0, 0)),
-            );
-        }
         if let Some(tcp) = &self.inner.main_tcp_channel {
-            let _ = tcp.send(vec![]);
+            tcp.lock().shutdown(Shutdown::Both)?;
         }
         Ok(())
     }
     pub fn is_main_tcp(&self) -> bool {
         self.inner.main_tcp_channel.is_some()
     }
+    pub fn is_first_latency(&self) -> bool {
+        self.inner.first_latency
+    }
     pub fn switch(&self, nat_type: NatType) {
         match nat_type {
             NatType::Symmetric => {
@@ -109,91 +106,65 @@ impl Context {
     pub fn switch_to_symmetric(&self) {
         let _ = self.inner.status_sender.send(Status::Symmetric);
     }
-    pub fn main_local_ipv4_port(&self) -> io::Result<u16> {
+    pub fn main_local_udp_port(&self) -> io::Result<u16> {
         self.inner.main_channel.local_addr().map(|k| k.port())
     }
-    pub fn main_local_ipv6_port(&self) -> io::Result<u16> {
-        if let Some(ipv6) = &self.inner.main_channel_ipv6 {
-            ipv6.local_addr().map(|k| k.port())
-        } else {
-            Err(io::Error::new(io::ErrorKind::Other, "not ipv6"))
-        }
-    }
     fn insert_udp(&self, id: usize, udp: Arc<UdpSocket>) {
-        self.insert_udp_(id, Some(udp))
+        self.inner.udp_map.write().insert(id, udp);
     }
     fn remove_udp(&self, id: usize) {
-        self.insert_udp_(id, None)
+        self.inner.udp_map.write().remove(&id);
     }
-    fn insert_udp_(&self, id: usize, udp: Option<Arc<UdpSocket>>) {
+    #[inline]
-        let guard = &crossbeam_epoch::pin();
+    pub fn send_main_udp(&self, buf: &[u8], mut addr: SocketAddr) -> io::Result<usize> {
-        let udp_map = &self.inner.udp_map;
+        if let SocketAddr::V4(ipv4) = addr {
-        let mut udp_map_shared = udp_map.load(Ordering::Acquire, guard);
+            addr = SocketAddr::V6(SocketAddrV6::new(
-        loop {
+                ipv4.ip().to_ipv6_mapped(),
-            let mut map = unsafe { udp_map_shared.deref().clone() };
+                ipv4.port(),
-            match udp.clone() {
+                0,
-                None => {
+                0,
-                    map.remove(&id);
+            ));
-                }
-                Some(udp) => {
-                    map.insert(id, udp);
-                }
-            }
-            match udp_map.compare_exchange(
-                udp_map_shared,
-                Owned::new(map),
-                Ordering::AcqRel,
-                Ordering::Relaxed,
-                guard,
-            ) {
-                Ok(_p) => unsafe {
-                    guard.defer_destroy(udp_map_shared);
-                    return;
-                },
-                Err(e) => {
-                    udp_map_shared = e.current;
-                }
-            }
         }
+        self.inner.main_channel.send_to(buf, addr)
     }
-    pub fn send_main_udp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
+    #[inline]
-        if addr.is_ipv6() {
+    pub fn send_main_tcp(&self, buf: &[u8]) -> io::Result<usize> {
-            if let Some(udp_ipv6) = &self.inner.main_channel_ipv6 {
+        if let Some(sender) = &self.inner.main_tcp_channel {
-                udp_ipv6.send_to(buf, addr)
+            let mut stream = sender.lock();
-            } else {
+            let mut head = [0; 4];
-                Err(io::Error::new(io::ErrorKind::Other, "not ipv6"))
+            let len = buf.len();
-            }
+            head[2] = (len >> 8) as u8;
+            head[3] = (len & 0xFF) as u8;
+            stream.write_all(&head)?;
+            stream.write_all(buf)?;
+            Ok(len)
         } else {
-            self.inner.main_channel.send_to(buf, addr)
+            return Err(io::Error::new(io::ErrorKind::NotFound, "tcp not found"));
         }
     }
 
     pub fn send_main(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
         if let Some(sender) = &self.inner.main_tcp_channel {
-            if sender.try_send(buf.to_vec()).is_ok() {
+            let mut stream = sender.lock();
-                Ok(buf.len())
+            let mut head = [0; 4];
-            } else {
+            let len = buf.len();
-                Err(io::Error::new(io::ErrorKind::Other, "send_main err"))
+            head[2] = (len >> 8) as u8;
-            }
+            head[3] = (len & 0xFF) as u8;
+            stream.write_all(&head)?;
+            stream.write_all(buf)?;
+            Ok(len)
         } else {
             self.send_main_udp(buf, addr)
         }
     }
 
     pub(crate) fn try_send_all(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> {
-        let table = unsafe {
+        let table = self.inner.udp_map.read();
-            let guard = &crossbeam_epoch::pin();
-            self.inner
-                .udp_map
-                .load(Ordering::Relaxed, guard)
-                .deref()
-                .clone()
-        };
         if table.is_empty() {
             log::error!("udp列表为空,addr={}", addr);
             return Ok(());
         }
-        for (_, udp) in table {
+        for (_, udp) in table.iter() {
             //使用ipv6的udp发送ipv4报文会出错
             if let Err(e) = udp.try_send_to(buf, addr) {
                 log::error!("{:?}", e);
@@ -211,14 +182,7 @@ impl Context {
         self.try_send_by_key(buf, &route.route_key())
     }
     fn get_route_by_id(&self, id: &Ipv4Addr) -> io::Result<Route> {
-        let guard = &crossbeam_epoch::pin();
+        if let Some(v) = self.inner.route_table.read().get(id) {
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        if let Some(v) = table.get(id) {
             if v.is_empty() {
                 return Err(io::Error::new(io::ErrorKind::NotFound, "route not found"));
             }
@@ -240,25 +204,8 @@ impl Context {
 
     pub async fn send_by_key(&self, buf: &[u8], route_key: &RouteKey) -> io::Result<usize> {
         match route_key.index {
-            TCP_ID => {
+            TCP_ID => self.send_main_tcp(buf),
-                if let Some(sender) = &self.inner.main_tcp_channel {
+            UDP_ID => self.send_main_udp(buf, route_key.addr),
-                    if sender.send(buf.to_vec()).is_ok() {
-                        Ok(buf.len())
-                    } else {
-                        Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                    }
-                } else {
-                    Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                }
-            }
-            UDP_ID => self.inner.main_channel.send_to(buf, route_key.addr),
-            UDP_V6_ID => {
-                if let Some(udp_ipv6) = &self.inner.main_channel_ipv6 {
-                    udp_ipv6.send_to(buf, route_key.addr)
-                } else {
-                    Err(io::Error::new(io::ErrorKind::Other, "not ipv6 udp"))
-                }
-            }
             _ => {
                 if let Some(udp) = self.get_udp_by_route(route_key) {
                     return udp.send_to(buf, route_key.addr).await;
@@ -269,25 +216,8 @@ impl Context {
     }
     pub fn try_send_by_key(&self, buf: &[u8], route_key: &RouteKey) -> io::Result<usize> {
         match route_key.index {
-            TCP_ID => {
+            TCP_ID => self.send_main_tcp(buf),
-                if let Some(sender) = &self.inner.main_tcp_channel {
+            UDP_ID => self.send_main_udp(buf, route_key.addr),
-                    if sender.try_send(buf.to_vec()).is_ok() {
-                        Ok(buf.len())
-                    } else {
-                        Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                    }
-                } else {
-                    Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                }
-            }
-            UDP_ID => self.inner.main_channel.send_to(buf, route_key.addr),
-            UDP_V6_ID => {
-                if let Some(udp_ipv6) = &self.inner.main_channel_ipv6 {
-                    udp_ipv6.send_to(buf, route_key.addr)
-                } else {
-                    Err(io::Error::new(io::ErrorKind::Other, "not ipv6 udp"))
-                }
-            }
             _ => {
                 if let Some(udp) = self.get_udp_by_route(route_key) {
                     return udp.try_send_to(buf, route_key.addr);
@@ -297,9 +227,7 @@ impl Context {
         }
     }
     fn get_udp_by_route(&self, route_key: &RouteKey) -> Option<Arc<UdpSocket>> {
-        let guard = &crossbeam_epoch::pin();
+        self.inner.udp_map.read().get(&route_key.index).cloned()
-        let udp_map = unsafe { self.inner.udp_map.load(Ordering::Relaxed, guard).deref() };
-        udp_map.get(&route_key.index).cloned()
     }
 
     pub fn add_route_if_absent(&self, id: Ipv4Addr, route: Route) {
@@ -310,96 +238,62 @@ impl Context {
     }
     fn add_route_(&self, id: Ipv4Addr, route: Route, only_if_absent: bool) {
         let key = route.route_key();
-        let guard = &crossbeam_epoch::pin();
+        let mut route_table = self.inner.route_table.write();
-        let route_table = &self.inner.route_table;
+        let list = route_table
-        let mut table_share = route_table.load(Ordering::Acquire, guard);
+            .entry(id)
-        loop {
+            .or_insert_with(|| Vec::with_capacity(4));
-            let mut table = unsafe { table_share.deref().clone() };
+        let mut exist = false;
-            let list = table.entry(id).or_insert_with(|| Vec::with_capacity(4));
+        for (x, time) in list.iter_mut() {
-            let mut exist = false;
+            if x.metric < route.metric && !self.inner.first_latency {
-            for (x, time) in list.iter_mut() {
+                //非优先延迟的情况下 不能比当前的路径更长
-                if x.metric < route.metric {
+                return;
-                    //不能比当前的路径更长
+            }
+            if x.route_key() == key {
+                if only_if_absent {
                     return;
                 }
-                if x.route_key() == key {
+                x.metric = route.metric;
-                    if only_if_absent {
+                x.rt = route.rt;
-                        return;
+                exist = true;
-                    }
+                time.store(Instant::now());
-                    x.metric = route.metric;
+                break;
-                    x.rt = route.rt;
-                    exist = true;
-                    time.store(Instant::now());
-                    break;
-                }
             }
-            if exist {
+        }
-                list.sort_by_key(|(k, _)| k.sort_key());
+        if exist {
+            list.sort_by_key(|(k, _)| k.rt);
+        } else {
+            let max_len = if self.inner.first_latency {
+                self.inner.channel_num + 1
             } else {
                 if route.metric == 1 {
-                    //添加了直连的则排除非直连的
+                    //非优先延迟的情况下 添加了直连的则排除非直连的
                     list.retain(|(k, _)| k.metric == 1);
                 }
-                list.push((route, Arc::new(AtomicCell::new(Instant::now()))));
+                self.inner.channel_num
-                list.sort_by_key(|(k, _)| k.sort_key());
+            };
-                let max_len = self.inner.channel_num + 1;
+            list.sort_by_key(|(k, _)| k.rt);
-                if list.len() > max_len {
+            if list.len() > max_len {
-                    list.truncate(max_len);
+                list.truncate(max_len);
-                }
-            }
-            match route_table.compare_exchange(
-                table_share,
-                Owned::new(table),
-                Ordering::AcqRel,
-                Ordering::Relaxed,
-                guard,
-            ) {
-                Ok(_p) => unsafe {
-                    guard.defer_destroy(table_share);
-                    break;
-                },
-                Err(e) => {
-                    table_share = e.current;
-                }
             }
+            list.push((route, AtomicCell::new(Instant::now())));
         }
     }
     pub fn route(&self, id: &Ipv4Addr) -> Option<Vec<Route>> {
-        let guard = &crossbeam_epoch::pin();
+        if let Some(v) = self.inner.route_table.read().get(id) {
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        if let Some(v) = table.get(id) {
             Some(v.iter().map(|(i, _)| *i).collect())
         } else {
             None
         }
     }
     pub fn route_one(&self, id: &Ipv4Addr) -> Option<Route> {
-        let guard = &crossbeam_epoch::pin();
+        if let Some(v) = self.inner.route_table.read().get(id) {
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        if let Some(v) = table.get(id) {
             v.first().map(|(i, _)| *i)
         } else {
             None
         }
     }
     pub fn route_to_id(&self, route_key: &RouteKey) -> Option<Ipv4Addr> {
-        let guard = &crossbeam_epoch::pin();
+        let table = self.inner.route_table.read();
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
         for (k, v) in table.iter() {
             for (route, _) in v {
                 if &route.route_key() == route_key && route.is_p2p() {
@@ -410,14 +304,7 @@ impl Context {
         None
     }
     pub fn need_punch(&self, id: &Ipv4Addr) -> bool {
-        let guard = &crossbeam_epoch::pin();
+        if let Some(v) = self.inner.route_table.read().get(id) {
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        if let Some(v) = table.get(id) {
             if v.iter().filter(|(k, _)| k.is_p2p()).count() >= self.inner.channel_num {
                 return false;
             }
@@ -425,13 +312,7 @@ impl Context {
         true
     }
     pub fn route_table(&self) -> Vec<(Ipv4Addr, Vec<Route>)> {
-        let guard = &crossbeam_epoch::pin();
+        let table = self.inner.route_table.read();
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
         table
             .iter()
             .map(|(k, v)| (k.clone(), v.iter().map(|(i, _)| *i).collect()))
@@ -439,14 +320,8 @@ impl Context {
     }
     pub fn route_table_one(&self) -> Vec<(Ipv4Addr, Route)> {
         let mut list = Vec::with_capacity(8);
-        let guard = &crossbeam_epoch::pin();
+        let table = self.inner.route_table.read();
-        let table = unsafe {
+        for (k, v) in table.iter() {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        for (k, v) in table {
             if let Some((route, _)) = v.first() {
                 list.push((*k, *route));
             }
@@ -455,14 +330,8 @@ impl Context {
     }
     pub fn direct_route_table_one(&self) -> Vec<(Ipv4Addr, Route)> {
         let mut list = Vec::with_capacity(8);
-        let guard = &crossbeam_epoch::pin();
+        let table = self.inner.route_table.read();
-        let table = unsafe {
+        for (k, v) in table.iter() {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        for (k, v) in table {
             if let Some((route, _)) = v.first() {
                 if route.metric == 1 {
                     list.push((*k, *route));
@@ -473,38 +342,14 @@ impl Context {
     }
 
     pub fn remove_route(&self, id: &Ipv4Addr, route_key: RouteKey) {
-        let guard = &crossbeam_epoch::pin();
+        if let Some(routes) = self.inner.route_table.write().get_mut(id) {
-        let route_table = &self.inner.route_table;
+            routes.retain(|(x, _)| x.route_key() != route_key);
-        let mut table_share = route_table.load(Ordering::Acquire, guard);
+        } else {
-        loop {
+            return;
-            let mut table = unsafe { table_share.deref().clone() };
-            if let Some(routes) = table.get_mut(id) {
-                routes.retain(|(x, _)| x.route_key() != route_key);
-                match route_table.compare_exchange(
-                    table_share,
-                    Owned::new(table),
-                    Ordering::AcqRel,
-                    Ordering::Relaxed,
-                    guard,
-                ) {
-                    Ok(_p) => unsafe {
-                        guard.defer_destroy(table_share);
-                        return;
-                    },
-                    Err(e) => {
-                        table_share = e.current;
-                    }
-                }
-            } else {
-                return;
-            }
         }
     }
     pub fn update_read_time(&self, id: &Ipv4Addr, route_key: &RouteKey) {
-        let guard = &crossbeam_epoch::pin();
+        if let Some(routes) = self.inner.route_table.read().get(id) {
-        let table_share = self.inner.route_table.load(Ordering::Relaxed, guard);
-        let table = unsafe { table_share.deref() };
-        if let Some(routes) = table.get(id) {
             for (route, time) in routes {
                 if &route.route_key() == route_key {
                     time.store(Instant::now());
@@ -560,14 +405,17 @@ fn buf_channel_group(size: usize) -> (BufSenderGroup, BufReceiverGroup) {
 impl Channel {
     fn tcp_handle(
         tcp_r: &mut TcpStream,
-        context: Context,
+        context: &Context,
-        handler: ChannelDataHandler,
+        handler: &ChannelDataHandler,
         head_reserve: usize,
     ) -> io::Result<()> {
         let mut head = [0; 4];
         let addr = tcp_r.peer_addr()?;
         let key = RouteKey::new(TCP_ID, addr);
         loop {
+            if context.is_close() {
+                return Ok(());
+            }
             let mut buf = [0; 4096];
             tcp_r.read_exact(&mut head)?;
             let len = (((head[2] as u16) << 8) | head[3] as u16) as usize;
@@ -578,97 +426,57 @@ impl Channel {
                 ));
             }
             tcp_r.read_exact(&mut buf[head_reserve..head_reserve + len])?;
-            handler.handle(&mut buf, head_reserve, head_reserve + len, key, &context);
+            handler.handle(&mut buf, head_reserve, head_reserve + len, key, context);
         }
     }
     fn start_tcp(
-        worker: VntWorker,
         mut tcp_stream: TcpStream,
-        receiver: std::sync::mpsc::Receiver<Vec<u8>>,
         context: Context,
         handler: ChannelDataHandler,
         head_reserve: usize,
     ) {
         let current_device = context.inner.current_device.clone();
-        {
-            let mut tcp_r = tcp_stream.try_clone().unwrap();
-            let context = context.clone();
-            let handler = handler.clone();
-            thread::Builder::new()
-                .name("tcp_reader".into())
-                .spawn(move || {
-                    if let Err(e) = Self::tcp_handle(&mut tcp_r, context, handler, head_reserve) {
-                        log::info!("tcp链接断开:{:?}", e);
-                    }
-                    if let Err(e) = tcp_r.shutdown(Shutdown::Both) {
-                        log::info!("tcp链接关闭异常:{:?}", e);
-                    }
-                })
-                .unwrap();
-        }
-        let mut head = [0; 4];
         loop {
-            let data = match receiver.recv() {
+            if let Err(e) = tcp_stream.set_nodelay(true) {
-                Ok(data) => data,
+                log::info!("set_nodelay:{:?}", e);
-                Err(_) => {
-                    break;
-                }
-            };
-            let len = data.len();
-            if len == 0 {
-                break;
             }
-            head[2] = (len >> 8) as u8;
+            if let Err(e) = tcp_stream.set_write_timeout(Some(Duration::from_secs(3))) {
-            head[3] = (len & 0xFF) as u8;
+                log::info!("set_write_timeout:{:?}", e);
-            let mut err = false;
-            if let Err(e) = tcp_stream.write_all(&head) {
-                err = true;
-                log::info!("发送失败,需要重连:{:?}", e);
-            } else if let Err(e) = tcp_stream.write_all(&data) {
-                err = true;
-                log::info!("发送失败,需要重连:{:?}", e);
             }
-            if err {
+            if let Err(e) = tcp_stream.set_read_timeout(Some(Duration::from_secs(10))) {
-                if let Err(e) = tcp_stream.shutdown(Shutdown::Both) {
+                log::info!("set_read_timeout:{:?}", e);
-                    log::info!("tcp链接关闭异常:{:?}", e);
+            }
+            if let Err(e) = Self::tcp_handle(&mut tcp_stream, &context, &handler, head_reserve) {
+                log::info!("tcp链接断开:{:?}", e);
+            }
+            if let Err(e) = tcp_stream.shutdown(Shutdown::Both) {
+                log::info!("tcp链接关闭异常:{:?}", e);
+            }
+            loop {
+                if context.is_close() {
+                    return;
                 }
-                match TcpStream::connect(current_device.load().connect_server) {
+                let device_info = current_device.load();
+                match TcpStream::connect(device_info.connect_server) {
                     Ok(tcp) => {
-                        tcp.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
+                        tcp_stream = tcp.try_clone().unwrap();
-                        tcp_stream = tcp;
+                        let mut guard = context.inner.main_tcp_channel.as_ref().unwrap().lock();
-                        let mut tcp_r = tcp_stream.try_clone().unwrap();
+                        *guard = tcp;
-                        let context = context.clone();
+                        break;
-                        let handler = handler.clone();
-                        thread::Builder::new()
-                            .name("tcp_reader".into())
-                            .spawn(move || {
-                                if let Err(e) =
-                                    Self::tcp_handle(&mut tcp_r, context, handler, head_reserve)
-                                {
-                                    log::info!("重连后 tcp链接断开:{:?}", e);
-                                }
-                                if let Err(e) = tcp_r.shutdown(Shutdown::Both) {
-                                    log::info!("重连后 tcp链接关闭异常:{:?}", e);
-                                }
-                            })
-                            .unwrap();
                     }
                     Err(e) => {
-                        log::info!("重连失败:{:?}", e);
+                        log::info!("重连失败,{},{:?}", device_info.connect_server, e);
+                        thread::sleep(Duration::from_secs(3));
                     }
                 }
             }
         }
-        if let Err(e) = tcp_stream.shutdown(Shutdown::Both) {
-            log::info!("tcp链接关闭异常:{:?}", e);
-        }
-        worker.stop_all();
     }
 
     pub async fn start(
         self,
         mut worker: VntWorker,
-        tcp: Option<(TcpStream, std::sync::mpsc::Receiver<Vec<u8>>)>,
+        tcp: Option<TcpStream>,
         head_reserve: usize,          //头部预留字节
         symmetric_channel_num: usize, //对称网络，则再加一组监听，提升打洞成功率
         relay: bool,
@@ -676,7 +484,7 @@ impl Channel {
     ) {
         let handler = self.handler.clone();
         let context = self.context;
-        let main_channel = context.inner.main_channel.clone();
+        let main_channel = context.inner.main_channel.try_clone().unwrap();
         let buf_sender = if parallel > 1 {
             let (buf_sender, buf_receiver) = buf_channel_group(parallel);
             let mut num = 0;
@@ -698,54 +506,26 @@ impl Channel {
         } else {
             None
         };
-        if let Some((tcp_stream, receiver)) = tcp {
+        if let Some(tcp_stream) = tcp {
             let context = context.clone();
             let handler = handler.clone();
             let main_channel_tcp = worker.worker("main_channel_tcp");
             thread::Builder::new()
-                .name("main_channel_tcp".into())
+                .name("channel_tcp".into())
                 .spawn(move || {
-                    Self::start_tcp(
+                    Self::start_tcp(tcp_stream, context, handler, head_reserve);
-                        main_channel_tcp,
+                    drop(main_channel_tcp)
-                        tcp_stream,
-                        receiver,
-                        context,
-                        handler,
-                        head_reserve,
-                    )
-                })
-                .unwrap();
-        }
-        if let Some(main_channel_ipv6) = &context.inner.main_channel_ipv6 {
-            let worker = worker.worker("main_channel_ipv6");
-            let context = context.clone();
-            let main_channel_ipv6 = main_channel_ipv6.clone();
-            let handler = handler.clone();
-            let buf_sender = buf_sender.clone();
-            thread::Builder::new()
-                .name("ipv6-recv".into())
-                .spawn(move || {
-                    log::info!("启动udp v6");
-                    Self::main_start_(
-                        worker,
-                        context,
-                        UDP_V6_ID,
-                        main_channel_ipv6,
-                        handler,
-                        buf_sender,
-                        head_reserve,
-                    )
                 })
                 .unwrap();
         }
         {
-            let worker = worker.worker("main_channel_1");
+            let worker = worker.worker("main_channel_udp");
             let context = context.clone();
-            let main_channel = main_channel.clone();
+            let main_channel = main_channel.try_clone().unwrap();
             let handler = handler.clone();
             let buf_sender = buf_sender.clone();
             thread::Builder::new()
-                .name("ipv4-recv".into())
+                .name("channel_udp".into())
                 .spawn(move || {
                     log::info!("启动udp v4");
                     Self::main_start_(
@@ -766,6 +546,7 @@ impl Channel {
         }
         let mut cur_status = Status::Cone;
         let mut status_receiver = context.inner.status_receiver.clone();
+        let channel_num = context.inner.channel_num;
         loop {
             tokio::select! {
                 _=worker.stop_wait()=>{
@@ -784,7 +565,7 @@ impl Channel {
                                         continue;
                                     }
                                     cur_status = Status::Symmetric;
-                                    for _ in 0..symmetric_channel_num {
+                                    for _ in 0..symmetric_channel_num - channel_num {
                                         match UdpSocket::bind("0.0.0.0:0").await {
                                             Ok(udp) => {
                                                 let udp = Arc::new(udp);
@@ -815,7 +596,7 @@ impl Channel {
         worker: VntWorker,
         context: Context,
         id: usize,
-        udp: Arc<StdUdpSocket>,
+        udp: StdUdpSocket,
         handler: ChannelDataHandler,
         buf_sender: Option<BufSenderGroup>,
         head_reserve: usize,

vnt/src/channel/idle.rs

@@ -1,11 +1,11 @@
-use crate::channel::channel::Context;
-use crate::channel::RouteKey;
 use std::io;
 use std::io::{Error, ErrorKind};
 use std::net::Ipv4Addr;
-use std::sync::atomic::Ordering;
 use std::time::Duration;
 
+use crate::channel::channel::Context;
+use crate::channel::RouteKey;
+
 pub struct Idle {
     read_idle: Duration,
     context: Context,
@@ -23,15 +23,7 @@ impl Idle {
         loop {
             let mut max = Duration::from_secs(0);
             {
-                let guard = &crossbeam_epoch::pin();
+                for (ip, routes) in self.context.inner.route_table.read().iter() {
-                let table = unsafe {
-                    self.context
-                        .inner
-                        .route_table
-                        .load(Ordering::Relaxed, guard)
-                        .deref()
-                };
-                for (ip, routes) in table.iter() {
                     for (route, time) in routes {
                         let last_read = time.load().elapsed();
                         if last_read >= self.read_idle {

vnt/src/channel/mod.rs

@@ -7,7 +7,6 @@ pub mod sender;
 
 const TCP_ID: usize = 0;
 const UDP_ID: usize = 1;
-const UDP_V6_ID: usize = 2;
 
 #[derive(Copy, Clone, Eq, PartialEq)]
 pub enum Status {

vnt/src/channel/punch.rs

@@ -52,7 +52,13 @@ impl NatInfo {
         ipv6_addr: SocketAddrV6,
         mut nat_type: NatType,
     ) -> Self {
-        public_ips.retain(|ip| !ip.is_loopback() && !ip.is_private() && !ip.is_unspecified());
+        public_ips.retain(|ip| {
+            !ip.is_multicast()
+                && !ip.is_broadcast()
+                && !ip.is_unspecified()
+                && !ip.is_loopback()
+                && !ip.is_private()
+        });
         if public_ips.len() > 1 {
             nat_type = NatType::Symmetric;
         }
@@ -65,6 +71,20 @@ impl NatInfo {
             nat_type,
         }
     }
+    pub fn update_addr(&mut self, ip: Ipv4Addr, port: u16) {
+        if !ip.is_multicast()
+            && !ip.is_broadcast()
+            && !ip.is_unspecified()
+            && !ip.is_loopback()
+            && !ip.is_private()
+            && port != 0
+        {
+            self.public_port = port;
+            if !self.public_ips.contains(&ip) {
+                self.public_ips.push(ip);
+            }
+        }
+    }
 }
 
 #[derive(Clone)]
@@ -165,15 +185,17 @@ impl Punch {
                 self.port_index.insert(id, index);
             }
             NatType::Cone => {
-                let is_cone = self.context.is_cone();
+                if nat_info.public_port != 0 {
-                for ip in nat_info.public_ips {
+                    let is_cone = self.context.is_cone();
-                    let addr = SocketAddr::V4(SocketAddrV4::new(ip, nat_info.public_port));
+                    for ip in nat_info.public_ips {
-                    self.context.send_main_udp(buf, addr)?;
+                        let addr = SocketAddr::V4(SocketAddrV4::new(ip, nat_info.public_port));
-                    if !is_cone {
+                        self.context.send_main_udp(buf, addr)?;
-                        //只有一方是对称，则对称方要使用全部端口发送数据，符合上述计算的概率
+                        if !is_cone {
-                        self.context.try_send_all(buf, addr)?;
+                            //只有一方是对称，则对称方要使用全部端口发送数据，符合上述计算的概率
+                            self.context.try_send_all(buf, addr)?;
+                        }
+                        tokio::time::sleep(Duration::from_millis(2)).await;
                     }
-                    tokio::time::sleep(Duration::from_millis(2)).await;
                 }
             }
         }

vnt/src/core/mod.rs

@@ -3,13 +3,11 @@ use std::io;
 use std::net::TcpStream;
 use std::net::UdpSocket;
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
-use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
 
-use crossbeam_epoch::Atomic;
 use crossbeam_utils::atomic::AtomicCell;
-use parking_lot::Mutex;
+use parking_lot::{Mutex, RwLock};
 use rand::Rng;
 use tokio::sync::mpsc::channel;
 
@@ -53,13 +51,12 @@ pub struct Vnt {
     device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
     nat_test: NatTest,
     connect_status: Arc<AtomicCell<ConnectStatus>>,
-    peer_nat_info_map: Arc<Atomic<HashMap<Ipv4Addr, NatInfo>>>,
+    peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
 }
 
 pub struct VntUtil {
     config: Config,
     main_channel: UdpSocket,
-    main_channel_ipv6: Option<UdpSocket>,
     main_tcp_channel: Option<TcpStream>,
     response: Option<RegResponse>,
     iface: Option<(DeviceWriter, DeviceReader)>,
@@ -69,24 +66,14 @@ pub struct VntUtil {
 
 impl VntUtil {
     pub fn new(config: Config) -> io::Result<VntUtil> {
+        let address: SocketAddr = format!("[::]:{}", config.port).parse().unwrap();
         //单个udp用同步的性能更好，但是代理和多端口监听用异步更方便，这里将两者结合起来
-        let main_channel = UdpSocket::bind(format!("0.0.0.0:{}", config.port))?;
+        let socket = socket2::Socket::new(socket2::Domain::IPV6, socket2::Type::DGRAM, None)?;
+        socket.set_only_v6(false)?;
+        socket.bind(&address.into())?;
+        let main_channel: UdpSocket = socket.into();
         main_channel.set_write_timeout(Some(Duration::from_secs(5)))?;
         main_channel.set_read_timeout(Some(Duration::from_secs(2)))?;
-        let main_channel_ipv6 = if config.punch_model != PunchModel::IPv4 {
-            match UdpSocket::bind(format!("[::]:{}", config.port)) {
-                Ok(main_channel_ipv6) => {
-                    main_channel_ipv6.set_write_timeout(Some(Duration::from_secs(5)))?;
-                    Some(main_channel_ipv6)
-                }
-                Err(e) => {
-                    log::warn!("绑定ipv6地址失败:{}", e);
-                    None
-                }
-            }
-        } else {
-            None
-        };
         let server_cipher = if config.server_encrypt {
             let mut key = [0u8; 32];
             rand::thread_rng().fill(&mut key);
@@ -97,7 +84,6 @@ impl VntUtil {
         Ok(VntUtil {
             config,
             main_channel,
-            main_channel_ipv6,
             main_tcp_channel: None,
             response: None,
             iface: None,
@@ -250,18 +236,17 @@ impl VntUtil {
 
         let (cone_sender, cone_receiver) = channel(3);
         let (symmetric_sender, symmetric_receiver) = channel(2);
-        let (tcp_sender, tcp) = if let Some(main_tcp_channel) = self.main_tcp_channel {
+        let (tcp_sender, tcp_receiver) = if let Some(main_tcp_channel) = self.main_tcp_channel {
-            let (tcp_sender, tcp_receiver) = std::sync::mpsc::sync_channel::<Vec<u8>>(100);
+            (Some(main_tcp_channel.try_clone()?), Some(main_tcp_channel))
-            (Some(tcp_sender), Some((main_tcp_channel, tcp_receiver)))
         } else {
             (None, None)
         };
         let context = Context::new(
-            Arc::new(self.main_channel),
+            self.main_channel,
-            self.main_channel_ipv6.map(|v| Arc::new(v)),
             tcp_sender,
             current_device.clone(),
             1,
+            config.first_latency,
         );
         let punch = Punch::new(context.clone(), config.punch_model);
         let idle = Idle::new(Duration::from_secs(16), context.clone());
@@ -278,16 +263,15 @@ impl VntUtil {
         ));
         let device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>> =
             Arc::new(Mutex::new((response.epoch, response.device_info_list)));
-        let peer_nat_info_map: Arc<Atomic<HashMap<Ipv4Addr, NatInfo>>> =
+        let peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>> =
-            Arc::new(Atomic::new(HashMap::new()));
+            Arc::new(RwLock::new(HashMap::with_capacity(16)));
         let connect_status = Arc::new(AtomicCell::new(ConnectStatus::Connected));
         let public_ip = response.public_ip;
         let public_port = response.public_port;
-        let local_port = context.main_local_ipv4_port().unwrap_or(0);
+        let local_port = context.main_local_udp_port().unwrap_or(0);
 
         let local_ipv4_addr = crate::nat::local_ipv4_addr(local_port);
-        let ipv6_port = context.main_local_ipv6_port().unwrap_or(0);
+        let ipv6_addr = crate::nat::local_ipv6_addr(local_port);
-        let ipv6_addr = crate::nat::local_ipv6_addr(ipv6_port);
         // NAT检测
         let nat_test = NatTest::new(
             config.stun_server.clone(),
@@ -398,7 +382,7 @@ impl VntUtil {
             let relay = config.relay;
             tokio::spawn(async move {
                 channel
-                    .start(channel_worker, tcp, 14, 65, relay, config.parallel)
+                    .start(channel_worker, tcp_receiver, 14, 65, relay, config.parallel)
                     .await
             });
         }
@@ -504,10 +488,7 @@ impl Vnt {
         self.current_device.load()
     }
     pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option<NatInfo> {
-        let guard = &crossbeam_epoch::pin();
+        self.peer_nat_info_map.read().get(ip).cloned()
-        let shared = self.peer_nat_info_map.load(Ordering::Acquire, guard);
-        let map = unsafe { shared.deref() };
-        map.get(ip).map(|e| e.clone())
     }
     pub fn connection_status(&self) -> ConnectStatus {
         self.connect_status.load()
@@ -590,6 +571,7 @@ pub struct Config {
     pub finger: bool,
     pub punch_model: PunchModel,
     pub port: u16,
+    pub first_latency: bool,
 }
 
 impl Config {
@@ -616,13 +598,23 @@ impl Config {
         finger: bool,
         punch_model: PunchModel,
         port: u16,
-    ) -> Self {
+        first_latency: bool,
+    ) -> Result<Self, Error> {
         for x in stun_server.iter_mut() {
             if !x.contains(":") {
                 x.push_str(":3478");
             }
         }
-        Self {
+        if token.is_empty() || token.len() > 128 {
+            return Err(Error::Stop(String::from("token too long")));
+        }
+        if device_id.is_empty() || device_id.len() > 128 {
+            return Err(Error::Stop(String::from("device_id too long")));
+        }
+        if name.is_empty() || name.len() > 128 {
+            return Err(Error::Stop(String::from("name too long")));
+        }
+        Ok(Self {
             tap,
             token,
             device_id,
@@ -646,6 +638,7 @@ impl Config {
             finger,
             punch_model,
             port,
-        }
+            first_latency,
+        })
     }
 }

vnt/src/handle/handshake_handler.rs

@@ -1,4 +1,4 @@
-use std::net::SocketAddr;
+use std::net::{SocketAddr, SocketAddrV6};
 
 use crate::channel::channel::Context;
 use crate::channel::RouteKey;
@@ -168,16 +168,25 @@ fn send_recv(
         }
         Ok(len)
     } else {
+        let server_address = if let SocketAddr::V4(ipv4) = server_address {
+            SocketAddr::V6(SocketAddrV6::new(
+                ipv4.ip().to_ipv6_mapped(),
+                ipv4.port(),
+                0,
+                0,
+            ))
+        } else {
+            server_address
+        };
         if let Err(e) = main_channel.send_to(send_buf, server_address) {
             return Err(HandshakeEnum::Other(format!("send error:{}", e)));
         }
         match main_channel.recv_from(recv_buf) {
             Ok((len, addr)) => {
                 if server_address != addr {
-                    Err(HandshakeEnum::Other(format!("invalid data,from {}", addr)))
+                    log::warn!("请求{:?}和响应{:?}地址不一致", server_address, addr);
-                } else {
-                    Ok(len)
                 }
+                Ok(len)
             }
             Err(e) => Err(HandshakeEnum::Other(format!("receiver error:{}", e))),
         }

vnt/src/handle/heartbeat_handler.rs

@@ -3,15 +3,15 @@ use std::net::{Ipv4Addr, ToSocketAddrs};
 use std::sync::Arc;
 use std::time::Duration;
 
+use crossbeam_utils::atomic::AtomicCell;
+use parking_lot::Mutex;
+use rand::prelude::SliceRandom;
+
 use crate::channel::idle::Idle;
 use crate::channel::sender::ChannelSender;
 use crate::channel::Route;
 use crate::cipher::Cipher;
 use crate::core::status::VntWorker;
-use crossbeam_utils::atomic::AtomicCell;
-use parking_lot::Mutex;
-use rand::prelude::SliceRandom;
-
 use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
 use crate::protocol::body::ENCRYPTION_RESERVED;
 use crate::protocol::control_packet::PingPacket;
@@ -64,6 +64,7 @@ pub fn start_heartbeat(
         worker.stop_all();
     });
 }
+
 pub fn start_heartbeat_main(
     mut worker: VntWorker,
     sender: ChannelSender,
@@ -200,7 +201,7 @@ async fn start_heartbeat_(
 
         let src = current_dev.virtual_ip();
 
-        if count < 7 || count % 7 == 0 {
+        if count % 10 == 7 {
             let mut route_list: Option<Vec<(Ipv4Addr, Vec<Route>)>> = None;
             let peer_list = { device_list.lock().1.clone() };
             for peer in peer_list {
@@ -222,7 +223,7 @@ async fn start_heartbeat_(
                     {
                         log::warn!("virtual_ip:{},route:{:?},e:{:?}", peer.virtual_ip, route, e);
                     }
-                    if route.is_p2p() {
+                    if route.is_p2p() && !sender.is_first_latency() {
                         continue;
                     }
                 } else {

vnt/src/handle/recv_handler.rs

@@ -1,12 +1,10 @@
 use std::collections::HashMap;
 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
-use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use crossbeam_epoch::{Atomic, Owned};
 use crossbeam_utils::atomic::AtomicCell;
-use parking_lot::Mutex;
+use parking_lot::{Mutex, RwLock};
 use protobuf::Message;
 use tokio::sync::mpsc::Sender;
 
@@ -47,7 +45,7 @@ pub struct ChannelDataHandler {
     igmp_server: Option<IgmpServer>,
     device_writer: DeviceWriter,
     connect_status: Arc<AtomicCell<ConnectStatus>>,
-    peer_nat_info_map: Arc<Atomic<HashMap<Ipv4Addr, NatInfo>>>,
+    peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
     #[cfg(feature = "ip_proxy")]
     ip_proxy_map: Option<IpProxyMap>,
     out_external_route: AllowExternalRoute,
@@ -70,7 +68,7 @@ impl ChannelDataHandler {
         igmp_server: Option<IgmpServer>,
         device_writer: DeviceWriter,
         connect_status: Arc<AtomicCell<ConnectStatus>>,
-        peer_nat_info_map: Arc<Atomic<HashMap<Ipv4Addr, NatInfo>>>,
+        peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
         #[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
         out_external_route: AllowExternalRoute,
         cone_sender: Sender<(Ipv4Addr, NatInfo)>,
@@ -422,15 +420,8 @@ impl ChannelDataHandler {
                     punch_info.nat_type.enum_value_or_default().into(),
                 );
                 {
-                    let guard = &crossbeam_epoch::pin();
+                    let peer_nat_info = peer_nat_info.clone();
-                    let nat_map = &self.peer_nat_info_map;
+                    self.peer_nat_info_map.write().insert(source, peer_nat_info);
-                    let nat_map_shared = nat_map.load(Ordering::Acquire, guard);
-                    let mut map = unsafe { nat_map_shared.deref().clone() };
-                    map.insert(source, peer_nat_info.clone());
-                    nat_map.store(Owned::new(map), Ordering::Release);
-                    unsafe {
-                        guard.defer_destroy(nat_map_shared);
-                    }
                 }
                 if !punch_info.reply {
                     let mut punch_reply = PunchInfo::new();
@@ -601,9 +592,8 @@ impl ChannelDataHandler {
                             .build()
                             .unwrap()
                             .block_on(async move {
-                                let local_port = context.main_local_ipv4_port().unwrap_or(0);
+                                let local_port = context.main_local_udp_port().unwrap_or(0);
                                 let local_ipv4_addr = nat::local_ipv4_addr(local_port);
-                                let local_port = context.main_local_ipv6_port().unwrap_or(0);
                                 let ipv6_addr = nat::local_ipv6_addr(local_port);
                                 let nat_info = nat_test
                                     .re_test(

vnt/src/handle/registration_handler.rs

@@ -1,6 +1,6 @@
 use crossbeam_utils::atomic::AtomicCell;
 use std::io::{Read, Write};
-use std::net::{Ipv4Addr, SocketAddr};
+use std::net::{Ipv4Addr, SocketAddr, SocketAddrV6};
 use std::time::{Duration, Instant};
 
 use crate::channel::sender::ChannelSender;
@@ -82,13 +82,22 @@ pub fn registration(
         }
         &mut recv_buf[4..len]
     } else {
+        let server_address = match server_address {
+            SocketAddr::V4(ipv4) => SocketAddr::V6(SocketAddrV6::new(
+                ipv4.ip().to_ipv6_mapped(),
+                ipv4.port(),
+                0,
+                0,
+            )),
+            SocketAddr::V6(_) => server_address,
+        };
         if let Err(e) = main_channel.send_to(buf, server_address) {
             return Err(ReqEnum::Other(format!("send error:{}", e)));
         }
         match main_channel.recv_from(&mut recv_buf) {
             Ok((len, addr)) => {
                 if server_address != addr {
-                    return Err(ReqEnum::Other(format!("invalid data,from {}", addr)));
+                    log::warn!("请求{:?}和响应{:?}地址不一致", server_address, addr);
                 }
                 &mut recv_buf[..len]
             }

vnt/src/igmp_server/mod.rs

@@ -1,10 +1,8 @@
 use std::collections::{HashMap, HashSet};
 use std::net::Ipv4Addr;
-use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use crossbeam_epoch::{Atomic, Owned};
 use parking_lot::RwLock;
 
 use packet::igmp::igmp_v2::IgmpV2Packet;
@@ -51,13 +49,13 @@ impl Multicast {
 
 #[derive(Clone)]
 pub struct IgmpServer {
-    multicast: Arc<Atomic<HashMap<Ipv4Addr, Arc<RwLock<Multicast>>>>>,
+    multicast: Arc<RwLock<HashMap<Ipv4Addr, Arc<RwLock<Multicast>>>>>,
 }
 
 impl IgmpServer {
     pub fn new(device_writer: DeviceWriter) -> Self {
-        let multicast: Arc<Atomic<HashMap<Ipv4Addr, Arc<RwLock<Multicast>>>>> =
+        let multicast: Arc<RwLock<HashMap<Ipv4Addr, Arc<RwLock<Multicast>>>>> =
-            Arc::new(Atomic::new(HashMap::with_capacity(16)));
+            Arc::new(RwLock::new(HashMap::with_capacity(16)));
         std::thread::spawn(move || {
             //预留以太网帧头和ip头
             let mut buf = [0; 14 + 24 + 12];
@@ -98,17 +96,10 @@ impl IgmpServer {
         Self { multicast }
     }
     pub fn load(&self, multicast_addr: &Ipv4Addr) -> Option<Arc<RwLock<Multicast>>> {
-        let guard = &crossbeam_epoch::pin();
+        self.multicast.read().get(multicast_addr).cloned()
-        let multicast = unsafe { self.multicast.load(Ordering::Relaxed, guard).deref() };
-        if let Some(entry) = multicast.get(multicast_addr) {
-            Some(entry.clone())
-        } else {
-            None
-        }
     }
     pub fn handle(&self, buf: &[u8], source: Ipv4Addr) -> crate::Result<()> {
-        let guard = &crossbeam_epoch::pin();
+        let multicast = self.multicast.read();
-        let multicast = unsafe { self.multicast.load(Ordering::Relaxed, guard).deref() };
         for (_, v) in multicast.iter() {
             let mut list = Vec::new();
             let mut write_guard = v.write();
@@ -142,7 +133,7 @@ impl IgmpServer {
                 if !multicast_addr.is_multicast() {
                     return Ok(());
                 }
-                if let Some(entry) = self.get_multicast(&multicast_addr) {
+                if let Some(entry) = self.load(&multicast_addr) {
                     let mut guard = entry.write();
                     guard.map.remove(&source);
                     guard.members.remove(&source);
@@ -234,35 +225,9 @@ impl IgmpServer {
         }
         Ok(())
     }
-    fn get_multicast(&self, multicast_addr: &Ipv4Addr) -> Option<Arc<RwLock<Multicast>>> {
-        let guard = &crossbeam_epoch::pin();
-        let multicast = &self.multicast;
-        let table_share = multicast.load(Ordering::Acquire, guard);
-        unsafe { table_share.deref().get(multicast_addr).map(|v| v.clone()) }
-    }
     fn add_multicast(&self, multicast_addr: Ipv4Addr) -> Arc<RwLock<Multicast>> {
-        let guard = &crossbeam_epoch::pin();
-        let multicast = &self.multicast;
-        let mut table_share = multicast.load(Ordering::Acquire, guard);
         let value = Arc::new(RwLock::new(Multicast::new()));
-        loop {
+        self.multicast.write().insert(multicast_addr, value.clone());
-            let mut table = unsafe { table_share.deref().clone() };
+        value
-            table.insert(multicast_addr, value.clone());
-            match self.multicast.compare_exchange(
-                table_share,
-                Owned::new(table.clone()),
-                Ordering::AcqRel,
-                Ordering::Relaxed,
-                guard,
-            ) {
-                Ok(_p) => unsafe {
-                    guard.defer_destroy(table_share);
-                    return value;
-                },
-                Err(e) => {
-                    table_share = e.current;
-                }
-            }
-        }
     }
 }

vnt/src/ip_proxy/mod.rs

@@ -1,4 +1,3 @@
-#[cfg(not(target_os = "android"))]
 use std::net::Ipv4Addr;
 use std::net::SocketAddrV4;
 use std::sync::Arc;
@@ -14,6 +13,7 @@ use packet::ip::ipv4::packet::IpV4Packet;
 use crate::channel::sender::ChannelSender;
 use crate::cipher::Cipher;
 use crate::handle::CurrentDeviceInfo;
+#[cfg(not(target_os = "android"))]
 use crate::ip_proxy::icmp_proxy::IcmpHandler;
 use crate::ip_proxy::tcp_proxy::{TcpHandler, TcpProxy};
 use crate::ip_proxy::udp_proxy::{UdpHandler, UdpProxy};

vnt/src/lib.rs

@@ -1,5 +1,5 @@
 use crate::error::Error;
-pub const VNT_VERSION: &'static str = "1.2.6";
+pub const VNT_VERSION: &'static str = "1.2.8";
 pub type Result<T> = std::result::Result<T, Error>;
 
 pub mod channel;

vnt/src/nat/mod.rs

@@ -111,19 +111,8 @@ impl NatTest {
         self.info.lock().clone()
     }
     pub fn update_addr(&self, ip: Ipv4Addr, port: u16) {
-        if !ip.is_multicast()
+        let mut guard = self.info.lock();
-            && !ip.is_broadcast()
+        guard.update_addr(ip, port)
-            && !ip.is_unspecified()
-            && !ip.is_loopback()
-            && !ip.is_private()
-            && port != 0
-        {
-            let mut guard = self.info.lock();
-            guard.public_port = port;
-            if !guard.public_ips.contains(&ip) {
-                guard.public_ips.push(ip);
-            }
-        }
     }
     pub async fn re_test(
         &self,