diff --git a/vnt-cli/README.md b/vnt-cli/README.md index db9d771..e33a7b4 100644 --- a/vnt-cli/README.md +++ b/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是必须的 diff --git a/vnt-cli/src/config/mod.rs b/vnt-cli/src/config/mod.rs index c11a66a..97415f2 100644 --- a/vnt-cli/src/config/mod.rs +++ b/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)) } diff --git a/vnt-cli/src/main.rs b/vnt-cli/src/main.rs index c193d55..1f823c9 100644 --- a/vnt-cli/src/main.rs +++ b/vnt-cli/src/main.rs @@ -59,6 +59,7 @@ fn main() { opts.optopt("", "port", "监听的端口", ""); opts.optflag("", "cmd", "开启窗口输入"); opts.optflag("", "no-proxy", "关闭内置代理"); + opts.optflag("", "first-latency", "优先延迟"); opts.optopt("f", "", "配置文件", ""); //"后台运行时,查看其他设备列表" 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!( diff --git a/vnt-jni/src/vnt_util.rs b/vnt-jni/src/vnt_util.rs index 38bc3bb..9e6b478 100644 --- a/vnt-jni/src/vnt_util.rs +++ b/vnt-jni/src/vnt_util.rs @@ -66,6 +66,7 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result { 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 { finger, PunchModel::All, port, + first_latency, ); match VntUtilSync::new(config) { Ok(vnt_util) => Ok(vnt_util), diff --git a/vnt/src/channel/channel.rs b/vnt/src/channel/channel.rs index 36c369c..da032eb 100644 --- a/vnt/src/channel/channel.rs +++ b/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>, //在udp的基础上,可以选择使用tcp和服务端通信 pub(crate) main_tcp_channel: Option>>, - pub(crate) route_table: Atomic>)>>>, + pub(crate) route_table: RwLock)>>>, pub(crate) status_receiver: Receiver, pub(crate) status_sender: Sender, - pub(crate) udp_map: Atomic>>, + pub(crate) udp_map: RwLock>>, pub(crate) channel_num: usize, current_device: Arc>, + first_latency: bool, } #[derive(Clone)] @@ -45,6 +45,7 @@ impl Context { main_tcp_channel: Option>>, current_device: Arc>, _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) { - 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>) { - 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 { 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 { - 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> { - 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> { - 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 { - 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 { - 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)> { - 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()); diff --git a/vnt/src/channel/idle.rs b/vnt/src/channel/idle.rs index 93e178b..45501bf 100644 --- a/vnt/src/channel/idle.rs +++ b/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 { diff --git a/vnt/src/core/mod.rs b/vnt/src/core/mod.rs index 91fccb2..8ca924c 100644 --- a/vnt/src/core/mod.rs +++ b/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)>>, nat_test: NatTest, connect_status: Arc>, - peer_nat_info_map: Arc>>, + peer_nat_info_map: Arc>>, } 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)>> = Arc::new(Mutex::new((response.epoch, response.device_info_list))); - let peer_nat_info_map: Arc>> = - Arc::new(Atomic::new(HashMap::new())); + let peer_nat_info_map: Arc>> = + 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 { - 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, } } } diff --git a/vnt/src/handle/recv_handler.rs b/vnt/src/handle/recv_handler.rs index b5a5653..3717b7a 100644 --- a/vnt/src/handle/recv_handler.rs +++ b/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, device_writer: DeviceWriter, connect_status: Arc>, - peer_nat_info_map: Arc>>, + peer_nat_info_map: Arc>>, #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, out_external_route: AllowExternalRoute, @@ -70,7 +68,7 @@ impl ChannelDataHandler { igmp_server: Option, device_writer: DeviceWriter, connect_status: Arc>, - peer_nat_info_map: Arc>>, + peer_nat_info_map: Arc>>, #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, 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(); diff --git a/vnt/src/igmp_server/mod.rs b/vnt/src/igmp_server/mod.rs index fd32ee8..5877b05 100644 --- a/vnt/src/igmp_server/mod.rs +++ b/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>>>>, + multicast: Arc>>>>, } impl IgmpServer { pub fn new(device_writer: DeviceWriter) -> Self { - let multicast: Arc>>>> = - Arc::new(Atomic::new(HashMap::with_capacity(16))); + let multicast: Arc>>>> = + 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>> { - 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>> { - 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> { - 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 } } diff --git a/vnt/src/ip_proxy/mod.rs b/vnt/src/ip_proxy/mod.rs index e379708..9e70552 100644 --- a/vnt/src/ip_proxy/mod.rs +++ b/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};