From 9a4cf1dd0709632cf76ce4870d35cdc9b7288203 Mon Sep 17 00:00:00 2001 From: Han Xu Date: Tue, 3 Dec 2024 11:19:10 -0800 Subject: [PATCH 1/7] wip --- Cargo.toml | 3 ++- src/service_daemon.rs | 39 ++++++++++++++++++++++----------------- 2 files changed, 24 insertions(+), 18 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 1c87ecc1..5ef8ba3f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,8 +21,9 @@ fastrand = "2.1" flume = { version = "0.11", default-features = false } # channel between threads if-addrs = { version = "0.13", features = ["link-local"] } # get local IP addresses log = { version = "0.4", optional = true } # logging -polling = "2.1" # select/poll sockets +# polling = "2.1" # select/poll sockets socket2 = { version = "0.5.5", features = ["all"] } # socket APIs +mio = { version = "1.0", features = ["os-poll", "os-ext"] } [dev-dependencies] env_logger = { version = "= 0.10.2", default-features = false, features= ["humantime"] } diff --git a/src/service_daemon.rs b/src/service_daemon.rs index 3affa171..72b01437 100644 --- a/src/service_daemon.rs +++ b/src/service_daemon.rs @@ -43,7 +43,7 @@ use crate::{ }; use flume::{bounded, Sender, TrySendError}; use if_addrs::{IfAddr, Interface}; -use polling::Poller; +use mio::Poll; use socket2::{SockAddr, Socket}; use std::{ cmp::{self, Reverse}, @@ -179,7 +179,7 @@ impl ServiceDaemon { .set_nonblocking(true) .map_err(|e| e_fmt!("failed to set nonblocking for signal socket: {}", e))?; - let poller = Poller::new().map_err(|e| e_fmt!("Failed to create Poller: {}", e))?; + let poller = Poll::new().map_err(|e| e_fmt!("failed to create mio Poll: {e}"))?; let (sender, receiver) = bounded(100); @@ -417,7 +417,7 @@ impl ServiceDaemon { self.send_cmd(Command::Verify(instance_fullname, timeout)) } - fn daemon_thread(signal_sock: UdpSocket, poller: Poller, receiver: Receiver) { + fn daemon_thread(signal_sock: UdpSocket, poller: Poll, receiver: Receiver) { let zc = Zeroconf::new(signal_sock, poller); if let Some(cmd) = Self::run(zc, receiver) { @@ -436,16 +436,16 @@ impl ServiceDaemon { } } - fn handle_poller_events(zc: &mut Zeroconf, events: &[polling::Event]) { + fn handle_poller_events(zc: &mut Zeroconf, events: &mio::Events) { for ev in events.iter() { - trace!("event received with key {}", ev.key); + trace!("event received with key {:?}", ev.token()); if ev.key == SIGNAL_SOCK_EVENT_KEY { // Drain signals as we will drain commands as well. zc.signal_sock_drain(); if let Err(e) = zc .poller - .modify(&zc.signal_sock, polling::Event::readable(ev.key)) + .registry().reregister(&mut zc.signal_sock, ev.token(), mio::Interest::READABLE) { debug!("failed to modify poller for signal socket: {}", e); } @@ -464,7 +464,7 @@ impl ServiceDaemon { // we continue to monitor this socket. if let Some(sock) = zc.intf_socks.get(&intf) { - if let Err(e) = zc.poller.modify(sock, polling::Event::readable(ev.key)) { + if let Err(e) = zc.poller.registry().reregister(sock, ev.token(), mio::Interest::READABLE) { debug!("modify poller for interface {:?}: {}", &intf, e); break; } @@ -482,22 +482,27 @@ impl ServiceDaemon { /// 5. process retransmissions if any. fn run(mut zc: Zeroconf, receiver: Receiver) -> Option { // Add the daemon's signal socket to the poller. - if let Err(e) = zc.poller.add( - &zc.signal_sock, - polling::Event::readable(SIGNAL_SOCK_EVENT_KEY), + if let Err(e) = zc.poller.registry().register( + &mut zc.signal_sock, + mio::Token(SIGNAL_SOCK_EVENT_KEY), mio::Interest::READABLE ) { debug!("failed to add signal socket to the poller: {}", e); return None; } // Add mDNS sockets to the poller. - for (intf, sock) in zc.intf_socks.iter() { + for (intf, sock) in zc.intf_socks.iter_mut() { let key = Zeroconf::add_poll_impl(&mut zc.poll_ids, &mut zc.poll_id_count, intf.clone()); - if let Err(e) = zc.poller.add(sock, polling::Event::readable(key)) { - debug!("add socket of {:?} to poller: {}", intf, e); + + if let Err(e) = zc.poller.registry().register(sock, mio::Token(key), mio::Interest::READABLE) { + println!("add socket of {:?} to poller: {e}", intf); return None; } + // if let Err(e) = zc.poller.add(sock, polling::Event::readable(key)) { + // debug!("add socket of {:?} to poller: {}", intf, e); + // return None; + // } } // Setup timer for IP checks. @@ -507,7 +512,7 @@ impl ServiceDaemon { // Start the run loop. - let mut events = Vec::new(); + let mut events = mio::Events::with_capacity(1024); loop { let now = current_time_millis(); @@ -520,7 +525,7 @@ impl ServiceDaemon { // Process incoming packets, command events and optional timeout. events.clear(); - match zc.poller.wait(&mut events, timeout) { + match zc.poller.poll(&mut events, timeout) { Ok(_) => Self::handle_poller_events(&mut zc, &events), Err(e) => debug!("failed to select from sockets: {}", e), } @@ -865,7 +870,7 @@ struct Zeroconf { counters: Metrics, /// Waits for incoming packets. - poller: Poller, + poller: Poll, /// Channels to notify events. monitors: Vec>, @@ -896,7 +901,7 @@ struct Zeroconf { } impl Zeroconf { - fn new(signal_sock: UdpSocket, poller: Poller) -> Self { + fn new(signal_sock: UdpSocket, poller: Poll) -> Self { // Get interfaces. let my_ifaddrs = my_ip_interfaces(); From edc17ead6cfc68723ff5f5f6a7462f894fbed388 Mon Sep 17 00:00:00 2001 From: Han Xu Date: Tue, 3 Dec 2024 21:55:00 -0800 Subject: [PATCH 2/7] mio socket works --- Cargo.toml | 3 +- src/service_daemon.rs | 155 ++++++++++++++++++++++++++++-------------- 2 files changed, 106 insertions(+), 52 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 5ef8ba3f..a450e1d2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,9 +21,8 @@ fastrand = "2.1" flume = { version = "0.11", default-features = false } # channel between threads if-addrs = { version = "0.13", features = ["link-local"] } # get local IP addresses log = { version = "0.4", optional = true } # logging -# polling = "2.1" # select/poll sockets +mio = { version = "1.0", features = ["os-poll", "net"] } # select/poll sockets socket2 = { version = "0.5.5", features = ["all"] } # socket APIs -mio = { version = "1.0", features = ["os-poll", "os-ext"] } [dev-dependencies] env_logger = { version = "= 0.10.2", default-features = false, features= ["humantime"] } diff --git a/src/service_daemon.rs b/src/service_daemon.rs index 72b01437..2ef55218 100644 --- a/src/service_daemon.rs +++ b/src/service_daemon.rs @@ -44,12 +44,11 @@ use crate::{ use flume::{bounded, Sender, TrySendError}; use if_addrs::{IfAddr, Interface}; use mio::Poll; -use socket2::{SockAddr, Socket}; +use socket2::Socket; use std::{ cmp::{self, Reverse}, collections::{BinaryHeap, HashMap, HashSet}, fmt, - io::Read, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, UdpSocket}, str, thread, time::Duration, @@ -184,9 +183,10 @@ impl ServiceDaemon { let (sender, receiver) = bounded(100); // Spawn the daemon thread + let mio_sock = mio::net::UdpSocket::from_std(signal_sock); thread::Builder::new() .name("mDNS_daemon".to_string()) - .spawn(move || Self::daemon_thread(signal_sock, poller, receiver)) + .spawn(move || Self::daemon_thread(mio_sock, poller, receiver)) .map_err(|e| e_fmt!("thread builder failed to spawn: {}", e))?; Ok(Self { @@ -417,7 +417,7 @@ impl ServiceDaemon { self.send_cmd(Command::Verify(instance_fullname, timeout)) } - fn daemon_thread(signal_sock: UdpSocket, poller: Poll, receiver: Receiver) { + fn daemon_thread(signal_sock: mio::net::UdpSocket, poller: Poll, receiver: Receiver) { let zc = Zeroconf::new(signal_sock, poller); if let Some(cmd) = Self::run(zc, receiver) { @@ -438,33 +438,38 @@ impl ServiceDaemon { fn handle_poller_events(zc: &mut Zeroconf, events: &mio::Events) { for ev in events.iter() { - trace!("event received with key {:?}", ev.token()); - if ev.key == SIGNAL_SOCK_EVENT_KEY { + debug!("event received with key {:?}", ev.token()); + if ev.token().0 == SIGNAL_SOCK_EVENT_KEY { // Drain signals as we will drain commands as well. zc.signal_sock_drain(); - if let Err(e) = zc - .poller - .registry().reregister(&mut zc.signal_sock, ev.token(), mio::Interest::READABLE) - { + if let Err(e) = zc.poller.registry().reregister( + &mut zc.signal_sock, + ev.token(), + mio::Interest::READABLE, + ) { debug!("failed to modify poller for signal socket: {}", e); } continue; // Next event. } // Read until no more packets available. - let intf = match zc.poll_ids.get(&ev.key) { + let intf = match zc.poll_ids.get(&ev.token().0) { Some(interface) => interface.clone(), None => { - debug!("Ip for event key {} not found", ev.key); + debug!("Ip for event key {} not found", ev.token().0); break; } }; while zc.handle_read(&intf) {} // we continue to monitor this socket. - if let Some(sock) = zc.intf_socks.get(&intf) { - if let Err(e) = zc.poller.registry().reregister(sock, ev.token(), mio::Interest::READABLE) { + if let Some(sock) = zc.intf_socks.get_mut(&intf) { + if let Err(e) = + zc.poller + .registry() + .reregister(sock, ev.token(), mio::Interest::READABLE) + { debug!("modify poller for interface {:?}: {}", &intf, e); break; } @@ -484,7 +489,8 @@ impl ServiceDaemon { // Add the daemon's signal socket to the poller. if let Err(e) = zc.poller.registry().register( &mut zc.signal_sock, - mio::Token(SIGNAL_SOCK_EVENT_KEY), mio::Interest::READABLE + mio::Token(SIGNAL_SOCK_EVENT_KEY), + mio::Interest::READABLE, ) { debug!("failed to add signal socket to the poller: {}", e); return None; @@ -495,14 +501,14 @@ impl ServiceDaemon { let key = Zeroconf::add_poll_impl(&mut zc.poll_ids, &mut zc.poll_id_count, intf.clone()); - if let Err(e) = zc.poller.registry().register(sock, mio::Token(key), mio::Interest::READABLE) { - println!("add socket of {:?} to poller: {e}", intf); + if let Err(e) = + zc.poller + .registry() + .register(sock, mio::Token(key), mio::Interest::READABLE) + { + debug!("add socket of {:?} to poller: {e}", intf); return None; } - // if let Err(e) = zc.poller.add(sock, polling::Event::readable(key)) { - // debug!("add socket of {:?} to poller: {}", intf, e); - // return None; - // } } // Setup timer for IP checks. @@ -631,7 +637,7 @@ impl ServiceDaemon { } /// Creates a new UDP socket that uses `intf` to send and recv multicast. -fn new_socket_bind(intf: &Interface) -> Result { +fn new_socket_bind(intf: &Interface) -> Result { // Use the same socket for receiving and sending multicast packets. // Such socket has to bind to INADDR_ANY or IN6ADDR_ANY. let intf_ip = &intf.ip(); @@ -655,7 +661,7 @@ fn new_socket_bind(intf: &Interface) -> Result { sock.send_to(&packet, &multicast_addr) .map_err(|e| e_fmt!("send multicast packet on addr {}: {}", ip, e))?; } - Ok(sock) + Ok(UdpSocket::from(sock)) } IpAddr::V6(ip) => { let addr = SocketAddrV6::new(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0), MDNS_PORT, 0, 0); @@ -673,7 +679,7 @@ fn new_socket_bind(intf: &Interface) -> Result { // be many IPv6 interfaces on a host and could cause such send error: // "No buffer space available (os error 55)". - Ok(sock) + Ok(UdpSocket::from(sock)) } } } @@ -839,7 +845,7 @@ struct IfSelection { /// A struct holding the state. It was inspired by `zeroconf` package in Python. struct Zeroconf { /// Local interfaces with sockets to recv/send on these interfaces. - intf_socks: HashMap, + intf_socks: HashMap, /// Map poll id to Interface. poll_ids: HashMap, @@ -882,7 +888,7 @@ struct Zeroconf { if_selections: Vec, /// Socket for signaling. - signal_sock: UdpSocket, + signal_sock: mio::net::UdpSocket, /// Timestamps marking where we need another iteration of the run loop, /// to react to events like retransmissions, cache refreshes, interface IP address changes, etc. @@ -901,7 +907,7 @@ struct Zeroconf { } impl Zeroconf { - fn new(signal_sock: UdpSocket, poller: Poll) -> Self { + fn new(signal_sock: mio::net::UdpSocket, poller: Poll) -> Self { // Get interfaces. let my_ifaddrs = my_ip_interfaces(); @@ -922,7 +928,7 @@ impl Zeroconf { dns_registry_map.insert(intf.clone(), DnsRegistry::new()); - intf_socks.insert(intf, sock); + intf_socks.insert(intf, mio::net::UdpSocket::from_std(sock)); } let monitors = Vec::new(); @@ -1092,8 +1098,8 @@ impl Zeroconf { } } else { // Remove the interface - if let Some(sock) = self.intf_socks.remove(&intf) { - if let Err(e) = self.poller.delete(&sock) { + if let Some(mut sock) = self.intf_socks.remove(&intf) { + if let Err(e) = self.poller.registry().deregister(&mut sock) { debug!("process_if_selections: poller.delete {:?}: {}", &intf, e); } // Remove from poll_ids @@ -1113,10 +1119,10 @@ impl Zeroconf { // Remove unused sockets in the poller. let deleted_addrs = self .intf_socks - .iter() + .iter_mut() .filter_map(|(intf, sock)| { if !my_ifaddrs.contains(intf) { - if let Err(e) = poller.delete(sock) { + if let Err(e) = poller.registry().deregister(sock) { debug!("check_ip_changes: poller.delete {:?}: {}", intf, e); } // Remove from poll_ids @@ -1144,8 +1150,8 @@ impl Zeroconf { fn add_new_interface(&mut self, intf: Interface) { // Bind the new interface. let new_ip = intf.ip(); - let sock = match new_socket_bind(&intf) { - Ok(s) => s, + let mut sock = match new_socket_bind(&intf) { + Ok(s) => mio::net::UdpSocket::from_std(s), Err(e) => { debug!("bind a socket to {}: {}. Skipped.", &intf.ip(), e); return; @@ -1154,7 +1160,11 @@ impl Zeroconf { // Add the new interface into the poller. let key = self.add_poll(intf.clone()); - if let Err(e) = self.poller.add(&sock, polling::Event::readable(key)) { + if let Err(e) = + self.poller + .registry() + .register(&mut sock, mio::Token(key), mio::Interest::READABLE) + { debug!("check_ip_changes: poller add ip {}: {}", new_ip, e); return; } @@ -1422,7 +1432,12 @@ impl Zeroconf { } } - fn unregister_service(&self, info: &ServiceInfo, intf: &Interface, sock: &Socket) -> Vec { + fn unregister_service( + &self, + info: &ServiceInfo, + intf: &Interface, + sock: &mio::net::UdpSocket, + ) -> Vec { let mut out = DnsOutgoing::new(FLAGS_QR_RESPONSE | FLAGS_AA); out.add_answer_at_time( DnsPointer::new( @@ -1512,7 +1527,7 @@ impl Zeroconf { /// Sends out a list of `questions` (i.e. DNS questions) via multicast. fn send_query_vec(&self, questions: &[(&str, RRType)]) { - trace!("Sending query questions: {:?}", questions); + debug!("Sending query questions: {:?}", questions); let mut out = DnsOutgoing::new(FLAGS_QR_QUERY); let now = current_time_millis(); @@ -1564,7 +1579,7 @@ impl Zeroconf { // be truncated by the socket layer depending on the platform's libc. // In any case, such large datagram will not be decoded properly and // this function should return false but should not crash. - let sz = match sock.read(&mut buf) { + let sz = match sock.recv(&mut buf) { Ok(sz) => sz, Err(e) => { if e.kind() != std::io::ErrorKind::WouldBlock { @@ -1578,8 +1593,8 @@ impl Zeroconf { // If sz is 0, it means sock reached End-of-File. if sz == 0 { - debug!("socket {:?} was likely shutdown", sock); - if let Err(e) = self.poller.delete(&*sock) { + debug!("socket {:?} was likely shutdown", &sock); + if let Err(e) = self.poller.registry().deregister(sock) { debug!("failed to remove sock {:?} from poller: {}", sock, &e); } @@ -1587,7 +1602,8 @@ impl Zeroconf { match new_socket_bind(intf) { Ok(new_sock) => { trace!("reset socket for IP {}", intf.ip()); - self.intf_socks.insert(intf.clone(), new_sock); + self.intf_socks + .insert(intf.clone(), mio::net::UdpSocket::from_std(new_sock)); } Err(e) => debug!("re-bind a socket to {:?}: {}", intf, e), } @@ -2091,7 +2107,7 @@ impl Zeroconf { }; for question in msg.questions.iter() { - trace!("query question: {:?}", &question); + debug!("query question: {:?}", &question); let qtype = question.entry.ty; @@ -3024,7 +3040,11 @@ fn my_ip_interfaces() -> Vec { } /// Send an outgoing mDNS query or response, and returns the packet bytes. -fn send_dns_outgoing(out: &DnsOutgoing, intf: &Interface, sock: &Socket) -> Vec> { +fn send_dns_outgoing( + out: &DnsOutgoing, + intf: &Interface, + sock: &mio::net::UdpSocket, +) -> Vec> { let qtype = if out.is_query() { "query" } else { "response" }; trace!( "send outgoing {}: {} questions {} answers {} authorities {} additional", @@ -3042,7 +3062,7 @@ fn send_dns_outgoing(out: &DnsOutgoing, intf: &Interface, sock: &Socket) -> Vec< } /// Sends a multicast packet, and returns the packet bytes. -fn multicast_on_intf(packet: &[u8], intf: &Interface, socket: &Socket) { +fn multicast_on_intf(packet: &[u8], intf: &Interface, socket: &mio::net::UdpSocket) { if packet.len() > MAX_MSG_ABSOLUTE { debug!("Drop over-sized packet ({})", packet.len()); return; @@ -3061,9 +3081,8 @@ fn multicast_on_intf(packet: &[u8], intf: &Interface, socket: &Socket) { } /// Sends out `packet` to `addr` on the socket in `intf_sock`. -fn send_packet(packet: &[u8], addr: SocketAddr, intf: &Interface, sock: &Socket) { - let sockaddr = SockAddr::from(addr); - match sock.send_to(packet, &sockaddr) { +fn send_packet(packet: &[u8], addr: SocketAddr, intf: &Interface, sock: &mio::net::UdpSocket) { + match sock.send_to(packet, addr) { Ok(sz) => trace!("sent out {} bytes on interface {:?}", sz, intf), Err(e) => debug!("Failed to send to {} via {:?}: {}", addr, &intf, e), } @@ -3229,7 +3248,7 @@ fn announce_service_on_intf( dns_registry: &mut DnsRegistry, info: &ServiceInfo, intf: &Interface, - sock: &Socket, + sock: &mio::net::UdpSocket, ) -> bool { if let Some(out) = prepare_announce(info, intf, dns_registry) { send_dns_outgoing(&out, intf, sock); @@ -3302,6 +3321,41 @@ fn hostname_change(original: &str) -> String { parts.join(".") } +// #[derive(Debug)] +// struct MySocket(Socket); + +// impl mio::event::Source for MySocket { +// fn register( +// &mut self, +// registry: &mio::Registry, +// token: mio::Token, +// interests: mio::Interest, +// ) -> std::io::Result<()> { +// let owned_fd = self.0.as_fd().try_clone_to_owned()?; +// let mut mio_socket = mio::net::UdpSocket::from(owned_fd); +// // let mut mio_stream = mio::net::TcpStream::from(owned_fd); +// registry.register(&mut mio_socket, token, interests) +// } + +// fn deregister(&mut self, registry: &mio::Registry) -> std::io::Result<()> { +// let my_fd = self.0.as_fd().try_clone_to_owned()?; +// let mut mio_socket = mio::net::UdpSocket::from(my_fd); +// // let mut mio_stream = mio::net::TcpStream::from(my_fd); +// registry.deregister(&mut mio_socket) +// } + +// fn reregister( +// &mut self, +// registry: &mio::Registry, +// token: mio::Token, +// interests: mio::Interest, +// ) -> std::io::Result<()> { +// let my_fd = self.0.as_fd().try_clone_to_owned()?; +// let mut mio_socket = mio::net::UdpSocket::from(my_fd); +// registry.reregister(&mut mio_socket, token, interests) +// } +// } + #[cfg(test)] mod tests { use super::{ @@ -3457,7 +3511,8 @@ mod tests { for intf in intfs { let sock = new_socket_bind(&intf).unwrap(); - send_dns_outgoing(&packet_buffer, &intf, &sock); + let udp_sock = mio::net::UdpSocket::from_std(sock); + send_dns_outgoing(&packet_buffer, &intf, &udp_sock); } println!( @@ -3647,7 +3702,7 @@ mod tests { let mdns_client = ServiceDaemon::new().expect("Failed to create mdns client"); let browse_chan = mdns_client.browse(service_type).unwrap(); - let timeout = Duration::from_secs(1); + let timeout = Duration::from_secs(2); let mut resolved = false; // resolve the service first. From 1b2f94f07ccc5f31229727949e9d2f8f21c60455 Mon Sep 17 00:00:00 2001 From: Han Xu Date: Tue, 3 Dec 2024 21:56:19 -0800 Subject: [PATCH 3/7] remove unused code --- src/service_daemon.rs | 35 ----------------------------------- 1 file changed, 35 deletions(-) diff --git a/src/service_daemon.rs b/src/service_daemon.rs index 2ef55218..87e35f55 100644 --- a/src/service_daemon.rs +++ b/src/service_daemon.rs @@ -3321,41 +3321,6 @@ fn hostname_change(original: &str) -> String { parts.join(".") } -// #[derive(Debug)] -// struct MySocket(Socket); - -// impl mio::event::Source for MySocket { -// fn register( -// &mut self, -// registry: &mio::Registry, -// token: mio::Token, -// interests: mio::Interest, -// ) -> std::io::Result<()> { -// let owned_fd = self.0.as_fd().try_clone_to_owned()?; -// let mut mio_socket = mio::net::UdpSocket::from(owned_fd); -// // let mut mio_stream = mio::net::TcpStream::from(owned_fd); -// registry.register(&mut mio_socket, token, interests) -// } - -// fn deregister(&mut self, registry: &mio::Registry) -> std::io::Result<()> { -// let my_fd = self.0.as_fd().try_clone_to_owned()?; -// let mut mio_socket = mio::net::UdpSocket::from(my_fd); -// // let mut mio_stream = mio::net::TcpStream::from(my_fd); -// registry.deregister(&mut mio_socket) -// } - -// fn reregister( -// &mut self, -// registry: &mio::Registry, -// token: mio::Token, -// interests: mio::Interest, -// ) -> std::io::Result<()> { -// let my_fd = self.0.as_fd().try_clone_to_owned()?; -// let mut mio_socket = mio::net::UdpSocket::from(my_fd); -// registry.reregister(&mut mio_socket, token, interests) -// } -// } - #[cfg(test)] mod tests { use super::{ From bd6c78b955a8a36fee25b9b19a98047b08be09c5 Mon Sep 17 00:00:00 2001 From: Han Xu Date: Tue, 3 Dec 2024 22:07:43 -0800 Subject: [PATCH 4/7] bump up rustc to 1.70.0 --- .github/workflows/build.yml | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index ba34c8ac..6bceee7f 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -19,7 +19,7 @@ jobs: os: [ubuntu-20.04, windows-latest, macos-latest] steps: - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@1.65.0 + - uses: dtolnay/rust-toolchain@1.70.0 with: components: rustfmt, clippy - name: Run rustfmt and fail if any warnings diff --git a/Cargo.toml b/Cargo.toml index a450e1d2..215c58e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ name = "mdns-sd" version = "0.12.0" authors = ["keepsimple "] edition = "2018" -rust-version = "1.65.0" +rust-version = "1.70.0" license = "Apache-2.0 OR MIT" repository = "https://github.com/keepsimple1/mdns-sd" documentation = "https://docs.rs/mdns-sd" From 9bfa8bd40e0fce0886437d9cfad897969e8abcab Mon Sep 17 00:00:00 2001 From: Han Xu Date: Wed, 4 Dec 2024 11:06:35 -0800 Subject: [PATCH 5/7] simplify UdpSocket conversion --- src/service_daemon.rs | 44 +++++++++++++++++++------------------------ 1 file changed, 19 insertions(+), 25 deletions(-) diff --git a/src/service_daemon.rs b/src/service_daemon.rs index 87e35f55..d2e77be1 100644 --- a/src/service_daemon.rs +++ b/src/service_daemon.rs @@ -43,7 +43,7 @@ use crate::{ }; use flume::{bounded, Sender, TrySendError}; use if_addrs::{IfAddr, Interface}; -use mio::Poll; +use mio::{net::UdpSocket as MioUdpSocket, Poll}; use socket2::Socket; use std::{ cmp::{self, Reverse}, @@ -183,7 +183,7 @@ impl ServiceDaemon { let (sender, receiver) = bounded(100); // Spawn the daemon thread - let mio_sock = mio::net::UdpSocket::from_std(signal_sock); + let mio_sock = MioUdpSocket::from_std(signal_sock); thread::Builder::new() .name("mDNS_daemon".to_string()) .spawn(move || Self::daemon_thread(mio_sock, poller, receiver)) @@ -417,7 +417,7 @@ impl ServiceDaemon { self.send_cmd(Command::Verify(instance_fullname, timeout)) } - fn daemon_thread(signal_sock: mio::net::UdpSocket, poller: Poll, receiver: Receiver) { + fn daemon_thread(signal_sock: MioUdpSocket, poller: Poll, receiver: Receiver) { let zc = Zeroconf::new(signal_sock, poller); if let Some(cmd) = Self::run(zc, receiver) { @@ -637,7 +637,7 @@ impl ServiceDaemon { } /// Creates a new UDP socket that uses `intf` to send and recv multicast. -fn new_socket_bind(intf: &Interface) -> Result { +fn new_socket_bind(intf: &Interface) -> Result { // Use the same socket for receiving and sending multicast packets. // Such socket has to bind to INADDR_ANY or IN6ADDR_ANY. let intf_ip = &intf.ip(); @@ -661,7 +661,7 @@ fn new_socket_bind(intf: &Interface) -> Result { sock.send_to(&packet, &multicast_addr) .map_err(|e| e_fmt!("send multicast packet on addr {}: {}", ip, e))?; } - Ok(UdpSocket::from(sock)) + Ok(MioUdpSocket::from_std(UdpSocket::from(sock))) } IpAddr::V6(ip) => { let addr = SocketAddrV6::new(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0), MDNS_PORT, 0, 0); @@ -679,7 +679,7 @@ fn new_socket_bind(intf: &Interface) -> Result { // be many IPv6 interfaces on a host and could cause such send error: // "No buffer space available (os error 55)". - Ok(UdpSocket::from(sock)) + Ok(MioUdpSocket::from_std(UdpSocket::from(sock))) } } } @@ -845,7 +845,7 @@ struct IfSelection { /// A struct holding the state. It was inspired by `zeroconf` package in Python. struct Zeroconf { /// Local interfaces with sockets to recv/send on these interfaces. - intf_socks: HashMap, + intf_socks: HashMap, /// Map poll id to Interface. poll_ids: HashMap, @@ -888,7 +888,7 @@ struct Zeroconf { if_selections: Vec, /// Socket for signaling. - signal_sock: mio::net::UdpSocket, + signal_sock: MioUdpSocket, /// Timestamps marking where we need another iteration of the run loop, /// to react to events like retransmissions, cache refreshes, interface IP address changes, etc. @@ -907,7 +907,7 @@ struct Zeroconf { } impl Zeroconf { - fn new(signal_sock: mio::net::UdpSocket, poller: Poll) -> Self { + fn new(signal_sock: MioUdpSocket, poller: Poll) -> Self { // Get interfaces. let my_ifaddrs = my_ip_interfaces(); @@ -928,7 +928,7 @@ impl Zeroconf { dns_registry_map.insert(intf.clone(), DnsRegistry::new()); - intf_socks.insert(intf, mio::net::UdpSocket::from_std(sock)); + intf_socks.insert(intf, sock); } let monitors = Vec::new(); @@ -1151,7 +1151,7 @@ impl Zeroconf { // Bind the new interface. let new_ip = intf.ip(); let mut sock = match new_socket_bind(&intf) { - Ok(s) => mio::net::UdpSocket::from_std(s), + Ok(s) => s, Err(e) => { debug!("bind a socket to {}: {}. Skipped.", &intf.ip(), e); return; @@ -1436,7 +1436,7 @@ impl Zeroconf { &self, info: &ServiceInfo, intf: &Interface, - sock: &mio::net::UdpSocket, + sock: &MioUdpSocket, ) -> Vec { let mut out = DnsOutgoing::new(FLAGS_QR_RESPONSE | FLAGS_AA); out.add_answer_at_time( @@ -1602,8 +1602,7 @@ impl Zeroconf { match new_socket_bind(intf) { Ok(new_sock) => { trace!("reset socket for IP {}", intf.ip()); - self.intf_socks - .insert(intf.clone(), mio::net::UdpSocket::from_std(new_sock)); + self.intf_socks.insert(intf.clone(), new_sock); } Err(e) => debug!("re-bind a socket to {:?}: {}", intf, e), } @@ -3040,11 +3039,7 @@ fn my_ip_interfaces() -> Vec { } /// Send an outgoing mDNS query or response, and returns the packet bytes. -fn send_dns_outgoing( - out: &DnsOutgoing, - intf: &Interface, - sock: &mio::net::UdpSocket, -) -> Vec> { +fn send_dns_outgoing(out: &DnsOutgoing, intf: &Interface, sock: &MioUdpSocket) -> Vec> { let qtype = if out.is_query() { "query" } else { "response" }; trace!( "send outgoing {}: {} questions {} answers {} authorities {} additional", @@ -3062,7 +3057,7 @@ fn send_dns_outgoing( } /// Sends a multicast packet, and returns the packet bytes. -fn multicast_on_intf(packet: &[u8], intf: &Interface, socket: &mio::net::UdpSocket) { +fn multicast_on_intf(packet: &[u8], intf: &Interface, socket: &MioUdpSocket) { if packet.len() > MAX_MSG_ABSOLUTE { debug!("Drop over-sized packet ({})", packet.len()); return; @@ -3081,7 +3076,7 @@ fn multicast_on_intf(packet: &[u8], intf: &Interface, socket: &mio::net::UdpSock } /// Sends out `packet` to `addr` on the socket in `intf_sock`. -fn send_packet(packet: &[u8], addr: SocketAddr, intf: &Interface, sock: &mio::net::UdpSocket) { +fn send_packet(packet: &[u8], addr: SocketAddr, intf: &Interface, sock: &MioUdpSocket) { match sock.send_to(packet, addr) { Ok(sz) => trace!("sent out {} bytes on interface {:?}", sz, intf), Err(e) => debug!("Failed to send to {} via {:?}: {}", addr, &intf, e), @@ -3248,7 +3243,7 @@ fn announce_service_on_intf( dns_registry: &mut DnsRegistry, info: &ServiceInfo, intf: &Interface, - sock: &mio::net::UdpSocket, + sock: &MioUdpSocket, ) -> bool { if let Some(out) = prepare_announce(info, intf, dns_registry) { send_dns_outgoing(&out, intf, sock); @@ -3476,8 +3471,7 @@ mod tests { for intf in intfs { let sock = new_socket_bind(&intf).unwrap(); - let udp_sock = mio::net::UdpSocket::from_std(sock); - send_dns_outgoing(&packet_buffer, &intf, &udp_sock); + send_dns_outgoing(&packet_buffer, &intf, &sock); } println!( @@ -3667,7 +3661,7 @@ mod tests { let mdns_client = ServiceDaemon::new().expect("Failed to create mdns client"); let browse_chan = mdns_client.browse(service_type).unwrap(); - let timeout = Duration::from_secs(2); + let timeout = Duration::from_secs(1); let mut resolved = false; // resolve the service first. From e18b00d5533565dfbfc15741360998044500db58 Mon Sep 17 00:00:00 2001 From: Han Xu Date: Wed, 4 Dec 2024 22:24:52 -0800 Subject: [PATCH 6/7] fix README --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 228fe956..53cfea68 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ [![Build](https://github.com/keepsimple1/mdns-sd/actions/workflows/build.yml/badge.svg)](https://github.com/keepsimple1/mdns-sd/actions) [![Cargo](https://img.shields.io/crates/v/mdns-sd.svg)](https://crates.io/crates/mdns-sd) [![docs.rs](https://img.shields.io/docsrs/mdns-sd)](https://docs.rs/mdns-sd/latest/mdns_sd/) -[![Rust version: 1.63+](https://img.shields.io/badge/rust%20version-1.63+-orange)](https://blog.rust-lang.org/2022/08/11/Rust-1.63.0.html) +[![Rust version: 1.70+](https://img.shields.io/badge/rust%20version-1.70+-orange)](https://blog.rust-lang.org/2022/08/11/Rust-1.70.0.html) This is a small implementation of mDNS (Multicast DNS) based service discovery in safe Rust, with a small set of dependencies. Some highlights: From 9d39477b2831f59c84876c46f596f8bc5002b99b Mon Sep 17 00:00:00 2001 From: Han Xu Date: Wed, 4 Dec 2024 22:29:11 -0800 Subject: [PATCH 7/7] change some debug log back to trace log --- src/service_daemon.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/service_daemon.rs b/src/service_daemon.rs index d2e77be1..53462bc6 100644 --- a/src/service_daemon.rs +++ b/src/service_daemon.rs @@ -438,7 +438,7 @@ impl ServiceDaemon { fn handle_poller_events(zc: &mut Zeroconf, events: &mio::Events) { for ev in events.iter() { - debug!("event received with key {:?}", ev.token()); + trace!("event received with key {:?}", ev.token()); if ev.token().0 == SIGNAL_SOCK_EVENT_KEY { // Drain signals as we will drain commands as well. zc.signal_sock_drain(); @@ -1527,7 +1527,7 @@ impl Zeroconf { /// Sends out a list of `questions` (i.e. DNS questions) via multicast. fn send_query_vec(&self, questions: &[(&str, RRType)]) { - debug!("Sending query questions: {:?}", questions); + trace!("Sending query questions: {:?}", questions); let mut out = DnsOutgoing::new(FLAGS_QR_QUERY); let now = current_time_millis(); @@ -2106,7 +2106,7 @@ impl Zeroconf { }; for question in msg.questions.iter() { - debug!("query question: {:?}", &question); + trace!("query question: {:?}", &question); let qtype = question.entry.ty;