From 79716db2e00521efe7db7913d3f5becfb64edef5 Mon Sep 17 00:00:00 2001 From: Vicki Pfau Date: Fri, 24 Jan 2025 18:21:58 -0800 Subject: [PATCH] udev: Use new udev 0.9.3 version to vastly simplify udev code --- Cargo.lock | 4 +- Cargo.toml | 2 +- src/lib.rs | 1 - src/thread.rs | 113 --------------------------------------------- src/udev.rs | 124 +++++++++++++++++++------------------------------- src/wifi.rs | 4 +- 6 files changed, 53 insertions(+), 195 deletions(-) delete mode 100644 src/thread.rs diff --git a/Cargo.lock b/Cargo.lock index fe8420a..ebf1331 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1042,9 +1042,9 @@ dependencies = [ [[package]] name = "udev" -version = "0.9.2" +version = "0.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e28ba5932aebe14104df97f3c48feba2b6a398ae6b6f4ed91831e5fb059d992e" +checksum = "af4e37e9ea4401fc841ff54b9ddfc9be1079b1e89434c1a6a865dd68980f7e9f" dependencies = [ "io-lifetimes", "libc", diff --git a/Cargo.toml b/Cargo.toml index 0b09b8e..8732f28 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,7 +26,7 @@ tokio-util = { version = "0.7", default-features = false } toml = "0.8" tracing = { version = "0.1", default-features = false } tracing-subscriber = { version = "0.3", default-features = false, features = ["env-filter", "fmt"] } -udev = "0.9" +udev = { version = "^0.9.3", features = ["send", "sync"] } xdg = "2.5" zbus = { version = "5", default-features = false, features = ["tokio"] } zbus_xml = "5" diff --git a/src/lib.rs b/src/lib.rs index 8d41061..7f97bb8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,7 +26,6 @@ mod platform; mod process; mod sls; mod systemd; -mod thread; mod udev; pub mod cec; diff --git a/src/thread.rs b/src/thread.rs deleted file mode 100644 index 9a11e2b..0000000 --- a/src/thread.rs +++ /dev/null @@ -1,113 +0,0 @@ -/* - * Copyright © 2023 Collabora Ltd. - * Copyright © 2024 Valve Software - * - * SPDX-License-Identifier: MIT - */ - -use std::future::Future; -use std::pin::Pin; -use std::sync::{Arc, Mutex}; -use std::task::{Context, Poll, Waker}; -use std::thread::{self, JoinHandle}; - -pub(crate) struct AsyncJoinHandle -where - T: Send + 'static, -{ - join_handle: Option>, - context: Arc>, -} - -struct JoinContext { - waker: Option, - exited: bool, -} - -struct JoinGuard { - context: Arc>, -} - -impl Future for AsyncJoinHandle { - type Output = T; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let this = Pin::into_inner(self); - let guard = this.context.lock(); - let mut context = guard.unwrap(); - context.waker.replace(cx.waker().clone()); - if let Some(join_handle) = this.join_handle.as_mut() { - if join_handle.is_finished() || context.exited { - let join_handle = this.join_handle.take().unwrap(); - return Poll::Ready(join_handle.join().unwrap()); - } - } - Poll::Pending - } -} - -impl Drop for JoinGuard { - fn drop(&mut self) { - let guard = self.context.lock(); - let mut context = guard.unwrap(); - context.exited = true; - let waker = context.waker.take(); - if let Some(waker) = waker { - waker.wake(); - } - } -} - -pub(crate) fn spawn(f: F) -> AsyncJoinHandle -where - F: FnOnce() -> T + Send + 'static, - T: Send + 'static, -{ - let context = Arc::new(Mutex::new(JoinContext { - waker: None, - exited: false, - })); - - let thread_context = context.clone(); - let join_handle = Some(thread::spawn(move || { - let _guard = JoinGuard { - context: thread_context, - }; - f() - })); - - AsyncJoinHandle { - join_handle, - context, - } -} - -#[cfg(test)] -mod test { - use super::*; - use std::thread::sleep as sync_sleep; - use std::time::Duration; - use tokio::time::sleep as async_sleep; - - #[tokio::test] - async fn test_join() { - let handle = spawn(|| true); - assert!(handle.await); - } - - #[tokio::test] - async fn test_slow_join() { - let handle = spawn(|| true); - async_sleep(Duration::from_millis(100)).await; - assert!(handle.await); - } - - #[tokio::test] - async fn test_slow_thread() { - let handle = spawn(|| { - sync_sleep(Duration::from_millis(100)); - true - }); - assert!(handle.await); - } -} diff --git a/src/udev.rs b/src/udev.rs index f5daa50..82b2b30 100644 --- a/src/udev.rs +++ b/src/udev.rs @@ -6,19 +6,20 @@ */ use anyhow::{anyhow, bail, ensure, Result}; -use nix::poll::{poll, PollFd, PollFlags, PollTimeout}; -use nix::unistd::pipe; -use std::os::fd::{AsFd, AsRawFd, OwnedFd}; +use std::os::fd::AsFd; use std::path::PathBuf; -use std::sync::mpsc::channel; -use tokio::net::unix::pipe::Sender; -use tokio::sync::mpsc::{unbounded_channel, UnboundedSender}; +use std::time::Duration; +use tokio::io::unix::AsyncFd; +use tokio::io::Interest; +use tokio::select; +use tokio::sync::mpsc::{channel, unbounded_channel, Receiver, Sender, UnboundedSender}; +use tokio::task::{spawn, JoinHandle}; +use tokio::time::sleep; use tracing::debug; use udev::{Event, EventType, MonitorBuilder}; use zbus::object_server::{InterfaceRef, SignalEmitter}; use zbus::{self, interface, Connection}; -use crate::thread::{spawn, AsyncJoinHandle}; use crate::Service; const PATH: &str = "/com/steampowered/SteamOSManager1"; @@ -27,8 +28,8 @@ pub(crate) struct UdevMonitor where Self: 'static + Send, { - shutdown_sender: Sender, - shutdown_receiver: Option, + shutdown_sender: Sender<()>, + shutdown_receiver: Option>, udev_object: InterfaceRef, } @@ -54,12 +55,12 @@ impl Service for UdevMonitor { .shutdown_receiver .take() .ok_or(anyhow!("UdevMonitor cannot be run twice"))?; - let mut handle = spawn(move || run_udev(&ev_sender, &shutdown_receiver)); + let mut handle = spawn(run_udev(ev_sender, shutdown_receiver)); loop { let handle = &mut handle; let ev = tokio::select! { - r = handle => break r, + r = handle => break r?, r = ev_receiver.recv() => r.ok_or(anyhow!("udev event pipe broke"))?, }; match ev { @@ -81,7 +82,7 @@ impl Service for UdevMonitor { } async fn shutdown(&mut self) -> Result<()> { - self.shutdown_sender.try_write(&[0u8])?; + let _ = self.shutdown_sender.send(()).await; Ok(()) } } @@ -94,10 +95,10 @@ impl UdevMonitor { "Could not register UdevEvents1" ); let udev_object: InterfaceRef = object_server.interface(PATH).await?; - let (shutdown_receiver, shutdown_sender) = pipe()?; + let (shutdown_sender, shutdown_receiver) = channel(1); Ok(UdevMonitor { udev_object, - shutdown_sender: Sender::from_owned_fd(shutdown_sender)?, + shutdown_sender, shutdown_receiver: Some(shutdown_receiver), }) } @@ -114,40 +115,23 @@ impl UdevDbusObject { ) -> zbus::Result<()>; } -fn run_udev(tx: &UnboundedSender, rx: &OwnedFd) -> Result<()> { +async fn run_udev(tx: UnboundedSender, mut shutdown_rx: Receiver<()>) -> Result<()> { let usb_monitor = MonitorBuilder::new()? .match_subsystem_devtype("usb", "usb_interface")? .listen()?; - let fd = usb_monitor.as_fd(); + let fd = AsyncFd::new(usb_monitor.as_fd())?; let mut iter = usb_monitor.iter(); - let ev_poller = PollFd::new(fd, PollFlags::POLLIN); - let shutdown_poller = PollFd::new(rx.as_fd(), PollFlags::POLLIN); - debug!( - "Listening on event poller {} and shutdown poller {}", - ev_poller.as_fd().as_raw_fd(), - shutdown_poller.as_fd().as_raw_fd() - ); loop { - let fds = &mut [ev_poller, shutdown_poller]; - let ret = poll(fds, PollTimeout::NONE)?; - if ret < 0 { - return Err(std::io::Error::from_raw_os_error(-ret).into()); - } - let [ev_poller, shutdown_poller] = fds; - match ev_poller.any() { - None => bail!("Event poller encountered unknown flags"), - Some(true) => { - let ev = iter - .next() - .ok_or(anyhow!("Poller said event was present, but it was not"))?; - process_usb_event(&ev, tx)?; - } - Some(false) => (), - } - match shutdown_poller.any() { - None => bail!("Shutdown poller encountered unknown flags"), - Some(true) => break Ok(()), - Some(false) => (), + select! { + guard = fd.ready(Interest::READABLE) => { + let mut guard = guard?; + for ev in iter.by_ref() { + process_usb_event(&ev, &tx)?; + }; + guard.clear_ready(); + }, + _ = shutdown_rx.recv() => break Ok(()), + _ = fd.ready(Interest::ERROR) => bail!("Event poller encountered unknown flags"), } } } @@ -155,46 +139,34 @@ fn run_udev(tx: &UnboundedSender, rx: &OwnedFd) -> Result<()> { pub(crate) fn single_poll( subsystem: &str, callback: F, - timeout: PollTimeout, -) -> AsyncJoinHandle> + timeout: Duration, +) -> Result>> where F: Fn(&Event) -> bool + Send + 'static, { - let (tx, rx) = channel(); - let subsystem = subsystem.to_string(); - let handle = spawn(move || { - let monitor = MonitorBuilder::new()? - .match_subsystem(subsystem)? - .listen()?; - let fd = monitor.as_fd(); + let monitor = MonitorBuilder::new()? + .match_subsystem(subsystem)? + .listen()?; + let handle = spawn(async move { + let fd = AsyncFd::new(monitor.as_fd())?; let mut iter = monitor.iter(); - let ev_poller = PollFd::new(fd, PollFlags::POLLIN); - let _ = tx.send(()); loop { - let fds = &mut [ev_poller]; - // TODO: Subtract the time from the last loop, if relevant - let ret = poll(fds, timeout)?; - if ret < 0 { - return Err(std::io::Error::from_raw_os_error(-ret).into()); - } - ensure!(ret == 1, "Udev poller timed out"); - let [ev_poller] = fds; - match ev_poller.any() { - None => bail!("Udev poller encountered unknown flags"), - Some(true) => { - let ev = iter - .next() - .ok_or(anyhow!("Poller said event was present, but it was not"))?; - if callback(&ev) { - return Ok(ev.syspath().to_path_buf()); - } - } - Some(false) => (), - } + select! { + _ = sleep(timeout) => bail!("Udev poller timed out"), + guard = fd.ready(Interest::READABLE) => { + let mut guard = guard?; + for ev in iter.by_ref() { + if callback(&ev) { + return Ok(ev.syspath().to_path_buf()); + } + }; + guard.clear_ready(); + }, + _ = fd.ready(Interest::ERROR) => bail!("Udev poller encountered unknown flags"), + }; } }); - let _ = rx.recv(); - handle + Ok(handle) } fn process_usb_event(ev: &Event, tx: &UnboundedSender) -> Result<()> { diff --git a/src/wifi.rs b/src/wifi.rs index cde865e..ac2594d 100644 --- a/src/wifi.rs +++ b/src/wifi.rs @@ -293,13 +293,13 @@ async fn generate_wifi_dump_inner() -> Result { link.file_name() == Some(OsStr::new("ath11k_pci")) } - let poller = single_poll("devcoredump", cb, Duration::from_secs(5).try_into()?); + let poller = single_poll("devcoredump", cb, Duration::from_secs(5)); fs::write( path("/sys/kernel/debug/ath11k/pci-0000:03:00.0/simulate_fw_crash"), "mhi-rddm\n", ) .await?; - let devcd = poller.await?; + let devcd = poller?.await??; let data = devcd.join("data"); let (mut output, path) = make_tempfile("wifi-dump-")?;