使用读写锁简化操作，增加延迟优先选项

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)
@@ -116,6 +118,7 @@ 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,6 +157,7 @@ pub fn read_config(file_path: &str) -> io::Result<(Config, bool)> {
         file_conf.finger,
         punch_model,
         file_conf.port,
+        file_conf.first_latency,
     );
     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,6 +287,7 @@ fn main() {
             finger,
             punch_model,
             port,
+            first_latency,
         );
         (config, cmd)
     };
@@ -541,6 +544,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/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;
@@ -152,6 +153,7 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result<VntUtilSync, Error> {
         finger,
         PunchModel::All,
         port,
+        first_latency,
     );
     match VntUtilSync::new(config) {
         Ok(vnt_util) => Ok(vnt_util),

vnt/src/channel/channel.rs

@@ -3,13 +3,12 @@ use std::io::{Read, Write};
 use std::net::TcpStream;
 use std::net::UdpSocket as StdUdpSocket;
 use std::net::{Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr};
-use std::sync::atomic::Ordering;
 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::RwLock;
 use tokio::net::UdpSocket;
 use tokio::sync::watch::{channel, Receiver, Sender};
 
@@ -25,12 +24,13 @@ pub struct ContextInner {
     pub(crate) main_channel_ipv6: Option<Arc<StdUdpSocket>>,
     //在udp的基础上，可以选择使用tcp和服务端通信
     pub(crate) main_tcp_channel: Option<std::sync::mpsc::SyncSender<Vec<u8>>>,
-    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)]
@@ -45,6 +45,7 @@ impl Context {
         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;
@@ -53,12 +54,13 @@ impl Context {
             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 }
     }
@@ -120,41 +122,10 @@ impl Context {
         }
     }
     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)
-    }
-    fn insert_udp_(&self, id: usize, udp: Option<Arc<UdpSocket>>) {
-        let guard = &crossbeam_epoch::pin();
-        let udp_map = &self.inner.udp_map;
-        let mut udp_map_shared = udp_map.load(Ordering::Acquire, guard);
-        loop {
-            let mut map = unsafe { udp_map_shared.deref().clone() };
-            match udp.clone() {
-                None => {
-                    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.udp_map.write().remove(&id);
     }
     pub fn send_main_udp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
         if addr.is_ipv6() {
@@ -181,19 +152,12 @@ impl Context {
     }
 
     pub(crate) fn try_send_all(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> {
-        let table = unsafe {
-            let guard = &crossbeam_epoch::pin();
-            self.inner
-                .udp_map
-                .load(Ordering::Relaxed, guard)
-                .deref()
-                .clone()
-        };
+        let table = self.inner.udp_map.read();
         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 +175,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();
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        if let Some(v) = table.get(id) {
+        if let Some(v) = self.inner.route_table.read().get(id) {
             if v.is_empty() {
                 return Err(io::Error::new(io::ErrorKind::NotFound, "route not found"));
             }
@@ -297,9 +254,7 @@ impl Context {
         }
     }
     fn get_udp_by_route(&self, route_key: &RouteKey) -> Option<Arc<UdpSocket>> {
-        let guard = &crossbeam_epoch::pin();
-        let udp_map = unsafe { self.inner.udp_map.load(Ordering::Relaxed, guard).deref() };
-        udp_map.get(&route_key.index).cloned()
+        self.inner.udp_map.read().get(&route_key.index).cloned()
     }
 
     pub fn add_route_if_absent(&self, id: Ipv4Addr, route: Route) {
@@ -310,96 +265,58 @@ 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 route_table = &self.inner.route_table;
-        let mut table_share = route_table.load(Ordering::Acquire, guard);
-        loop {
-            let mut table = unsafe { table_share.deref().clone() };
-            let list = table.entry(id).or_insert_with(|| Vec::with_capacity(4));
-            let mut exist = false;
-            for (x, time) in list.iter_mut() {
-                if x.metric < route.metric {
-                    //不能比当前的路径更长
+        let mut route_table = self.inner.route_table.write();
+        let list = route_table
+            .entry(id)
+            .or_insert_with(|| Vec::with_capacity(4));
+        let mut exist = false;
+        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 {
-                    if only_if_absent {
-                        return;
-                    }
-                    x.metric = route.metric;
-                    x.rt = route.rt;
-                    exist = true;
-                    time.store(Instant::now());
-                    break;
-                }
+                x.metric = route.metric;
+                x.rt = route.rt;
+                exist = true;
+                time.store(Instant::now());
+                break;
             }
-            if exist {
-                list.sort_by_key(|(k, _)| k.sort_key());
-            } else {
-                if route.metric == 1 {
-                    //添加了直连的则排除非直连的
-                    list.retain(|(k, _)| k.metric == 1);
-                }
-                list.push((route, Arc::new(AtomicCell::new(Instant::now()))));
-                list.sort_by_key(|(k, _)| k.sort_key());
-                let max_len = self.inner.channel_num + 1;
-                if list.len() > max_len {
-                    list.truncate(max_len);
-                }
+        }
+        if exist {
+            list.sort_by_key(|(k, _)| k.rt);
+        } else {
+            if route.metric == 1 && !self.inner.first_latency {
+                //非优先延迟的情况下 添加了直连的则排除非直连的
+                list.retain(|(k, _)| k.metric == 1);
             }
-            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.sort_by_key(|(k, _)| k.rt);
+            let max_len = self.inner.channel_num;
+            if list.len() > max_len {
+                list.truncate(max_len);
             }
+            list.push((route, AtomicCell::new(Instant::now())));
         }
     }
     pub fn route(&self, id: &Ipv4Addr) -> Option<Vec<Route>> {
-        let guard = &crossbeam_epoch::pin();
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        if let Some(v) = table.get(id) {
+        if let Some(v) = self.inner.route_table.read().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();
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        if let Some(v) = table.get(id) {
+        if let Some(v) = self.inner.route_table.read().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 = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
+        let table = self.inner.route_table.read();
         for (k, v) in table.iter() {
             for (route, _) in v {
                 if &route.route_key() == route_key && route.is_p2p() {
@@ -410,14 +327,7 @@ impl Context {
         None
     }
     pub fn need_punch(&self, id: &Ipv4Addr) -> bool {
-        let guard = &crossbeam_epoch::pin();
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        if let Some(v) = table.get(id) {
+        if let Some(v) = self.inner.route_table.read().get(id) {
             if v.iter().filter(|(k, _)| k.is_p2p()).count() >= self.inner.channel_num {
                 return false;
             }
@@ -425,13 +335,7 @@ impl Context {
         true
     }
     pub fn route_table(&self) -> Vec<(Ipv4Addr, Vec<Route>)> {
-        let guard = &crossbeam_epoch::pin();
-        let table = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
+        let table = self.inner.route_table.read();
         table
             .iter()
             .map(|(k, v)| (k.clone(), v.iter().map(|(i, _)| *i).collect()))
@@ -439,14 +343,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 = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        for (k, v) in table {
+        let table = self.inner.route_table.read();
+        for (k, v) in table.iter() {
             if let Some((route, _)) = v.first() {
                 list.push((*k, *route));
             }
@@ -455,14 +353,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 = unsafe {
-            self.inner
-                .route_table
-                .load(Ordering::Relaxed, guard)
-                .deref()
-        };
-        for (k, v) in table {
+        let table = self.inner.route_table.read();
+        for (k, v) in table.iter() {
             if let Some((route, _)) = v.first() {
                 if route.metric == 1 {
                     list.push((*k, *route));
@@ -473,38 +365,14 @@ impl Context {
     }
 
     pub fn remove_route(&self, id: &Ipv4Addr, route_key: RouteKey) {
-        let guard = &crossbeam_epoch::pin();
-        let route_table = &self.inner.route_table;
-        let mut table_share = route_table.load(Ordering::Acquire, guard);
-        loop {
-            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;
-            }
+        if let Some(routes) = self.inner.route_table.write().get_mut(id) {
+            routes.retain(|(x, _)| x.route_key() != route_key);
+        } else {
+            return;
         }
     }
     pub fn update_read_time(&self, id: &Ipv4Addr, route_key: &RouteKey) {
-        let guard = &crossbeam_epoch::pin();
-        let table_share = self.inner.route_table.load(Ordering::Relaxed, guard);
-        let table = unsafe { table_share.deref() };
-        if let Some(routes) = table.get(id) {
+        if let Some(routes) = self.inner.route_table.read().get(id) {
             for (route, time) in routes {
                 if &route.route_key() == route_key {
                     time.store(Instant::now());

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();
-                let table = unsafe {
-                    self.context
-                        .inner
-                        .route_table
-                        .load(Ordering::Relaxed, guard)
-                        .deref()
-                };
-                for (ip, routes) in table.iter() {
+                for (ip, routes) in self.context.inner.route_table.read().iter() {
                     for (route, time) in routes {
                         let last_read = time.load().elapsed();
                         if last_read >= self.read_idle {

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,7 +51,7 @@ 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 {
@@ -262,6 +260,7 @@ impl VntUtil {
             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,8 +277,8 @@ 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>>> =
-            Arc::new(Atomic::new(HashMap::new()));
+        let peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>> =
+            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;
@@ -504,10 +503,7 @@ impl Vnt {
         self.current_device.load()
     }
     pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option<NatInfo> {
-        let guard = &crossbeam_epoch::pin();
-        let shared = self.peer_nat_info_map.load(Ordering::Acquire, guard);
-        let map = unsafe { shared.deref() };
-        map.get(ip).map(|e| e.clone())
+        self.peer_nat_info_map.read().get(ip).cloned()
     }
     pub fn connection_status(&self) -> ConnectStatus {
         self.connect_status.load()
@@ -590,6 +586,7 @@ pub struct Config {
     pub finger: bool,
     pub punch_model: PunchModel,
     pub port: u16,
+    pub first_latency: bool,
 }
 
 impl Config {
@@ -616,6 +613,7 @@ impl Config {
         finger: bool,
         punch_model: PunchModel,
         port: u16,
+        first_latency: bool,
     ) -> Self {
         for x in stun_server.iter_mut() {
             if !x.contains(":") {
@@ -646,6 +644,7 @@ impl Config {
             finger,
             punch_model,
             port,
+            first_latency,
         }
     }
 }

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 nat_map = &self.peer_nat_info_map;
-                    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);
-                    }
+                    let peer_nat_info = peer_nat_info.clone();
+                    self.peer_nat_info_map.write().insert(source, peer_nat_info);
                 }
                 if !punch_info.reply {
                     let mut punch_reply = PunchInfo::new();

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>>>>> =
-            Arc::new(Atomic::new(HashMap::with_capacity(16)));
+        let multicast: Arc<RwLock<HashMap<Ipv4Addr, Arc<RwLock<Multicast>>>>> =
+            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();
-        let multicast = unsafe { self.multicast.load(Ordering::Relaxed, guard).deref() };
-        if let Some(entry) = multicast.get(multicast_addr) {
-            Some(entry.clone())
-        } else {
-            None
-        }
+        self.multicast.read().get(multicast_addr).cloned()
     }
     pub fn handle(&self, buf: &[u8], source: Ipv4Addr) -> crate::Result<()> {
-        let guard = &crossbeam_epoch::pin();
-        let multicast = unsafe { self.multicast.load(Ordering::Relaxed, guard).deref() };
+        let multicast = self.multicast.read();
         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 {
-            let mut table = unsafe { table_share.deref().clone() };
-            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;
-                }
-            }
-        }
+        self.multicast.write().insert(multicast_addr, value.clone());
+        value
     }
 }

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};