diff --git a/vnt/src/channel/tcp_channel.rs b/vnt/src/channel/tcp_channel.rs index 07dbde4..c5a0219 100644 --- a/vnt/src/channel/tcp_channel.rs +++ b/vnt/src/channel/tcp_channel.rs @@ -31,7 +31,7 @@ pub fn tcp_listen( stop_manager: StopManager, recv_handler: H, context: ChannelContext, -) -> io::Result>)>> +) -> anyhow::Result>)>> where H: RecvChannelHandler, { @@ -75,7 +75,7 @@ fn tcp_listen0( accept_tcp_receiver: Receiver<(TcpStream, SocketAddr, Option>)>, mut recv_handler: H, context: ChannelContext, -) -> io::Result<()> +) -> anyhow::Result<()> where H: RecvChannelHandler, { @@ -109,7 +109,7 @@ where if e.kind() == io::ErrorKind::WouldBlock { break; } - return Err(e); + return Err(e)?; } } }, @@ -164,7 +164,7 @@ fn init_writable_handler( receiver: Receiver<(TcpStream, Token, SocketAddr, Option>)>, stop_manager: StopManager, context: ChannelContext, -) -> io::Result { +) -> anyhow::Result { let poll = Poll::new()?; let writable_notify = WritableNotify::new(Waker::new(poll.registry(), NOTIFY)?); let worker = { diff --git a/vnt/src/channel/udp_channel.rs b/vnt/src/channel/udp_channel.rs index b5adbd7..4a0a26a 100644 --- a/vnt/src/channel/udp_channel.rs +++ b/vnt/src/channel/udp_channel.rs @@ -17,7 +17,7 @@ pub fn udp_listen( stop_manager: StopManager, recv_handler: H, context: ChannelContext, -) -> io::Result>>> +) -> anyhow::Result>>> where H: RecvChannelHandler, { @@ -31,7 +31,7 @@ fn sub_udp_listen( stop_manager: StopManager, recv_handler: H, context: ChannelContext, -) -> io::Result>>> +) -> anyhow::Result>>> where H: RecvChannelHandler, { @@ -208,7 +208,7 @@ fn main_udp_listen( stop_manager: StopManager, recv_handler: H, context: ChannelContext, -) -> io::Result<()> +) -> anyhow::Result<()> where H: RecvChannelHandler, { diff --git a/vnt/src/handle/maintain/heartbeat.rs b/vnt/src/handle/maintain/heartbeat.rs index d9ce1dc..e5b4a94 100644 --- a/vnt/src/handle/maintain/heartbeat.rs +++ b/vnt/src/handle/maintain/heartbeat.rs @@ -1,4 +1,3 @@ -use std::io; use std::net::Ipv4Addr; use std::sync::Arc; use std::time::Duration; @@ -167,7 +166,7 @@ fn client_relay0( current_device: &CurrentDeviceInfo, device_list: &Mutex<(u16, Vec)>, client_cipher: &Cipher, -) -> io::Result<()> { +) -> anyhow::Result<()> { // 离线了不再探测 if current_device.status.offline() { return Ok(()); @@ -211,7 +210,7 @@ fn client_relay0( fn heartbeat_packet( src: Ipv4Addr, dest: Ipv4Addr, -) -> io::Result> { +) -> anyhow::Result> { let mut net_packet = NetPacket::new_encrypt([0u8; 12 + 4 + ENCRYPTION_RESERVED])?; net_packet.set_default_version(); net_packet.set_protocol(Protocol::Control); @@ -228,7 +227,7 @@ fn heartbeat_packet_client( client_cipher: &Cipher, src: Ipv4Addr, dest: Ipv4Addr, -) -> io::Result> { +) -> anyhow::Result> { let mut net_packet = heartbeat_packet(src, dest)?; client_cipher.encrypt_ipv4(&mut net_packet)?; Ok(net_packet) @@ -239,7 +238,7 @@ fn heartbeat_packet_server( server_cipher: &Cipher, src: Ipv4Addr, dest: Ipv4Addr, -) -> io::Result> { +) -> anyhow::Result> { let mut net_packet = heartbeat_packet(src, dest)?; let mut ping = PingPacket::new(net_packet.payload_mut())?; ping.set_epoch(device_list.lock().0); diff --git a/vnt/src/handle/maintain/punch.rs b/vnt/src/handle/maintain/punch.rs index 8c7c226..8c00f45 100644 --- a/vnt/src/handle/maintain/punch.rs +++ b/vnt/src/handle/maintain/punch.rs @@ -2,9 +2,10 @@ use std::collections::HashMap; use std::net::Ipv4Addr; use std::sync::mpsc::{sync_channel, Receiver, SyncSender}; use std::sync::Arc; +use std::thread; use std::time::Duration; -use std::{io, thread}; +use anyhow::anyhow; use crossbeam_utils::atomic::AtomicCell; use parking_lot::Mutex; use protobuf::Message; @@ -222,7 +223,7 @@ fn punch0( punch_record: &Mutex>, last_punch_record: &mut HashMap, total_count: usize, -) -> io::Result<()> { +) -> anyhow::Result<()> { let nat_info = nat_test.nat_info(); if total_count < 10 && (nat_info.public_ips.is_empty() @@ -297,7 +298,7 @@ fn punch_packet( virtual_ip: Ipv4Addr, nat_info: &NatInfo, dest: Ipv4Addr, -) -> io::Result>> { +) -> anyhow::Result>> { let mut punch_reply = PunchInfo::new(); punch_reply.reply = false; punch_reply.public_ip_list = nat_info @@ -320,7 +321,7 @@ fn punch_packet( log::info!("请求打洞={:?}", punch_reply); let bytes = punch_reply .write_to_bytes() - .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("punch_packet {:?}", e)))?; + .map_err(|e| anyhow!("punch_packet {:?}", e))?; let mut net_packet = NetPacket::new_encrypt(vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED])?; net_packet.set_default_version(); net_packet.set_protocol(Protocol::OtherTurn); diff --git a/vnt/src/handle/recv_data/client.rs b/vnt/src/handle/recv_data/client.rs index 70e8f99..5d619d6 100644 --- a/vnt/src/handle/recv_data/client.rs +++ b/vnt/src/handle/recv_data/client.rs @@ -1,13 +1,16 @@ -use parking_lot::RwLock; -use protobuf::Message; +use anyhow::anyhow; use std::collections::HashMap; -use std::io; use std::net::{Ipv4Addr, Ipv6Addr}; use std::sync::Arc; +use parking_lot::RwLock; +use protobuf::Message; + use packet::icmp::{icmp, Kind}; use packet::ip::ipv4; use packet::ip::ipv4::packet::IpV4Packet; +#[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] +use tun::device::IFace; use crate::channel::context::ChannelContext; use crate::channel::punch::NatInfo; @@ -28,8 +31,6 @@ use crate::protocol::{ control_packet, ip_turn_packet, other_turn_packet, NetPacket, Protocol, MAX_TTL, }; use crate::tun_tap_device::tun_create_helper::DeviceAdapter; -#[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] -use tun::device::IFace; /// 处理来源于客户端的包 #[derive(Clone)] @@ -116,7 +117,7 @@ impl ClientPacketHandler { context: &ChannelContext, current_device: &CurrentDeviceInfo, route_key: RouteKey, - ) -> io::Result<()> { + ) -> anyhow::Result<()> { let destination = net_packet.destination(); let source = net_packet.source(); match ip_turn_packet::Protocol::from(net_packet.transport_protocol()) { @@ -203,7 +204,7 @@ impl ClientPacketHandler { current_device: &CurrentDeviceInfo, mut net_packet: NetPacket<&mut [u8]>, route_key: RouteKey, - ) -> io::Result<()> { + ) -> anyhow::Result<()> { let metric = net_packet.source_ttl() - net_packet.ttl() + 1; let source = net_packet.source(); match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? { @@ -291,17 +292,15 @@ impl ClientPacketHandler { current_device: &CurrentDeviceInfo, net_packet: NetPacket<&mut [u8]>, route_key: RouteKey, - ) -> io::Result<()> { + ) -> anyhow::Result<()> { if context.use_channel_type().is_only_relay() { return Ok(()); } let source = net_packet.source(); match other_turn_packet::Protocol::from(net_packet.transport_protocol()) { other_turn_packet::Protocol::Punch => { - let mut punch_info = - PunchInfo::parse_from_bytes(net_packet.payload()).map_err(|e| { - io::Error::new(io::ErrorKind::Other, format!("PunchInfo {:?}", e)) - })?; + let mut punch_info = PunchInfo::parse_from_bytes(net_packet.payload()) + .map_err(|e| anyhow!("PunchInfo {:?}", e))?; let public_ips = punch_info .public_ip_list .iter() @@ -361,9 +360,9 @@ impl ClientPacketHandler { punch_reply.ipv6 = ipv6.octets().to_vec(); punch_reply.ipv6_port = nat_info.udp_ports[0] as u32; } - let bytes = punch_reply.write_to_bytes().map_err(|e| { - io::Error::new(io::ErrorKind::Other, format!("punch_reply {:?}", e)) - })?; + let bytes = punch_reply + .write_to_bytes() + .map_err(|e| anyhow!("punch_reply {:?}", e))?; let mut punch_packet = NetPacket::new_encrypt(vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED])?; punch_packet.set_default_version(); diff --git a/vnt/src/handle/recv_data/mod.rs b/vnt/src/handle/recv_data/mod.rs index 8c2bdea..bb14c20 100644 --- a/vnt/src/handle/recv_data/mod.rs +++ b/vnt/src/handle/recv_data/mod.rs @@ -62,7 +62,12 @@ impl RecvChannelHandler for RecvDataHandler { } } if let Err(e) = self.handle0(buf, extend, route_key, context) { - log::error!("[{}]-{:?}", thread::current().name().unwrap_or(""), e); + log::error!( + "[{}]-{:?}-{:?}", + thread::current().name().unwrap_or(""), + route_key.addr, + e + ); } } } diff --git a/vnt/src/handle/recv_data/server.rs b/vnt/src/handle/recv_data/server.rs index c42ec48..747e3be 100644 --- a/vnt/src/handle/recv_data/server.rs +++ b/vnt/src/handle/recv_data/server.rs @@ -1,3 +1,4 @@ +use anyhow::anyhow; use std::io; use std::net::Ipv4Addr; use std::sync::Arc; @@ -11,6 +12,8 @@ use protobuf::Message; use packet::icmp::{icmp, Kind}; use packet::ip::ipv4; use packet::ip::ipv4::packet::IpV4Packet; +#[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] +use tun::device::IFace; use crate::channel::context::ChannelContext; use crate::channel::{Route, RouteKey}; @@ -34,8 +37,6 @@ use crate::protocol::error_packet::InErrorPacket; use crate::protocol::{ip_turn_packet, service_packet, NetPacket, Protocol, MAX_TTL}; use crate::tun_tap_device::tun_create_helper::DeviceAdapter; use crate::{proto, PeerClientInfo}; -#[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] -use tun::device::IFace; /// 处理来源于服务端的包 #[derive(Clone)] @@ -136,10 +137,8 @@ impl PacketHandler for ServerPacketHandler { } else if net_packet.protocol() == Protocol::Service && net_packet.transport_protocol() == service_packet::Protocol::HandshakeResponse.into() { - let response = - HandshakeResponse::parse_from_bytes(net_packet.payload()).map_err(|e| { - io::Error::new(io::ErrorKind::Other, format!("HandshakeResponse {:?}", e)) - })?; + let response = HandshakeResponse::parse_from_bytes(net_packet.payload()) + .map_err(|e| anyhow!("HandshakeResponse {:?}", e))?; log::info!("握手响应:{:?},{}", route_key, response); //如果开启了加密,则发送加密握手请求 #[cfg(feature = "server_encrypt")] @@ -254,7 +253,7 @@ impl ServerPacketHandler { current_device: &CurrentDeviceInfo, net_packet: NetPacket<&mut [u8]>, route_key: RouteKey, - ) -> io::Result<()> { + ) -> anyhow::Result<()> { match service_packet::Protocol::from(net_packet.transport_protocol()) { service_packet::Protocol::RegistrationResponse => { let response = RegistrationResponse::parse_from_bytes(net_packet.payload()) @@ -440,7 +439,7 @@ impl ServerPacketHandler { &self, current_device: &CurrentDeviceInfo, context: &ChannelContext, - ) -> io::Result<()> { + ) -> anyhow::Result<()> { if current_device.status.online() { log::info!("已连接的不需要注册,{:?}", self.config_info); return Ok(()); @@ -469,7 +468,8 @@ impl ServerPacketHandler { )?; log::info!("发送注册请求,{:?}", self.config_info); //注册请求只发送到默认通道 - context.send_default(response.buffer(), current_device.connect_server) + context.send_default(response.buffer(), current_device.connect_server)?; + Ok(()) } fn error( &self, @@ -527,7 +527,7 @@ impl ServerPacketHandler { current_device: &CurrentDeviceInfo, net_packet: NetPacket<&mut [u8]>, route_key: RouteKey, - ) -> io::Result<()> { + ) -> anyhow::Result<()> { match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? { ControlPacket::PongPacket(pong_packet) => { let current_time = crate::handle::now_time() as u16; diff --git a/vnt/src/handle/registrar.rs b/vnt/src/handle/registrar.rs index 160850d..417fea5 100644 --- a/vnt/src/handle/registrar.rs +++ b/vnt/src/handle/registrar.rs @@ -1,4 +1,4 @@ -use std::io; +use anyhow::anyhow; use std::net::Ipv4Addr; use protobuf::Message; @@ -19,7 +19,7 @@ pub fn registration_request_packet( is_fast: bool, allow_ip_change: bool, client_secret_hash: Option<&[u8]>, -) -> io::Result>> { +) -> anyhow::Result>> { let mut request = RegistrationRequest::new(); request.token = token; request.device_id = device_id; @@ -36,9 +36,9 @@ pub fn registration_request_packet( .client_secret_hash .extend_from_slice(client_secret_hash); } - let bytes = request.write_to_bytes().map_err(|e| { - io::Error::new(io::ErrorKind::Other, format!("RegistrationRequest {:?}", e)) - })?; + let bytes = request + .write_to_bytes() + .map_err(|e| anyhow!("RegistrationRequest {:?}", e))?; let buf = vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED]; let mut net_packet = NetPacket::new_encrypt(buf)?; net_packet.set_destination(GATEWAY_IP); diff --git a/vnt/src/handle/tun_tap/tun_handler.rs b/vnt/src/handle/tun_tap/tun_handler.rs index f5217be..4d94d86 100644 --- a/vnt/src/handle/tun_tap/tun_handler.rs +++ b/vnt/src/handle/tun_tap/tun_handler.rs @@ -5,7 +5,6 @@ use std::{io, thread}; use crossbeam_utils::atomic::AtomicCell; use parking_lot::Mutex; -use crate::channel::BUFFER_SIZE; use packet::icmp::icmp::IcmpPacket; use packet::icmp::Kind; use packet::ip::ipv4::packet::IpV4Packet; @@ -14,6 +13,7 @@ use tun::device::IFace; use tun::Device; use crate::channel::context::ChannelContext; +use crate::channel::BUFFER_SIZE; use crate::cipher::Cipher; use crate::compression::Compressor; use crate::external_route::ExternalRoute; @@ -187,7 +187,7 @@ fn broadcast( net_packet: &mut NetPacket<&mut [u8]>, current_device: &CurrentDeviceInfo, device_list: &Mutex<(u16, Vec)>, -) -> io::Result<()> { +) -> anyhow::Result<()> { let list: Vec = device_list .lock() .1 @@ -261,7 +261,8 @@ fn broadcast( broadcast.set_address(&p2p_ips)?; broadcast.set_data(net_packet.buffer())?; server_cipher.encrypt_ipv4(&mut server_packet)?; - sender.send_default(server_packet.buffer(), current_device.connect_server) + sender.send_default(server_packet.buffer(), current_device.connect_server)?; + Ok(()) } /// 实现一个原地发送,必须保证是如下结构 @@ -302,6 +303,38 @@ fn base_handle( } return Ok(()); } + if !dest_ip.is_multicast() && !dest_ip.is_broadcast() && current_device.broadcast_ip != dest_ip + { + if !check_dest( + dest_ip, + current_device.virtual_netmask, + current_device.virtual_network, + ) { + if let Some(r_dest_ip) = ip_route.route(&dest_ip) { + //路由的目标不能是自己 + if r_dest_ip == src_ip { + return Ok(()); + } + //需要修改目的地址 + dest_ip = r_dest_ip; + net_packet.set_destination(r_dest_ip); + } else { + return Ok(()); + } + } + #[cfg(feature = "ip_proxy")] + if let Some(proxy_map) = proxy_map { + let mut ipv4_packet = IpV4Packet::new(net_packet.payload_mut())?; + proxy_map.send_handle(&mut ipv4_packet)?; + } + } + + if dest_ip.is_multicast() { + //当作广播处理 + dest_ip = Ipv4Addr::BROADCAST; + net_packet.set_destination(Ipv4Addr::BROADCAST); + } + let mut net_packet = if compressor.compress(&net_packet, &mut out)? { out.set_default_version(); out.set_protocol(protocol::Protocol::IpTurn); @@ -313,11 +346,6 @@ fn base_handle( } else { net_packet }; - if dest_ip.is_multicast() { - //当作广播处理 - dest_ip = Ipv4Addr::BROADCAST; - net_packet.set_destination(Ipv4Addr::BROADCAST); - } if dest_ip.is_broadcast() || current_device.broadcast_ip == dest_ip { // 广播 发送到直连目标 client_cipher.encrypt_ipv4(&mut net_packet)?; @@ -330,28 +358,7 @@ fn base_handle( )?; return Ok(()); } - if !check_dest( - dest_ip, - current_device.virtual_netmask, - current_device.virtual_network, - ) { - if let Some(r_dest_ip) = ip_route.route(&dest_ip) { - //路由的目标不能是自己 - if r_dest_ip == src_ip { - return Ok(()); - } - //需要修改目的地址 - dest_ip = r_dest_ip; - net_packet.set_destination(r_dest_ip); - } else { - return Ok(()); - } - } - #[cfg(feature = "ip_proxy")] - if let Some(proxy_map) = proxy_map { - let mut ipv4_packet = IpV4Packet::new(net_packet.payload_mut())?; - proxy_map.send_handle(&mut ipv4_packet)?; - } + client_cipher.encrypt_ipv4(&mut net_packet)?; context.send_ipv4_by_id( net_packet.buffer(), diff --git a/vnt/src/handle/tun_tap/unix.rs b/vnt/src/handle/tun_tap/unix.rs index 09b0d4e..c49323a 100644 --- a/vnt/src/handle/tun_tap/unix.rs +++ b/vnt/src/handle/tun_tap/unix.rs @@ -33,7 +33,7 @@ pub(crate) fn start_simple( up_counter: &mut SingleU64Adder, device_list: Arc)>>, compressor: Compressor, -) -> io::Result<()> { +) -> anyhow::Result<()> { let poll = Poll::new()?; let waker = Arc::new(Waker::new(poll.registry(), STOP)?); let _waker = waker.clone(); @@ -73,7 +73,7 @@ fn start_simple0( up_counter: &mut SingleU64Adder, device_list: Arc)>>, compressor: Compressor, -) -> io::Result<()> { +) -> anyhow::Result<()> { let mut buf = [0; BUFFER_SIZE]; let mut extend = [0; BUFFER_SIZE]; let fd = device.as_tun_fd(); @@ -134,7 +134,7 @@ pub(crate) fn start_multi( device: Arc, group_sync_sender: GroupSyncSender<(Vec, usize)>, up_counter: &mut SingleU64Adder, -) -> io::Result<()> { +) -> anyhow::Result<()> { let poll = Poll::new()?; let waker = Arc::new(Waker::new(poll.registry(), STOP)?); let _waker = waker.clone(); @@ -154,7 +154,7 @@ fn start_multi0( device: Arc, mut group_sync_sender: GroupSyncSender<(Vec, usize)>, up_counter: &mut SingleU64Adder, -) -> io::Result<()> { +) -> anyhow::Result<()> { let fd = device.as_tun_fd(); fd.set_nonblock()?; SourceFd(&fd.as_raw_fd()).register(poll.registry(), FD, Interest::READABLE)?; diff --git a/vnt/src/handle/tun_tap/windows.rs b/vnt/src/handle/tun_tap/windows.rs index f1c86c7..0d03b1e 100644 --- a/vnt/src/handle/tun_tap/windows.rs +++ b/vnt/src/handle/tun_tap/windows.rs @@ -10,7 +10,6 @@ use crate::ip_proxy::IpProxyMap; use crate::util::{SingleU64Adder, StopManager}; use crossbeam_utils::atomic::AtomicCell; use parking_lot::Mutex; -use std::io; use std::sync::Arc; use tun::device::IFace; use tun::Device; @@ -27,7 +26,7 @@ pub(crate) fn start_simple( up_counter: &mut SingleU64Adder, device_list: Arc)>>, compressor: Compressor, -) -> io::Result<()> { +) -> anyhow::Result<()> { let worker = { let device = device.clone(); stop_manager.add_listener("tun_device".into(), move || { @@ -65,7 +64,7 @@ fn start_simple0( up_counter: &mut SingleU64Adder, device_list: Arc)>>, compressor: Compressor, -) -> io::Result<()> { +) -> anyhow::Result<()> { let mut buf = [0; BUFFER_SIZE]; let mut extend = [0; BUFFER_SIZE]; loop { @@ -101,7 +100,7 @@ pub(crate) fn start_multi( device: Arc, group_sync_sender: GroupSyncSender<(Vec, usize)>, up_counter: &mut SingleU64Adder, -) -> io::Result<()> { +) -> anyhow::Result<()> { let worker = { let device = device.clone(); stop_manager.add_listener("tun_device_multi".into(), move || { @@ -120,7 +119,7 @@ fn start_multi0( device: Arc, mut group_sync_sender: GroupSyncSender<(Vec, usize)>, up_counter: &mut SingleU64Adder, -) -> io::Result<()> { +) -> anyhow::Result<()> { loop { let mut buf = vec![0; 1024 * 16]; let len = device.read(&mut buf[12..])? + 12; diff --git a/vnt/src/util/notify.rs b/vnt/src/util/notify.rs index 5e8ec68..defa700 100644 --- a/vnt/src/util/notify.rs +++ b/vnt/src/util/notify.rs @@ -1,9 +1,10 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; +use std::thread; use std::thread::Thread; use std::time::Duration; -use std::{io, thread}; +use anyhow::anyhow; use parking_lot::Mutex; #[derive(Clone)] @@ -20,7 +21,7 @@ impl StopManager { inner: Arc::new(StopManagerInner::new(f)), } } - pub fn add_listener(&self, name: String, f: F) -> io::Result + pub fn add_listener(&self, name: String, f: F) -> anyhow::Result where F: FnOnce() + Send + 'static, { @@ -61,23 +62,20 @@ impl StopManagerInner { stop_call: Mutex::new(Some(Box::new(f))), } } - fn add_listener(self: &Arc, name: String, f: F) -> io::Result + fn add_listener(self: &Arc, name: String, f: F) -> anyhow::Result where F: FnOnce() + Send + 'static, { if name.is_empty() { - return Err(io::Error::new(io::ErrorKind::Other, "name cannot be empty")); + return Err(anyhow!("name cannot be empty")); } let mut guard = self.listeners.lock(); if guard.0 { - return Err(io::Error::new(io::ErrorKind::Other, "stopped")); + return Err(anyhow!("stopped")); } for (n, _) in &guard.1 { if &name == n { - return Err(io::Error::new( - io::ErrorKind::Other, - format!("stop add_listener {:?} name already exists", name), - )); + return Err(anyhow!("stop add_listener {:?} name already exists", name)); } } guard.1.push((name.clone(), Box::new(f))); diff --git a/vnt/src/util/scheduler.rs b/vnt/src/util/scheduler.rs index 3680ea8..5d5485e 100644 --- a/vnt/src/util/scheduler.rs +++ b/vnt/src/util/scheduler.rs @@ -2,7 +2,6 @@ use crate::util::StopManager; use std::collections::BinaryHeap; use std::{ cmp::Ordering, - io, sync::mpsc::{sync_channel, Receiver, SyncSender}, time::{Duration, Instant}, }; @@ -36,7 +35,7 @@ pub struct Scheduler { sender: SyncSender, } impl Scheduler { - pub fn new(stop_manager: StopManager) -> io::Result { + pub fn new(stop_manager: StopManager) -> anyhow::Result { let (sender, receiver) = sync_channel::(32); let s = Self { sender }; let s_inner = s.clone();