diff --git a/switch-desktop/Cargo.toml b/switch-desktop/Cargo.toml index 16fad16..0de5a18 100644 --- a/switch-desktop/Cargo.toml +++ b/switch-desktop/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "switch-desktop" -version = "1.0.6" +version = "1.0.7" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html diff --git a/switch-desktop/locales/en.yml b/switch-desktop/locales/en.yml index 83fde24..cc67be2 100644 --- a/switch-desktop/locales/en.yml +++ b/switch-desktop/locales/en.yml @@ -11,6 +11,7 @@ switch_tap_help: "Use tap mode, tun mode will be used by default" switch_in_ip_help: "Use when configuring point-to-network (IP proxy), --in-ip 192.168.10.0/24,10.26.0.3, which means it is allowed to receive data from the network segment 192.168.10.0/24 and forward it to 10.26.0.3" switch_out_ip_help: "Use when configuring point-to-network, --out-ip 192.168.10.0/24,192.168.1.10, which means that the data with the target of 192.168.10.0/24 is allowed to be forwarded from the network card 192.168.1.10" switch_password_help: "Client Data Encryption" +switch_simulate_multicast_help: "Simulate multicast. By default, multicast data will be sent as broadcast, which is more compatible, but it will cause traffic waste. After it is turned on, it will simulate real multicast data transmission" switch_config_help: "Read configuration file" switch_stop_about: "Stop background service" switch_route_about: "View route" diff --git a/switch-desktop/locales/zh-CN.yml b/switch-desktop/locales/zh-CN.yml index 7b69724..1d7f378 100644 --- a/switch-desktop/locales/zh-CN.yml +++ b/switch-desktop/locales/zh-CN.yml @@ -11,6 +11,7 @@ switch_tap_help: "使用tap模式，默认会使用tun模式" switch_in_ip_help: "配置点对网(IP代理)时使用，--in-ip 192.168.10.0/24,10.26.0.3，表示允许接收网段192.168.10.0/24的数据并转发到10.26.0.3" switch_out_ip_help: "配置点对网时使用，--out-ip 192.168.10.0/24,192.168.1.10，表示允许目标为192.168.10.0/24的数据从网卡192.168.1.10转发出去" switch_password_help: "使用该密码生成的密钥对客户端数据进行加密，并且服务端无法解密。使用相同密码的客户端才能通信" +switch_simulate_multicast_help: "模拟组播，默认情况下组播数据会被当作广播发送，兼容性更强，但是会造成流量浪费。开启后会模拟真实组播的数据发送" switch_config_help: "读取配置文件" switch_stop_about: "停止后台服务" switch_route_about: "查看路由" diff --git a/switch-desktop/src/command_args.rs b/switch-desktop/src/command_args.rs index 04bb773..3c78032 100644 --- a/switch-desktop/src/command_args.rs +++ b/switch-desktop/src/command_args.rs @@ -50,7 +50,8 @@ fn common() -> Command { Arg::new("tap") .long("tap") .help(switch_tap_help()) - .action(ArgAction::SetTrue), + .action(ArgAction::SetTrue) + .value_parser(BoolishValueParser::new()), ).arg( Arg::new("in_ip") .long("in-ip") @@ -67,6 +68,12 @@ fn common() -> Command { .long("password") .help(switch_password_help()) .action(ArgAction::Set) + ).arg( + Arg::new("simulate_multicast") + .long("simulate-multicast") + .help(switch_simulate_multicast_help()) + .action(ArgAction::SetTrue) + .value_parser(BoolishValueParser::new()), ).arg( Arg::new("config") .long("config") diff --git a/switch-desktop/src/config/mod.rs b/switch-desktop/src/config/mod.rs index cc24154..992ffbf 100644 --- a/switch-desktop/src/config/mod.rs +++ b/switch-desktop/src/config/mod.rs @@ -39,6 +39,7 @@ pub struct StartConfig { pub off_command_server: bool, pub log: bool, pub password: Option, + pub simulate_multicast:bool, } fn ips_parse(ips: &Vec) -> Result, String> { @@ -189,6 +190,7 @@ pub fn default_config(start_args: StartArgs) -> Result { off_command_server: start_args.off_command_server, log: start_args.log, password: start_args.password, + simulate_multicast:start_args.simulate_multicast, }; println!("========参数配置========"); Ok(base_config) @@ -296,7 +298,8 @@ pub fn read_config_file(config_path: PathBuf) -> Result { #[cfg(any(unix))] off_command_server: args_config.off_command_server, log, - password:args_config.password + password:args_config.password, + simulate_multicast:args_config.simulate_multicast }; println!("========参数配置========"); Ok(base_config) @@ -335,6 +338,8 @@ pub struct ArgsConfig { #[serde(default = "default_false")] pub log: bool, pub password: Option, + #[serde(default = "default_false")] + pub simulate_multicast:bool, } #[cfg(windows)] @@ -360,6 +365,7 @@ impl ArgsConfig { #[cfg(any(unix))] off_command_server: start_config.off_command_server, password: start_config.password, + simulate_multicast:start_config.simulate_multicast, } } } diff --git a/switch-desktop/src/i18n.rs b/switch-desktop/src/i18n.rs index d26ce98..9386b1e 100644 --- a/switch-desktop/src/i18n.rs +++ b/switch-desktop/src/i18n.rs @@ -76,6 +76,9 @@ pub fn switch_out_ip_help() -> String { pub fn switch_password_help() -> String { rust_i18n::t!("switch_password_help") } +pub fn switch_simulate_multicast_help() -> String { + rust_i18n::t!("switch_simulate_multicast_help") +} pub fn switch_config_help() -> String { rust_i18n::t!("switch_config_help") diff --git a/switch-desktop/src/main.rs b/switch-desktop/src/main.rs index 52ceac6..60514d2 100644 --- a/switch-desktop/src/main.rs +++ b/switch-desktop/src/main.rs @@ -110,6 +110,9 @@ pub struct StartArgs { /// 客户端数据加密 #[arg(long)] password:Option, + /// 模拟组播，默认情况下组播数据会被当作广播发送，兼容性更强，但是会造成流量浪费，开启后会模拟真实组播的数据发送 + #[arg(long)] + simulate_multicast:bool, /// 读取配置文件 --config config_file_path /// Read configuration file #[arg(long)] diff --git a/switch-desktop/src/unix/mod.rs b/switch-desktop/src/unix/mod.rs index e588090..5f75f43 100644 --- a/switch-desktop/src/unix/mod.rs +++ b/switch-desktop/src/unix/mod.rs @@ -46,6 +46,7 @@ pub async fn main0(base_args: BaseArgs) { start_config.in_ips.clone(), start_config.out_ips.clone(), start_config.password.clone(), + start_config.simulate_multicast, ); let lock = match config::lock_file() { Ok(lock) => { diff --git a/switch-desktop/src/windows/mod.rs b/switch-desktop/src/windows/mod.rs index 125b483..96dcd0e 100644 --- a/switch-desktop/src/windows/mod.rs +++ b/switch-desktop/src/windows/mod.rs @@ -119,6 +119,7 @@ pub fn main0(base_args: BaseArgs) { start_config.in_ips, start_config.out_ips, start_config.password, + start_config.simulate_multicast, ); let lock = match config::lock_file() { Ok(lock) => { diff --git a/switch-desktop/src/windows/service.rs b/switch-desktop/src/windows/service.rs index 0817199..24de78e 100644 --- a/switch-desktop/src/windows/service.rs +++ b/switch-desktop/src/windows/service.rs @@ -171,6 +171,7 @@ async fn start_switch(arguments: Vec) -> switch::Result<()> { start_config.in_ips, start_config.out_ips, start_config.password, + start_config.simulate_multicast, ); log::info!("switch-service服务启动"); diff --git a/switch/Cargo.toml b/switch/Cargo.toml index 8175ae7..6ee23b9 100644 --- a/switch/Cargo.toml +++ b/switch/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "switch" -version = "1.0.6" +version = "1.0.7" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -10,21 +10,19 @@ packet = { path = "./packet" } bytes = "1.3.0" log = "0.4.17" libc = "0.2.137" - -dashmap = "5.4.0" -crossbeam = "0.8.2" +crossbeam-utils = "0.8" crossbeam-skiplist = "0.1" parking_lot = "0.12.1" -rsa = "0.7.2" +#rsa = "0.7.2" rand = "0.8.5" sha2 = { version = "0.10.6", features = ["oid"] } aes-gcm = "0.10.2" thiserror = "1.0.37" -chrono = "0.4.23" +#chrono = "0.4.23" #lazy_static = "1.4.0" -moka = "0.9.6" +#moka = "0.9.6" protobuf = "3.2.0" #local-ip-address = "0.4.9" socket2 ={ version = "0.5.2", features = ["all"] } diff --git a/switch/src/channel/channel.rs b/switch/src/channel/channel.rs index f0220af..0075edf 100644 --- a/switch/src/channel/channel.rs +++ b/switch/src/channel/channel.rs @@ -1,27 +1,32 @@ use std::io; use std::net::{Ipv4Addr, SocketAddr}; use std::sync::Arc; -use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Instant; use crossbeam_skiplist::SkipMap; -use dashmap::DashMap; +use crossbeam_utils::atomic::AtomicCell; +use parking_lot::Mutex; use tokio::net::UdpSocket; -use tokio::sync::Notify; use tokio::sync::watch::{channel, Receiver, Sender}; use crate::channel::{Route, RouteKey, Status}; use crate::channel::punch::NatType; use crate::handle::recv_handler::ChannelDataHandler; +pub struct ContextInner { + pub(crate) lock:Mutex<()>, + pub(crate) count: AtomicUsize, + pub(crate) main_channel: Arc, + pub(crate) route_table: SkipMap>, + pub(crate) route_table_time: SkipMap<(RouteKey, Ipv4Addr), AtomicCell>, + pub(crate) status_receiver: Receiver, + pub(crate) status_sender: Sender, + pub(crate) udp_map: SkipMap>, + pub(crate) channel_num: usize, +} + #[derive(Clone)] pub struct Context { - pub(crate) count: Arc, - pub(crate) main_channel: Arc, - pub(crate) route_table: Arc>>, - pub(crate) route_table_time: Arc>, - pub(crate) status_receiver: Receiver, - pub(crate) status_sender: Arc>, - pub(crate) udp_map: Arc>>, - pub(crate) channel_num: usize, - pub(crate) notify: Arc, + pub(crate) inner: Arc, } impl Context { @@ -29,30 +34,32 @@ impl Context { //当前版本只支持一个通道 let channel_num = 1; let (status_sender, status_receiver) = channel(Status::Cone); - let status_sender = Arc::new(status_sender); - Self { - count: Arc::new(AtomicUsize::new(0)), + let inner = Arc::new(ContextInner { + lock:Mutex::new(()), + count: AtomicUsize::new(0), main_channel, - route_table: Arc::new(DashMap::with_capacity(16)), - route_table_time: Arc::new(SkipMap::new()), + route_table: SkipMap::new(), + route_table_time: SkipMap::new(), status_receiver, status_sender, - udp_map: Arc::new(SkipMap::new()), + udp_map: SkipMap::new(), channel_num, - notify: Arc::new(Notify::new()), + }); + Self { + inner } } } impl Context { pub fn is_close(&self) -> bool { - *self.status_receiver.borrow() == Status::Close + *self.inner.status_receiver.borrow() == Status::Close } pub fn is_cone(&self) -> bool { - *self.status_receiver.borrow() == Status::Cone + *self.inner.status_receiver.borrow() == Status::Cone } pub fn close(&self) { - let _ = self.status_sender.send(Status::Close); + let _ = self.inner.status_sender.send(Status::Close); } pub fn switch(&self, nat_type: NatType) { match nat_type { @@ -65,62 +72,64 @@ impl Context { } } pub fn switch_to_cone(&self) { - let _ = self.status_sender.send(Status::Cone); + let _ = self.inner.status_sender.send(Status::Cone); } pub fn switch_to_symmetric(&self) { - let _ = self.status_sender.send(Status::Symmetric); + let _ = self.inner.status_sender.send(Status::Symmetric); } pub fn main_local_port(&self) -> io::Result { - self.main_channel.local_addr().map(|k| k.port()) + self.inner.main_channel.local_addr().map(|k| k.port()) } pub async fn send_main(&self, buf: &[u8], addr: SocketAddr) -> io::Result { - self.main_channel.send_to(buf, addr).await + self.inner.main_channel.send_to(buf, addr).await } pub(crate) async fn send_all(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> { - for udp in self.udp_map.iter() { + for udp in self.inner.udp_map.iter() { udp.value().send_to(buf, addr).await?; } Ok(()) } pub fn try_send_main(&self, buf: &[u8], addr: SocketAddr) -> io::Result { - self.main_channel.try_send_to(buf, addr) + self.inner.main_channel.try_send_to(buf, addr) } pub async fn send_by_id(&self, buf: &[u8], id: &Ipv4Addr) -> io::Result { - if let Some(v) = self.route_table.get(id) { - let route = match v.len() { + if let Some(v) = self.inner.route_table.get(id) { + let route = match v.value().len() { 0 => { return Err(io::Error::new(io::ErrorKind::NotFound, "route not found")); } - 1 => &v[0], - len => &v[self.count.fetch_add(1, Ordering::Relaxed) % len] + 1 => v.value()[0], + len => v.value()[self.inner.count.fetch_add(1, Ordering::Relaxed) % len] }; - if let Some(udp) = self.udp_map.get(&route.index) { + drop(v); + if let Some(udp) = self.inner.udp_map.get(&route.index) { return udp.value().send_to(buf, route.addr).await; } } Err(io::Error::new(io::ErrorKind::NotFound, "route not found")) } pub fn try_send_by_id(&self, buf: &[u8], id: &Ipv4Addr) -> io::Result { - if let Some(v) = self.route_table.get(id) { - if v.is_empty() { + if let Some(v) = self.inner.route_table.get(id) { + if v.value().is_empty() { return Err(io::Error::new(io::ErrorKind::NotFound, "route not found")); } - let route = &v[self.count.fetch_add(1, Ordering::Relaxed) % v.len()]; - if let Some(udp) = self.udp_map.get(&route.index) { + let route = v.value()[self.inner.count.fetch_add(1, Ordering::Relaxed) % v.value().len()]; + drop(v); + if let Some(udp) = self.inner.udp_map.get(&route.index) { return udp.value().try_send_to(buf, route.addr); } } Err(io::Error::new(io::ErrorKind::NotFound, "route not found")) } pub async fn send_by_key(&self, buf: &[u8], route_key: &RouteKey) -> io::Result { - if let Some(udp) = self.udp_map.get(&route_key.index) { + if let Some(udp) = self.inner.udp_map.get(&route_key.index) { return udp.value().send_to(buf, route_key.addr).await; } Err(io::Error::new(io::ErrorKind::NotFound, "route not found")) } pub fn try_send_by_key(&self, buf: &[u8], route_key: &RouteKey) -> io::Result { - if let Some(udp) = self.udp_map.get(&route_key.index) { + if let Some(udp) = self.inner.udp_map.get(&route_key.index) { return udp.value().try_send_to(buf, route_key.addr); } Err(io::Error::new(io::ErrorKind::NotFound, "route not found")) @@ -133,9 +142,14 @@ impl Context { } fn add_route_(&self, id: Ipv4Addr, route: Route, only_if_absent: bool) { let key = route.route_key(); - let mut ref_mut = self.route_table.entry(id.clone()).or_insert(Vec::with_capacity(4)); + let guard = self.inner.lock.lock(); + let mut list = if let Some(entry) = self.inner.route_table.get(&id){ + entry.value().clone() + }else{ + Vec::with_capacity(4) + }; let mut exist = false; - for x in ref_mut.iter_mut() { + for x in list.iter_mut() { if x.metric < route.metric { //不能比当前的路径更长 return; @@ -150,37 +164,40 @@ impl Context { break; } } - if !exist { + if exist { + list.sort_by_key(|k| k.sort_key()); + } else { if route.metric == 1 { //添加了直连的则排除非直连的 - ref_mut.retain(|k| k.metric == 1); + list.retain(|k| k.metric == 1); } - ref_mut.push(route); - let max_len = self.channel_num; - if ref_mut.len() > max_len { - ref_mut.sort_by_key(|k| k.sort_key()); - ref_mut.truncate(max_len); + list.push(route); + 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); } } - self.route_table_time.insert((key, id), AtomicI64::new(chrono::Local::now().timestamp_millis())); - self.notify.notify_one(); + self.inner.route_table.insert(id,list); + self.inner.route_table_time.insert((key, id), AtomicCell::new(Instant::now())); + drop(guard); } pub fn route(&self, id: &Ipv4Addr) -> Option> { - if let Some(v) = self.route_table.get(id) { + if let Some(v) = self.inner.route_table.get(id) { Some(v.value().clone()) } else { None } } pub fn route_one(&self, id: &Ipv4Addr) -> Option { - if let Some(v) = self.route_table.get(id) { + if let Some(v) = self.inner.route_table.get(id) { v.value().iter().max_by_key(|k| k.sort_key()).map(|k| *k) } else { None } } pub fn route_to_id(&self, route_key: &RouteKey) -> Option { - for x in self.route_table_time.iter() { + for x in self.inner.route_table_time.iter() { if &x.key().0 == route_key { return Some(x.key().1); } @@ -188,20 +205,20 @@ impl Context { None } pub fn need_punch(&self, id: &Ipv4Addr) -> bool { - if let Some(v) = self.route_table.get(id) { - if v.iter().filter(|k| k.is_p2p()).count() >= self.channel_num { + if let Some(v) = self.inner.route_table.get(id) { + if v.value().iter().filter(|k| k.is_p2p()).count() >= self.inner.channel_num { return false; } } true } pub fn route_table(&self) -> Vec<(Ipv4Addr, Vec)> { - self.route_table.iter().map(|k| (k.key().clone(), k.value().clone())).collect() + self.inner.route_table.iter().map(|k| (k.key().clone(), k.value().clone())).collect() } pub fn route_table_one(&self) -> Vec<(Ipv4Addr, Route)> { let mut v = Vec::with_capacity(8); - for x in self.route_table.iter() { - if let Some(route) = x.value().iter().max_by_key(|k| k.sort_key()) { + for x in self.inner.route_table.iter() { + if let Some(route) = x.value().first() { v.push((*x.key(), *route)); } } @@ -209,8 +226,8 @@ impl Context { } pub fn direct_route_table_one(&self) -> Vec<(Ipv4Addr, Route)> { let mut v = Vec::with_capacity(8); - for x in self.route_table.iter() { - if let Some(route) = x.value().iter().max_by_key(|k| k.sort_key()) { + for x in self.inner.route_table.iter() { + if let Some(route) = x.value().first() { if route.metric == 1 { v.push((*x.key(), *route)); } @@ -219,21 +236,28 @@ impl Context { v } pub fn remove_route_all(&self, id: &Ipv4Addr) { - if let Some((_, v)) = self.route_table.remove(id) { - for x in v { - self.route_table_time.remove(&(x.route_key(), id.clone())); + let guard = self.inner.lock.lock(); + if let Some(v) = self.inner.route_table.remove(id) { + for x in v.value() { + self.inner.route_table_time.remove(&(x.route_key(), *id)); } } + drop(guard); } pub fn remove_route(&self, id: &Ipv4Addr, route_key: RouteKey) { - if let Some(mut v) = self.route_table.get_mut(id) { - v.retain(|x| x.route_key() != route_key); - self.route_table_time.remove(&(route_key, id.clone())); + let guard = self.inner.lock.lock(); + if let Some(v) = self.inner.route_table.get(id) { + let mut routes = v.value().clone(); + drop(v); + routes.retain(|x| x.route_key() != route_key); + self.inner.route_table.insert(*id,routes); + self.inner.route_table_time.remove(&(route_key, *id)); } + drop(guard); } pub fn update_read_time(&self, id: &Ipv4Addr, route_key: &RouteKey) { - if let Some(time) = self.route_table_time.get(&(*route_key, *id)) { - time.value().store(chrono::Local::now().timestamp_millis(), Ordering::Relaxed); + if let Some(time) = self.inner.route_table_time.get(&(*route_key, *id)) { + time.value().store(Instant::now()); } } } @@ -273,15 +297,17 @@ impl Channel { head_reserve: usize,//头部预留字节 symmetric_channel_num: usize,//对称网络，则再加一组监听，提升打洞成功率 ) { - let mut context = self.context; - let main_channel = context.main_channel.clone(); + let context = self.context; + let main_channel = context.inner.main_channel.clone(); let handler = self.handler.clone(); + tokio::spawn(Self::start_(context.clone(), handler.clone(), main_channel.clone(), head_reserve, true)); tokio::spawn(Self::start_(context.clone(), handler, main_channel, head_reserve, true)); let mut cur_status = Status::Cone; + let mut status_receiver = context.inner.status_receiver.clone(); loop { - match context.status_receiver.changed().await { + match status_receiver.changed().await { Ok(_) => { - match *context.status_receiver.borrow() { + match *status_receiver.borrow() { Status::Cone => { cur_status = Status::Cone; } @@ -320,7 +346,7 @@ impl Channel { udp: Arc, head_reserve: usize, is_core: bool) { - let mut status_receiver = context.status_receiver.clone(); + let mut status_receiver = context.inner.status_receiver.clone(); #[cfg(target_os = "windows")] use std::os::windows::io::AsRawSocket; #[cfg(target_os = "windows")] @@ -329,8 +355,8 @@ impl Channel { use std::os::fd::AsRawFd; #[cfg(any(unix))] let id = udp.as_raw_fd() as usize; - context.udp_map.insert(id, udp.clone()); - let mut buf = [0; 65536]; + context.inner.udp_map.insert(id, udp.clone()); + let mut buf = [0; 4096]; loop { tokio::select! { rs=udp.recv_from(&mut buf[head_reserve..])=>{ @@ -358,41 +384,6 @@ impl Channel { } } } - context.udp_map.remove(&id); + context.inner.udp_map.remove(&id); } } - - -// pub async fn start(mut handler: H, -// mut status_receiver: Receiver, -// head_reserve: usize, -// core_channel_num: usize, -// symmetric_channel_num: usize) -> io::Result<()> { -// for _ in 0..core_channel_num { -// let d = channel(1); -// } -// let udp = UdpSocket::bind("0.0.0.0:0").await?; -// let d = status_receiver.changed().await; -// match d { -// Ok(_) => { -// match *status_receiver.borrow() { -// Status::Cone => {} -// Status::Symmetric => {} -// Status::Close => {} -// } -// } -// Err(_) => {} -// } -// let mut buf = [0; 65546]; -// let result = udp.recv_from(&mut buf[head_reserve..]).await; -// match result { -// Ok((len, addr)) => {} -// Err(e) => {} -// } -// Ok(()) -// } -// -// pub struct Channel { -// handler: H, -// -// } \ No newline at end of file diff --git a/switch/src/channel/idle.rs b/switch/src/channel/idle.rs index 46584b4..80010ea 100644 --- a/switch/src/channel/idle.rs +++ b/switch/src/channel/idle.rs @@ -1,19 +1,18 @@ 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: i64, + read_idle: Duration, context: Context, } impl Idle { - pub fn new(read_idle: i64, + pub fn new(read_idle: Duration, context: Context, ) -> Self { Self { read_idle, @@ -26,34 +25,19 @@ impl Idle { /// 获取空闲路由 pub async fn next_idle(&self) -> io::Result<(Ipv4Addr, RouteKey)> { loop { - let now = chrono::Local::now().timestamp_millis(); - let last_read_idle = now - self.read_idle; - let mut min = i64::MAX; - for entry in self.context.route_table_time.iter() { - let mut is_read_idle = false; - if self.read_idle > 0 { - let last_read = entry.value().load(Ordering::Relaxed); - if last_read < last_read_idle { - is_read_idle = true; - } else { - if min > last_read { - min = last_read; - } + let mut max = Duration::from_secs(10); + for entry in self.context.inner.route_table_time.iter() { + let last_read = entry.value().load().elapsed(); + if last_read >= self.read_idle { + return Ok((entry.key().1.clone(), entry.key().0.clone())); + } else { + if max < last_read { + max = last_read; } } - if is_read_idle { - return Ok((entry.key().1.clone(), entry.key().0.clone())); - } - } - if self.context.route_table_time.is_empty() { - self.context.notify.notified().await; - } else { - let sleep_time = chrono::Local::now().timestamp_millis() - min; - if sleep_time > 0 { - tokio::time::sleep(Duration::from_millis(sleep_time as u64)).await; - // let _ = tokio::time::timeout(Duration::from_millis(sleep_time as u64), self.context.notify.notified()).await; - } } + let sleep_time = self.read_idle - max; + tokio::time::sleep(sleep_time).await; if self.context.is_close() { return Err(Error::new(ErrorKind::Other, "closed")); } diff --git a/switch/src/core/mod.rs b/switch/src/core/mod.rs index efed388..cf4c831 100644 --- a/switch/src/core/mod.rs +++ b/switch/src/core/mod.rs @@ -1,9 +1,10 @@ use std::{io, thread}; use std::net::{Ipv4Addr, SocketAddr}; use std::sync::Arc; +use std::time::Duration; use aes_gcm::{Aes256Gcm, Key, KeyInit}; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use crossbeam_skiplist::SkipMap; use parking_lot::Mutex; use tokio::net::UdpSocket; @@ -54,7 +55,7 @@ impl Switch { let (symmetric_sender, symmetric_receiver) = channel(2); let context = Context::new(main_channel, 1); let punch = Punch::new(context.clone()); - let idle = Idle::new(16000, context.clone()); + let idle = Idle::new(Duration::from_secs(16), context.clone()); let channel_sender = ChannelSender::new(context.clone()); let register = Arc::new(registration_handler::Register::new(channel_sender.clone(), config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone())); @@ -73,9 +74,17 @@ impl Switch { let out_ips = config.out_ips.iter().map(|(_, _, ip)| *ip).collect::>(); let out_external_route = ExternalRoute::new(config.out_ips); - let in_external_route = ExternalRoute::new(config.in_ips); + let in_external_route = if config.in_ips.is_empty() { + None + } else { + Some(ExternalRoute::new(config.in_ips)) + }; let current_device = Arc::new(AtomicCell::new(CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, config.server_address))); - let ip_proxy_map = crate::ip_proxy::init_proxy(channel_sender.clone(), out_ips, current_device.clone()).await?; + let ip_proxy_map = if out_ips.is_empty(){ + None + }else{ + Some(crate::ip_proxy::init_proxy(channel_sender.clone(), out_ips, current_device.clone()).await?) + }; let (device_writer, igmp_server) = if config.tap { #[cfg(windows)] { @@ -83,7 +92,11 @@ impl Switch { tun_tap_device::delete_device(tun_tap_device::DeviceType::Tap); } let (tap_writer, tap_reader) = tun_tap_device::create_device(tun_tap_device::DeviceType::Tap, virtual_ip, virtual_netmask, virtual_gateway, in_ips)?; - let igmp_server = IgmpServer::new(tap_writer.clone()); + let igmp_server = if config.simulate_multicast { + Some(IgmpServer::new(tap_writer.clone())) + } else { + None + }; //tap数据处理 tap_handler::start(channel_sender.clone(), tap_reader.clone(), tap_writer.clone(), igmp_server.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone(), cipher.clone()); @@ -96,7 +109,11 @@ impl Switch { } // tun通道 let (tun_writer, tun_reader) = tun_tap_device::create_device(tun_tap_device::DeviceType::Tun, virtual_ip, virtual_netmask, virtual_gateway, in_ips)?; - let igmp_server = IgmpServer::new(tun_writer.clone()); + let igmp_server = if config.simulate_multicast { + Some(IgmpServer::new(tun_writer.clone())) + } else { + None + }; //tun数据接收处理 tun_handler::start(channel_sender.clone(), tun_reader.clone(), tun_writer.clone(), igmp_server.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone(), cipher.clone()); @@ -188,6 +205,7 @@ pub struct Config { pub in_ips: Vec<(u32, u32, Ipv4Addr)>, pub out_ips: Vec<(u32, u32, Ipv4Addr)>, pub key: Option<[u8; 32]>, + pub simulate_multicast: bool, } use sha2::Digest; @@ -198,7 +216,8 @@ impl Config { name: String, server_address: SocketAddr, nat_test_server: Vec, - in_ips: Vec<(u32, u32, Ipv4Addr)>, out_ips: Vec<(u32, u32, Ipv4Addr)>, password: Option, ) -> Self { + in_ips: Vec<(u32, u32, Ipv4Addr)>, out_ips: Vec<(u32, u32, Ipv4Addr)>, + password: Option, simulate_multicast: bool, ) -> Self { let key = if let Some(password) = password { let mut hasher = sha2::Sha256::new(); hasher.update(password.as_bytes()); @@ -217,6 +236,7 @@ impl Config { in_ips, out_ips, key, + simulate_multicast, } } } \ No newline at end of file diff --git a/switch/src/error/mod.rs b/switch/src/error/mod.rs index 889a925..f7032aa 100644 --- a/switch/src/error/mod.rs +++ b/switch/src/error/mod.rs @@ -1,14 +1,11 @@ use std::io; -use crossbeam::channel::RecvError; use thiserror::Error; #[derive(Error, Debug)] pub enum Error { #[error("Io error")] Io(#[from] io::Error), - #[error("Channel error")] - Channel(#[from] RecvError), #[error("Protobuf error")] Protobuf(#[from] protobuf::Error), #[error("Invalid packet")] diff --git a/switch/src/handle/heartbeat_handler.rs b/switch/src/handle/heartbeat_handler.rs index 8496832..44a36ff 100644 --- a/switch/src/handle/heartbeat_handler.rs +++ b/switch/src/handle/heartbeat_handler.rs @@ -3,8 +3,7 @@ use std::sync::Arc; use std::time::Duration; use std::io; -use chrono::Local; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use parking_lot::Mutex; use rand::prelude::SliceRandom; use crate::channel::idle::Idle; @@ -52,7 +51,7 @@ pub async fn start_heartbeat( } fn set_now_time(packet: &mut NetPacket<[u8; 16]>) -> io::Result<()> { - let current_time = Local::now().timestamp_millis() as u16; + let current_time = crate::handle::now_time() as u16; let mut ping = PingPacket::new(packet.payload_mut())?; ping.set_time(current_time); Ok(()) @@ -90,40 +89,46 @@ async fn start_heartbeat_( } if count < 7 || count % 7 == 0 { let mut route_list: Option)>> = None; - let peer_list = {device_list.lock().1.clone()}; + let peer_list = { device_list.lock().1.clone() }; for peer in peer_list { + if peer.virtual_ip == current_device.virtual_ip { + continue; + } set_now_time(&mut net_packet)?; net_packet.set_destination(peer.virtual_ip); - if sender - .send_by_id(net_packet.buffer(), &peer.virtual_ip).await - .is_err() - { - //没有路由则发送到网关 + if let Some(route) = sender.route_one(&peer.virtual_ip) { + let _ = sender.send_by_key(net_packet.buffer(), &route.route_key()).await; + if route.is_p2p() { + continue; + } + } else { + //没有直连路由则发送到网关 let _ = sender.try_send_main(net_packet.buffer(), current_device.connect_server); - //再随机发送到其他地址，看有没有客户端符合转发条件 - let route_list = route_list.get_or_insert_with(|| { - let mut l = sender.route_table(); - l.shuffle(&mut rand::thread_rng()); - l - }); - let mut num = 0; - 'a: for (peer_ip, route_list) in route_list.iter() { - for route in route_list { - if peer_ip != &peer.virtual_ip && route.metric == 1 { - set_now_time(&mut net_packet)?; - let _ = sender.try_send_by_key(net_packet.buffer(), &route.route_key()); - num += 1; - break; - } - if num >= 3 { - break 'a; - } + continue; + } + + //再随机发送到其他地址，看有没有客户端符合转发条件 + let route_list = route_list.get_or_insert_with(|| { + let mut l = sender.route_table(); + l.shuffle(&mut rand::thread_rng()); + l + }); + let mut num = 0; + 'a: for (peer_ip, route_list) in route_list.iter() { + for route in route_list { + if peer_ip != &peer.virtual_ip && route.is_p2p() { + set_now_time(&mut net_packet)?; + let _ = sender.try_send_by_key(net_packet.buffer(), &route.route_key()); + num += 1; + break; + } + if num >= 3 { + break 'a; } } } tokio::time::sleep(Duration::from_millis(1)).await; } - } else { for (peer_ip, route_list) in sender.route_table().iter() { set_now_time(&mut net_packet)?; diff --git a/switch/src/handle/mod.rs b/switch/src/handle/mod.rs index f20651b..19bc083 100644 --- a/switch/src/handle/mod.rs +++ b/switch/src/handle/mod.rs @@ -6,6 +6,15 @@ pub mod recv_handler; pub mod registration_handler; pub mod tun_tap; +pub fn now_time() -> u64 { + let now = std::time::SystemTime::now(); + if let Ok(timestamp) = now.duration_since(std::time::UNIX_EPOCH) { + timestamp.as_secs() * 1000 + u64::from(timestamp.subsec_millis()) + } else { + 0 + } +} + /// 是否在一个网段 fn check_dest(dest: Ipv4Addr, virtual_netmask: Ipv4Addr, virtual_network: Ipv4Addr) -> bool { u32::from_be_bytes(dest.octets()) & u32::from_be_bytes(virtual_netmask.octets()) diff --git a/switch/src/handle/punch_handler.rs b/switch/src/handle/punch_handler.rs index 38cd179..eec450f 100644 --- a/switch/src/handle/punch_handler.rs +++ b/switch/src/handle/punch_handler.rs @@ -2,7 +2,7 @@ use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo}; use crate::nat::NatTest; use crate::proto::message::{PunchInfo, PunchNatType}; use crate::protocol::{control_packet, other_turn_packet, NetPacket, Protocol, Version, MAX_TTL}; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use parking_lot::Mutex; use protobuf::Message; use rand::prelude::SliceRandom; @@ -89,7 +89,7 @@ async fn start_punch_( break; } let buf = punch_packet(current_device.virtual_ip(), &nat_info, info.virtual_ip)?; - sender.send_main(&buf, current_device.connect_server).await?; + let _ = sender.send_main(&buf, current_device.connect_server).await; } } num += 1; diff --git a/switch/src/handle/recv_handler.rs b/switch/src/handle/recv_handler.rs index 8a56bb7..20b2d77 100644 --- a/switch/src/handle/recv_handler.rs +++ b/switch/src/handle/recv_handler.rs @@ -4,8 +4,7 @@ use aes_gcm::{AeadInPlace, Aes256Gcm, Nonce, Tag}; use aes_gcm::aead::consts::{U12, U16}; use aes_gcm::aead::generic_array::GenericArray; -use chrono::Local; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use crossbeam_skiplist::SkipMap; use parking_lot::Mutex; use protobuf::Message; @@ -40,11 +39,11 @@ pub struct ChannelDataHandler { device_list: Arc)>>, register: Arc, nat_test: NatTest, - igmp_server: IgmpServer, + igmp_server: Option, device_writer: DeviceWriter, connect_status: Arc>, peer_nat_info_map: Arc>, - ip_proxy_map: IpProxyMap, + ip_proxy_map: Option, out_external_route: ExternalRoute, cone_sender: Sender<(Ipv4Addr, NatInfo)>, symmetric_sender: Sender<(Ipv4Addr, NatInfo)>, @@ -56,11 +55,11 @@ impl ChannelDataHandler { device_list: Arc)>>, register: Arc, nat_test: NatTest, - igmp_server: IgmpServer, + igmp_server: Option, device_writer: DeviceWriter, connect_status: Arc>, peer_nat_info_map: Arc>, - ip_proxy_map: IpProxyMap, + ip_proxy_map: Option, out_external_route: ExternalRoute, cone_sender: Sender<(Ipv4Addr, NatInfo)>, symmetric_sender: Sender<(Ipv4Addr, NatInfo)>, @@ -103,8 +102,9 @@ impl ChannelDataHandler { let source = net_packet.source(); let current_device = self.current_device.load(); let destination = net_packet.destination(); + let not_broadcast = !destination.is_broadcast() && !destination.is_multicast() && destination != current_device.broadcast_address; if current_device.virtual_ip() != destination - && !destination.is_broadcast() && !destination.is_multicast() && destination != current_device.broadcast_address + && not_broadcast && self.connect_status.load() == ConnectStatus::Connected { if !check_dest(source, current_device.virtual_netmask, current_device.virtual_network) { log::warn!("转发数据，源地址错误:{:?},当前网络:{:?},route_key:{:?}",source,current_device.virtual_network,route_key); @@ -141,9 +141,11 @@ impl ChannelDataHandler { } } ip_turn_packet::Protocol::Igmp => { - let ipv4 = IpV4Packet::new(net_packet.payload())?; - if ipv4.protocol() == ipv4::protocol::Protocol::Igmp { - self.igmp_server.handle(ipv4.payload(), source)?; + if let Some(igmp_server) = &self.igmp_server { + let ipv4 = IpV4Packet::new(net_packet.payload())?; + if ipv4.protocol() == ipv4::protocol::Protocol::Igmp { + igmp_server.handle(ipv4.payload(), source)?; + } } return Ok(()); } @@ -180,7 +182,9 @@ impl ChannelDataHandler { let mut ipv4 = IpV4Packet::new(data)?; match ipv4.protocol() { ipv4::protocol::Protocol::Igmp => { - self.igmp_server.handle(ipv4.payload(), source)?; + if let Some(igmp_server) = &self.igmp_server { + igmp_server.handle(ipv4.payload(), source)?; + } return Ok(()); } ipv4::protocol::Protocol::Icmp => { @@ -226,51 +230,53 @@ impl ChannelDataHandler { } _ => {} } - if ipv4.destination_ip() != destination { - if let Some(gate_way) = self.out_external_route.route(&ipv4.destination_ip()) { - match ipv4.protocol() { - ipv4::protocol::Protocol::Tcp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source, destination, ipv4.payload_mut())?; - let source_port = tcp_packet.source_port(); - let dest_port = tcp_packet.destination_port(); - tcp_packet.set_destination_port(self.ip_proxy_map.tcp_proxy_port); - tcp_packet.update_checksum(); - ipv4.set_destination_ip(destination); - ipv4.update_checksum(); - self.ip_proxy_map.tcp_proxy_map.insert(SocketAddrV4::new(source, source_port), - (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port))); - } - ipv4::protocol::Protocol::Udp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let mut udp_packet = packet::udp::udp::UdpPacket::new(source, destination, ipv4.payload_mut())?; - let source_port = udp_packet.source_port(); - let dest_port = udp_packet.destination_port(); - udp_packet.set_destination_port(self.ip_proxy_map.udp_proxy_port); - udp_packet.update_checksum(); - ipv4.set_destination_ip(destination); - ipv4.update_checksum(); - self.ip_proxy_map.udp_proxy_map.insert(SocketAddrV4::new(source, source_port), - (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port))); - } - ipv4::protocol::Protocol::Icmp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?; - match icmp_packet.header_other() { - HeaderOther::Identifier(id, seq) => { - self.ip_proxy_map.icmp_proxy_map.insert((dest_ip, id, seq), source); - self.ip_proxy_map.send_icmp(ipv4.payload(), &gate_way, &dest_ip)?; - } - _ => { - return Ok(()); + if not_broadcast && ipv4.destination_ip() != destination { + if let Some(ip_proxy_map) = &self.ip_proxy_map { + if let Some(gate_way) = self.out_external_route.route(&ipv4.destination_ip()) { + match ipv4.protocol() { + ipv4::protocol::Protocol::Tcp => { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source, destination, ipv4.payload_mut())?; + let source_port = tcp_packet.source_port(); + let dest_port = tcp_packet.destination_port(); + tcp_packet.set_destination_port(ip_proxy_map.tcp_proxy_port); + tcp_packet.update_checksum(); + ipv4.set_destination_ip(destination); + ipv4.update_checksum(); + ip_proxy_map.tcp_proxy_map.insert(SocketAddrV4::new(source, source_port), + (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port))); + } + ipv4::protocol::Protocol::Udp => { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let mut udp_packet = packet::udp::udp::UdpPacket::new(source, destination, ipv4.payload_mut())?; + let source_port = udp_packet.source_port(); + let dest_port = udp_packet.destination_port(); + udp_packet.set_destination_port(ip_proxy_map.udp_proxy_port); + udp_packet.update_checksum(); + ipv4.set_destination_ip(destination); + ipv4.update_checksum(); + ip_proxy_map.udp_proxy_map.insert(SocketAddrV4::new(source, source_port), + (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port))); + } + ipv4::protocol::Protocol::Icmp => { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?; + match icmp_packet.header_other() { + HeaderOther::Identifier(id, seq) => { + ip_proxy_map.icmp_proxy_map.insert((dest_ip, id, seq), source); + ip_proxy_map.send_icmp(ipv4.payload(), &gate_way, &dest_ip)?; + } + _ => { + return Ok(()); + } } } - } - _ => { - return Ok(()); + _ => { + return Ok(()); + } } } } @@ -406,7 +412,7 @@ impl ChannelDataHandler { } ControlPacket::PongPacket(pong_packet) => { context.update_read_time(&source, route_key); - let current_time = Local::now().timestamp_millis() as u16; + let current_time = crate::handle::now_time() as u16; if current_time < pong_packet.time() { return Ok(()); } diff --git a/switch/src/handle/registration_handler.rs b/switch/src/handle/registration_handler.rs index 3ac6c5a..8a796a1 100644 --- a/switch/src/handle/registration_handler.rs +++ b/switch/src/handle/registration_handler.rs @@ -1,9 +1,8 @@ use std::io; use std::net::SocketAddr; -use std::sync::atomic::{AtomicI64, Ordering}; -use std::time::Duration; +use std::time::{Duration, Instant}; +use crossbeam_utils::atomic::AtomicCell; -use chrono::Local; use protobuf::Message; use tokio::net::UdpSocket; use crate::channel::sender::ChannelSender; @@ -105,7 +104,7 @@ fn registration_request_packet( request.device_id = device_id; request.name = name; request.is_fast = is_fast; - request.version = "1.0.6".to_string(); + request.version = "1.0.7".to_string(); let bytes = request.write_to_bytes()?; let buf = vec![0u8; 12 + bytes.len()]; let mut net_packet = NetPacket::new(buf)?; @@ -123,7 +122,7 @@ pub struct Register { token: String, device_id: String, name: String, - time: AtomicI64, + time: AtomicCell, } impl Register { @@ -140,16 +139,15 @@ impl Register { token, device_id, name, - time: AtomicI64::new(0), + time: AtomicCell::new(Instant::now()), } } pub async fn fast_register(&self) -> io::Result<()> { - let last = self.time.load(Ordering::Relaxed); - let new = Local::now().timestamp_millis(); - if new - last < 1000 + let last = self.time.load(); + if last.elapsed() < Duration::from_secs(2) || self .time - .compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed) + .compare_exchange(last, Instant::now()) .is_err() { //短时间不重复注册 diff --git a/switch/src/handle/tun_tap/mod.rs b/switch/src/handle/tun_tap/mod.rs index 36725e5..0e001dc 100644 --- a/switch/src/handle/tun_tap/mod.rs +++ b/switch/src/handle/tun_tap/mod.rs @@ -22,11 +22,36 @@ pub mod tap_handler; async fn broadcast(sender: &ChannelSender, net_packet: &mut NetPacket<&mut [u8]>, data_len: usize, current_device: &CurrentDeviceInfo) -> Result<()> { let mut peer_ips = Vec::with_capacity(8); - let vec = sender.direct_route_table_one(); + let vec = sender.route_table_one(); + let mut relay_count = 0; + let mut last_peer = None; for (peer_ip, route) in vec { - if sender.send_by_key(&net_packet.buffer()[..data_len], &route.route_key()).await.is_ok() { - peer_ips.push(peer_ip); + if peer_ip == current_device.virtual_gateway { + continue; } + if peer_ips.len() < u8::MAX as usize && route.is_p2p() + && sender.send_by_key(&net_packet.buffer()[..data_len], &route.route_key()).await.is_ok() { + peer_ips.push(peer_ip); + } else { + relay_count += 1; + if relay_count == 1 { + last_peer = Some((peer_ip, route)); + } + if relay_count > 1 && peer_ips.len() == u8::MAX as usize { + break; + } + } + } + if relay_count == 0 && !peer_ips.is_empty() { + //不需要转发 + return Ok(()); + } + if relay_count == 1 && !net_packet.is_encrypt() { + //只有一个目标，并且没加密 + let (peer_ip, route) = last_peer.unwrap(); + net_packet.set_destination(peer_ip); + sender.send_by_key(&net_packet.buffer()[..data_len], &route.route_key()).await?; + return Ok(()); } if peer_ips.is_empty() { sender.send_main(&net_packet.buffer()[..data_len], current_device.connect_server).await?; @@ -44,20 +69,42 @@ async fn broadcast(sender: &ChannelSender, net_packet: &mut NetPacket<&mut [u8]> async fn multicast(igmp_server: &IgmpServer, multicast_addr: Ipv4Addr, sender: &ChannelSender, net_packet: &mut NetPacket<&mut [u8]>, data_len: usize, current_device: &CurrentDeviceInfo) -> Result<()> { let mut peer_ips = Vec::with_capacity(8); - let vec = sender.direct_route_table_one(); + let vec = sender.route_table_one(); + let mut relay_count = 0; + let mut last_peer = None; if let Some(members) = igmp_server.load(&multicast_addr) { for (peer_ip, route) in vec { - let is_send = {members.read().is_send(&peer_ip)}; + if peer_ip == current_device.virtual_gateway { + continue; + } + let is_send = { members.read().is_send(&peer_ip) }; if is_send { - if sender.send_by_key(&net_packet.buffer()[..data_len], &route.route_key()).await.is_ok() { + if peer_ips.len() < u8::MAX as usize && route.is_p2p() + && sender.send_by_key(&net_packet.buffer()[..data_len], &route.route_key()).await.is_ok() { peer_ips.push(peer_ip); - if peer_ips.len() == u8::MAX as usize { + } else { + relay_count += 1; + if relay_count == 1 { + last_peer = Some((peer_ip, route)); + } + if relay_count > 1 && peer_ips.len() == u8::MAX as usize { break; } } } } } + if relay_count == 0 && !peer_ips.is_empty() { + //不需要转发 + return Ok(()); + } + if relay_count == 1 && !net_packet.is_encrypt() { + //只有一个目标，并且没加密 + let (peer_ip, route) = last_peer.unwrap(); + net_packet.set_destination(peer_ip); + sender.send_by_key(&net_packet.buffer()[..data_len], &route.route_key()).await?; + return Ok(()); + } if peer_ips.is_empty() { sender.send_main(&net_packet.buffer()[..data_len], current_device.connect_server).await?; } else { @@ -78,9 +125,9 @@ async fn multicast(igmp_server: &IgmpServer, multicast_addr: Ipv4Addr, sender: & #[inline] pub async fn base_handle(sender: &ChannelSender, buf: &mut [u8], mut data_len: usize,//数据总长度=ip长度+12 - igmp_server: &IgmpServer, + igmp_server: &Option, current_device: CurrentDeviceInfo, - ip_route: &ExternalRoute, proxy_map: &IpProxyMap, cipher: &Option) -> Result<()> { + ip_route: &Option, proxy_map: &Option, cipher: &Option) -> Result<()> { let ipv4_packet = IpV4Packet::new(&buf[12..data_len])?; let protocol = ipv4_packet.protocol(); let ip_head_len = ipv4_packet.header_len() as usize * 4; @@ -102,11 +149,33 @@ pub async fn base_handle(sender: &ChannelSender, buf: &mut [u8], return Ok(()); } if dest_ip.is_multicast() { - if protocol == Protocol::Igmp { - net_packet.set_transport_protocol(ip_turn_packet::Protocol::Igmp.into()); - //发送到服务端 - sender.send_main(&net_packet.buffer()[..data_len], current_device.connect_server).await?; - return Ok(()); + match protocol { + Protocol::Igmp => { + if igmp_server.is_some() { + net_packet.set_transport_protocol(ip_turn_packet::Protocol::Igmp.into()); + //发送到服务端 + net_packet.set_destination(current_device.virtual_gateway); + sender.send_main(&net_packet.buffer()[..data_len], current_device.connect_server).await?; + } + return Ok(()); + } + Protocol::Udp => { + if let Some(igmp_server) = igmp_server { + if let Some(cipher) = cipher { + //需要加密 + encrypt(cipher, &mut data_len, &mut net_packet)?; + } + multicast(igmp_server, dest_ip, sender, &mut net_packet, data_len, ¤t_device).await?; + return Ok(()); + } else { + //当广播 + dest_ip = Ipv4Addr::BROADCAST; + net_packet.set_destination(dest_ip); + } + } + _ => { + return Ok(()); + } } } if dest_ip.is_broadcast() || current_device.broadcast_address == dest_ip { @@ -119,17 +188,9 @@ pub async fn base_handle(sender: &ChannelSender, buf: &mut [u8], broadcast(sender, &mut net_packet, data_len, ¤t_device).await?; } return Ok(()); - } else if dest_ip.is_multicast() { - if protocol == Protocol::Udp { - if let Some(cipher) = cipher { - //需要加密 - encrypt(cipher, &mut data_len, &mut net_packet)?; - } - multicast(igmp_server, dest_ip, sender, &mut net_packet, data_len, ¤t_device).await?; - } - return Ok(()); - } else { - if !check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) { + } + if !check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) { + if let Some(ip_route) = ip_route { if let Some(r_dest_ip) = ip_route.route(&dest_ip) { //路由的目标不能是自己 if r_dest_ip == src_ip { @@ -142,41 +203,43 @@ pub async fn base_handle(sender: &ChannelSender, buf: &mut [u8], return Ok(()); } } else { - match protocol { - Protocol::Tcp => { - let dest_addr = { - let tcp_packet = TcpPacket::new(src_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?; - SocketAddrV4::new(dest_ip, tcp_packet.destination_port()) - }; - if let Some(entry) = proxy_map.tcp_proxy_map.get(&dest_addr) { - let source_addr = entry.value().1; - let source_ip = *source_addr.ip(); - let mut tcp_packet = TcpPacket::new(source_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?; - tcp_packet.set_source_port(source_addr.port()); - tcp_packet.update_checksum(); - let mut ipv4_packet = IpV4Packet::new(&mut net_packet.buffer_mut()[12..data_len])?; - ipv4_packet.set_source_ip(source_ip); - ipv4_packet.update_checksum(); - } + return Ok(()); + } + } else if let Some(proxy_map) = proxy_map { + match protocol { + Protocol::Tcp => { + let dest_addr = { + let tcp_packet = TcpPacket::new(src_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?; + SocketAddrV4::new(dest_ip, tcp_packet.destination_port()) + }; + if let Some(entry) = proxy_map.tcp_proxy_map.get(&dest_addr) { + let source_addr = entry.value().1; + let source_ip = *source_addr.ip(); + let mut tcp_packet = TcpPacket::new(source_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?; + tcp_packet.set_source_port(source_addr.port()); + tcp_packet.update_checksum(); + let mut ipv4_packet = IpV4Packet::new(&mut net_packet.buffer_mut()[12..data_len])?; + ipv4_packet.set_source_ip(source_ip); + ipv4_packet.update_checksum(); } - Protocol::Udp => { - let dest_addr = { - let udp_packet = UdpPacket::new(src_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?; - SocketAddrV4::new(dest_ip, udp_packet.destination_port()) - }; - if let Some(entry) = proxy_map.udp_proxy_map.get(&dest_addr) { - let source_addr = entry.value().1; - let source_ip = *source_addr.ip(); - let mut udp_packet = UdpPacket::new(source_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?; - udp_packet.set_source_port(source_addr.port()); - udp_packet.update_checksum(); - let mut ipv4_packet = IpV4Packet::new(&mut net_packet.buffer_mut()[12..data_len])?; - ipv4_packet.set_source_ip(source_ip); - ipv4_packet.update_checksum(); - } - } - _ => {} } + Protocol::Udp => { + let dest_addr = { + let udp_packet = UdpPacket::new(src_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?; + SocketAddrV4::new(dest_ip, udp_packet.destination_port()) + }; + if let Some(entry) = proxy_map.udp_proxy_map.get(&dest_addr) { + let source_addr = entry.value().1; + let source_ip = *source_addr.ip(); + let mut udp_packet = UdpPacket::new(source_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?; + udp_packet.set_source_port(source_addr.port()); + udp_packet.update_checksum(); + let mut ipv4_packet = IpV4Packet::new(&mut net_packet.buffer_mut()[12..data_len])?; + ipv4_packet.set_source_ip(source_ip); + ipv4_packet.update_checksum(); + } + } + _ => {} } } if let Some(cipher) = cipher { diff --git a/switch/src/handle/tun_tap/tap_handler.rs b/switch/src/handle/tun_tap/tap_handler.rs index 9b17070..fbe30d9 100644 --- a/switch/src/handle/tun_tap/tap_handler.rs +++ b/switch/src/handle/tun_tap/tap_handler.rs @@ -1,7 +1,7 @@ use std::sync::Arc; use std::{io, thread}; use aes_gcm::Aes256Gcm; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use packet::arp::arp::ArpPacket; use packet::ethernet; use packet::ethernet::packet::EthernetPacket; @@ -19,10 +19,10 @@ use crate::tun_tap_device::{DeviceReader, DeviceWriter}; pub fn start(sender: ChannelSender, device_reader: DeviceReader, device_writer: DeviceWriter, - igmp_server: IgmpServer, + igmp_server: Option, current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap, + ip_route: Option, + ip_proxy_map: Option, cipher: Option) { thread::Builder::new().name("tap-handler".into()).spawn(move || { tokio::runtime::Builder::new_current_thread() @@ -40,23 +40,23 @@ pub fn start(sender: ChannelSender, async fn start_(sender: ChannelSender, device_reader: DeviceReader, device_writer: DeviceWriter, - igmp_server: IgmpServer, + igmp_server: Option, current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap, + ip_route: Option, + ip_proxy_map: Option, cipher: Option) -> io::Result<()> { - let mut buf = [0; 2048]; + let mut buf = [0; 4096]; loop { //ip拆包了会直接丢弃？ let len = device_reader.read(&mut buf)?; if let Err(e) = handle(&mut buf, len, &igmp_server, ¤t_device, &device_writer, &sender, &ip_route, &ip_proxy_map, &cipher).await { - log::error!("tap handle{:?}",e); + log::warn!("tap handle{:?}",e); } } } -async fn handle(buf: &mut [u8], len: usize, igmp_server: &IgmpServer, current_device: &AtomicCell, - device_writer: &DeviceWriter, sender: &ChannelSender, ip_route: &ExternalRoute, proxy_map: &IpProxyMap, cipher: &Option) -> crate::Result<()> { +async fn handle(buf: &mut [u8], len: usize, igmp_server: &Option, current_device: &AtomicCell, + device_writer: &DeviceWriter, sender: &ChannelSender, ip_route: &Option, proxy_map: &Option, cipher: &Option) -> crate::Result<()> { let mut ethernet_packet = EthernetPacket::new(&mut buf[..len])?; let current_device = current_device.load(); match ethernet_packet.protocol() { diff --git a/switch/src/handle/tun_tap/tun_handler.rs b/switch/src/handle/tun_tap/tun_handler.rs index 73331fa..4f09332 100644 --- a/switch/src/handle/tun_tap/tun_handler.rs +++ b/switch/src/handle/tun_tap/tun_handler.rs @@ -2,7 +2,7 @@ use std::{io, thread}; use std::sync::Arc; use aes_gcm::Aes256Gcm; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use packet::icmp::Kind; use packet::icmp::icmp::IcmpPacket; @@ -35,8 +35,8 @@ fn icmp(device_writer: &DeviceWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> /// 接收tun数据，并且转发到udp上 #[inline] -async fn handle(sender: &ChannelSender, data: &mut [u8], len: usize, device_writer: &DeviceWriter, igmp_server: &IgmpServer, current_device: CurrentDeviceInfo, - ip_route: &ExternalRoute, proxy_map: &IpProxyMap,cipher: &Option) -> Result<()> { +async fn handle(sender: &ChannelSender, data: &mut [u8], len: usize, device_writer: &DeviceWriter, igmp_server: &Option, current_device: CurrentDeviceInfo, + ip_route: &Option, proxy_map: &Option,cipher: &Option) -> Result<()> { let ipv4_packet = if let Ok(ipv4_packet) = IpV4Packet::new(&mut data[12..len]) { ipv4_packet } else { @@ -56,13 +56,13 @@ async fn handle(sender: &ChannelSender, data: &mut [u8], len: usize, device_writ pub fn start(sender: ChannelSender, device_reader: DeviceReader, device_writer: DeviceWriter, - igmp_server: IgmpServer, + igmp_server: Option, current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap, + ip_route: Option, + ip_proxy_map: Option, cipher: Option) { thread::Builder::new().name("tun-handler".into()).spawn(move || { - tokio::runtime::Builder::new_current_thread() + tokio::runtime::Builder::new_multi_thread() .enable_all().build().unwrap() .block_on(async move { if let Err(e) = start_(sender, device_reader, device_writer, igmp_server, current_device, ip_route, ip_proxy_map,cipher).await { @@ -75,19 +75,29 @@ pub fn start(sender: ChannelSender, async fn start_(sender: ChannelSender, device_reader: DeviceReader, device_writer: DeviceWriter, - igmp_server: IgmpServer, + igmp_server: Option, current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap, + ip_route: Option, + ip_proxy_map: Option, cipher: Option) -> io::Result<()> { - let mut buf = [0; 4096]; loop { + let mut buf = [0; 4096]; + let sender = sender.clone(); + let device_writer = device_writer.clone(); + let igmp_server = igmp_server.clone(); + let ip_route = ip_route.clone(); + let ip_proxy_map = ip_proxy_map.clone(); + let cipher = cipher.clone(); let len = device_reader.read(&mut buf[12..])? + 12; - match handle(&sender, &mut buf, len, &device_writer, &igmp_server, current_device.load(), &ip_route, &ip_proxy_map,&cipher).await { - Ok(_) => {} - Err(e) => { - log::warn!("{:?}", e) + let current_device = current_device.load(); + tokio::spawn(async move { + match handle(&sender, &mut buf, len, &device_writer, &igmp_server, current_device, &ip_route, &ip_proxy_map,&cipher).await { + Ok(_) => {} + Err(e) => { + log::warn!("{:?}", e) + } } - } + }); + } } diff --git a/switch/src/igmp_server/mod.rs b/switch/src/igmp_server/mod.rs index 2fc268b..348e568 100644 --- a/switch/src/igmp_server/mod.rs +++ b/switch/src/igmp_server/mod.rs @@ -1,8 +1,8 @@ use std::collections::{HashMap, HashSet}; use std::net::Ipv4Addr; use std::sync::Arc; -use std::time::Duration; -use moka::sync::Cache; +use std::time::{Duration, Instant}; +use crossbeam_skiplist::SkipMap; use parking_lot::RwLock; use packet::igmp::igmp_v2::IgmpV2Packet; use packet::igmp::igmp_v3::{IgmpV3QueryPacket, IgmpV3RecordType, IgmpV3ReportPacket}; @@ -15,7 +15,7 @@ use crate::tun_tap_device::DeviceWriter; #[derive(Clone, Debug)] pub struct Multicast { //成员虚拟ip - members: HashSet, + members: HashMap, //是否是过滤模式 //成员过滤或包含的源ip map: HashMap)>, @@ -29,7 +29,7 @@ impl Multicast { } } pub fn is_send(&self, ip: &Ipv4Addr) -> bool { - if self.members.contains(ip) { + if self.members.contains_key(ip) { if let Some((is_include, set)) = self.map.get(ip) { if *is_include { set.contains(ip) @@ -47,30 +47,13 @@ impl Multicast { #[derive(Clone)] pub struct IgmpServer { - multicast: Cache>>, - members: Cache<(Ipv4Addr, Ipv4Addr), ()>, + multicast: Arc>>>, } impl IgmpServer { pub fn new(device_writer: DeviceWriter) -> Self { - let multicast: Cache>> = Cache::builder() - .time_to_idle(Duration::from_secs(30 * 60)).build(); - let m = multicast.clone(); - let members: Cache<(Ipv4Addr, Ipv4Addr), ()> = Cache::builder() - .time_to_idle(Duration::from_secs(20 * 60)).eviction_listener(move |k: Arc<(Ipv4Addr, Ipv4Addr)>, _, cause| { - if cause == moka::notification::RemovalCause::Replaced { - return; - } - log::info!("MULTICAST_MEMBER eviction {:?}", k); - if let Some(v) = m.get(&k.0) { - let mut lock = v.write(); - lock.members.remove(&k.1); - lock.map.remove(&k.1); - } - }).build(); + let multicast: Arc>>> = Arc::new(SkipMap::new()); std::thread::spawn(move || { - //定时发送query，启动时20秒一次，连发3次，之后125秒一次 - let mut count = 0; //预留以太网帧头和ip头 let mut buf = [0; 14 + 24 + 12]; let dest = Ipv4Addr::new(224, 0, 0, 1); @@ -96,31 +79,42 @@ impl IgmpServer { { let mut igmp_query = IgmpV3QueryPacket::unchecked(&mut buf[14 + 24..]); igmp_query.set_igmp_type(); - igmp_query.set_max_resp_code(100); + igmp_query.set_max_resp_code(50); igmp_query.set_group_address(Ipv4Addr::UNSPECIFIED); igmp_query.set_qrv(2); - igmp_query.set_qqic(125); + igmp_query.set_qqic(10); igmp_query.update_checksum(); } loop { let _ = device_writer.write_ipv4(&mut buf); - if count < 3 { - count += 1; - std::thread::sleep(Duration::from_secs(20)) - } else { - std::thread::sleep(Duration::from_secs(125)) - } + std::thread::sleep(Duration::from_secs(20)) } }); Self { multicast, - members, } } pub fn load(&self, multicast_addr: &Ipv4Addr) -> Option>> { - self.multicast.get(multicast_addr) + if let Some(entry) = self.multicast.get(multicast_addr) { + Some(entry.value().clone()) + } else { + None + } } pub fn handle(&self, buf: &[u8], source: Ipv4Addr) -> crate::Result<()> { + for x in self.multicast.iter() { + let mut list = Vec::new(); + let mut write_guard = x.value().write(); + for (ip, time) in &write_guard.members { + if time.elapsed() > Duration::from_secs(30) { + list.push(*ip); + } + } + for ip in list { + write_guard.members.remove(&ip); + write_guard.map.remove(&ip); + } + } match IgmpType::from(buf[0]) { IgmpType::Query => {} IgmpType::ReportV1 | IgmpType::ReportV2 => { @@ -130,13 +124,11 @@ impl IgmpServer { if !multicast_addr.is_multicast() { return Ok(()); } - let multi = self.multicast.get_with(multicast_addr, || { + let multi = self.multicast.get_or_insert_with(multicast_addr, || { Arc::new(RwLock::new(Multicast::new())) }); - let mut guard = multi.write(); - guard.members.insert(source); - drop(guard); - self.members.insert((multicast_addr, source), ()); + let mut guard = multi.value().write(); + guard.members.insert(source, Instant::now()); } IgmpType::LeaveV2 => { //退出组播 @@ -145,7 +137,11 @@ impl IgmpServer { if !multicast_addr.is_multicast() { return Ok(()); } - self.members.invalidate(&(multicast_addr, source)); + if let Some(entry) = self.multicast.get(&multicast_addr) { + let mut guard = entry.value().write(); + guard.map.remove(&source); + guard.members.remove(&source); + } } IgmpType::ReportV3 => { let report = IgmpV3ReportPacket::new(buf)?; @@ -155,10 +151,10 @@ impl IgmpServer { if !multicast_addr.is_multicast() { return Ok(()); } - let multi = self.multicast.get_with(multicast_addr, || { + let multi = self.multicast.get_or_insert_with(multicast_addr, || { Arc::new(RwLock::new(Multicast::new())) }); - let mut guard = multi.write(); + let mut guard = multi.value().write(); match group_record.record_type() { IgmpV3RecordType::ModeIsInclude | IgmpV3RecordType::ChangeToIncludeMode => { @@ -169,10 +165,8 @@ impl IgmpServer { guard.map.remove(&source); } Some(src) => { - guard.members.insert(source); + guard.members.insert(source, Instant::now()); guard.map.insert(source, (true, HashSet::from_iter(src))); - drop(guard); - self.members.insert((multicast_addr, source), ()); } } } @@ -181,16 +175,14 @@ impl IgmpServer { match group_record.source_addresses() { None => { //接收所有 - guard.members.insert(source); + guard.members.insert(source, Instant::now()); guard.map.remove(&source); } Some(src) => { - guard.members.insert(source); + guard.members.insert(source, Instant::now()); guard.map.insert(source, (false, HashSet::from_iter(src))); } } - drop(guard); - self.members.insert((multicast_addr, source), ()); } IgmpV3RecordType::AllowNewSources => { //在已有源的基础上，接收目标源，如果是排除模式，则删除；是包含模式则添加 @@ -211,8 +203,6 @@ impl IgmpServer { } } } - drop(guard); - self.members.insert((multicast_addr, source), ()); } IgmpV3RecordType::BlockOldSources => { //在已有源的基础上，不接收目标源 @@ -233,8 +223,6 @@ impl IgmpServer { } } } - drop(guard); - self.members.insert((multicast_addr, source), ()); } IgmpV3RecordType::Unknown(_) => {} } diff --git a/switch/src/ip_proxy/icmp_proxy.rs b/switch/src/ip_proxy/icmp_proxy.rs index 87ff7d9..c5c5e8b 100644 --- a/switch/src/ip_proxy/icmp_proxy.rs +++ b/switch/src/ip_proxy/icmp_proxy.rs @@ -2,7 +2,7 @@ use std::io; use std::mem::MaybeUninit; use std::net::{IpAddr, Ipv4Addr, SocketAddrV4}; use std::sync::Arc; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use crossbeam_skiplist::SkipMap; use socket2::{Domain, SockAddr, Socket, Type}; @@ -68,7 +68,7 @@ impl IcmpProxy { net_packet.set_version(Version::V1); net_packet.set_protocol(Protocol::IpTurn); net_packet.set_transport_protocol(ipv4::protocol::Protocol::Icmp.into()); - net_packet.set_ttl(MAX_TTL); + net_packet.first_set_ttl(MAX_TTL); loop { match self.recv(data) { Ok((len, peer_ip)) => { diff --git a/switch/src/ip_proxy/mod.rs b/switch/src/ip_proxy/mod.rs index abfced0..ea7d5ba 100644 --- a/switch/src/ip_proxy/mod.rs +++ b/switch/src/ip_proxy/mod.rs @@ -2,7 +2,7 @@ use std::{io, thread}; use std::collections::HashMap; use std::net::{Ipv4Addr, SocketAddrV4}; use std::sync::Arc; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use crossbeam_skiplist::SkipMap; use socket2::{SockAddr, Socket}; use tokio::net::{TcpListener, UdpSocket}; diff --git a/switch/src/ip_proxy/tcp_proxy.rs b/switch/src/ip_proxy/tcp_proxy.rs index 6798003..5e2f3c2 100644 --- a/switch/src/ip_proxy/tcp_proxy.rs +++ b/switch/src/ip_proxy/tcp_proxy.rs @@ -6,7 +6,6 @@ use tokio::net::{TcpListener, TcpStream}; pub struct TcpProxy { tcp_listener: TcpListener, - // todo 怎么过期 map: Arc>, } @@ -34,6 +33,7 @@ impl TcpProxy { continue; } }; + let map = map.clone(); tokio::spawn(async move { match proxy(tcp_stream, peer_tcp_stream).await { Ok(_) => {} @@ -41,6 +41,7 @@ impl TcpProxy { log::warn!("tcp代理异常:{:?},来源:{},目标：{}",e,src_addr,dest_addr); } } + map.remove(&sender_addr); }); } } diff --git a/switch/src/ip_proxy/udp_proxy.rs b/switch/src/ip_proxy/udp_proxy.rs index 9e0be3d..4376562 100644 --- a/switch/src/ip_proxy/udp_proxy.rs +++ b/switch/src/ip_proxy/udp_proxy.rs @@ -8,7 +8,6 @@ use tokio::net::UdpSocket; /// 一个udp代理，作用是利用系统协议栈，将udp数据报解析出来再转发到目的地址 pub struct UdpProxy { udp_socket: Arc, - // todo 过期处理 map: Arc>, } @@ -61,6 +60,7 @@ async fn start0(buf: &[u8], sender_addr: SocketAddrV4, inner_map: &Arc for NatType { -// fn into(self) -> u8 { -// match self { -// NatType::Symmetric => 0, -// NatType::Cone => 1, -// } -// } -// } /// 返回所有公网ip和端口变化范围 pub fn public_ip_list(addrs: &Vec) -> io::Result<(NatType, Vec, u16)> { diff --git a/switch/src/tun_tap_device/unix.rs b/switch/src/tun_tap_device/unix.rs index 3e2dd13..b36d7e7 100644 --- a/switch/src/tun_tap_device/unix.rs +++ b/switch/src/tun_tap_device/unix.rs @@ -5,7 +5,7 @@ use bytes::BufMut; use tun::platform::posix::{Reader, Writer}; use std::net::Ipv4Addr; use std::os::unix::io::AsRawFd; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; #[cfg(any(target_os = "linux", target_os = "android"))] use tun::platform::linux::Device; #[cfg(any(target_os = "macos", target_os = "ios"))] diff --git a/switch/src/tun_tap_device/windows.rs b/switch/src/tun_tap_device/windows.rs index fff6978..d9021d6 100644 --- a/switch/src/tun_tap_device/windows.rs +++ b/switch/src/tun_tap_device/windows.rs @@ -2,7 +2,7 @@ use std::{io, thread}; use std::net::Ipv4Addr; use std::sync::Arc; use std::time::Duration; -use crossbeam::atomic::AtomicCell; +use crossbeam_utils::atomic::AtomicCell; use libloading::Library; use parking_lot::Mutex; use packet::ethernet;