From b9524017df177703feec5781a2ce73f4627d778b Mon Sep 17 00:00:00 2001 From: Johannes Leupolz Date: Tue, 11 Nov 2025 21:56:25 +0000 Subject: [PATCH] Get rid of a fixed wait of 2 seconds, which is sometimes too long --- .../src/container/inject_in_container_job.rs | 43 +++++++++++++++--- .../container/remove_from_container_job.rs | 44 +++++++++++++++++-- vuinputd/src/main.rs | 9 ++-- vuinputd/src/requesting_process.rs | 2 +- 4 files changed, 85 insertions(+), 13 deletions(-) diff --git a/vuinputd/src/container/inject_in_container_job.rs b/vuinputd/src/container/inject_in_container_job.rs index a566c67..241c6ac 100644 --- a/vuinputd/src/container/inject_in_container_job.rs +++ b/vuinputd/src/container/inject_in_container_job.rs @@ -2,7 +2,7 @@ // // Author: Johannes Leupolz -use std::{collections::HashMap, future::Future, pin::Pin, time::Duration}; +use std::{collections::HashMap, future::Future, pin::Pin, sync::{Arc, Condvar, Mutex}, time::Duration}; use async_io::Timer; use async_pidfd::AsyncPidFd; @@ -12,6 +12,13 @@ use crate::{ container::{mknod_input_device::ensure_input_device, netlink_message::send_udev_monitor_message_with_properties, runtime_data::{self, ensure_udev_structure, read_udev_data, write_udev_data}}, jobs::job::{Job, JobTarget}, monitor_udev::EVENT_STORE, requesting_process::{run_in_net_and_mnt_namespace, RequestingProcess} }; +#[derive(Clone,Debug,Copy,PartialOrd,PartialEq)] +pub enum State { + Initialized, + Started, + Finished, +} + #[derive(Clone,Debug)] pub struct InjectInContainerJob { requesting_process: RequestingProcess, @@ -20,6 +27,7 @@ pub struct InjectInContainerJob { sys_path: String, major: u64, minor: u64, + sync_state: Arc<(Mutex,Condvar)>, } impl InjectInContainerJob { @@ -31,8 +39,30 @@ impl InjectInContainerJob { sys_path: sys_path, major: major , minor: minor, + sync_state: Arc::new((Mutex::new(State::Initialized), Condvar::new())), } } + + fn set_state(&self, new_state: &State) -> () { + let (lock, cvar) = &*self.sync_state; + let mut current_state = lock.lock().unwrap(); + *current_state = *new_state; + // We notify the condvar that the value has changed. + cvar.notify_all(); + } + + pub fn get_awaiter_for_state(&self) -> impl FnOnce(&State) -> () { + // pattern is described on https://doc.rust-lang.org/stable/std/sync/struct.Condvar.html + let sync_state = self.sync_state.clone(); + let awaiter = move | state: &State| { + let (lock, cvar) = &*sync_state; + let mut current_state = lock.lock().unwrap(); + while *state <= *current_state { + current_state = cvar.wait(current_state).unwrap(); + } + }; + awaiter + } } impl Job for InjectInContainerJob { @@ -58,6 +88,7 @@ impl InjectInContainerJob { // temporary hack that needs to be replaced. We try 50 times // Should be: Wait for the device to be created, the runtime data to be written and the // netlink message to be sent + self.set_state(&State::Started); let mut netlink_data: Option> = None; let mut runtime_data: Option = None; let mut number_of_attempt = 1; @@ -97,16 +128,17 @@ impl InjectInContainerJob { let minor=self.minor; let runtime_data = runtime_data.unwrap(); let netlink_data = netlink_data.unwrap(); + let dev_path = self.dev_path.clone(); - let child_pid = run_in_net_and_mnt_namespace(self.requesting_process, Box::new(move || { + let child_pid = run_in_net_and_mnt_namespace(&self.requesting_process, Box::new(move || { - if let Err(e) = ensure_input_device(self.dev_path.clone(), self.major, self.minor) { - debug!("Error creating input device {}: {e}",self.dev_path.clone()); + if let Err(e) = ensure_input_device(dev_path.clone(), self.major, self.minor) { + debug!("Error creating input device {}: {e}",dev_path.clone()); }; ensure_udev_structure().unwrap(); if let Err(e) = write_udev_data(runtime_data.as_str(), major, minor) { - debug!("Error writing udev data for device {}: {e}",self.dev_path.clone()); + debug!("Error writing udev data for device {}: {e}",dev_path.clone()); }; send_udev_monitor_message_with_properties(netlink_data.clone()); @@ -114,6 +146,7 @@ impl InjectInContainerJob { .expect("subprocess should work"); let pid_fd = AsyncPidFd::from_pid(child_pid.as_raw()).unwrap(); let _exit_info = pid_fd.wait().await.unwrap(); + self.set_state(&State::Finished); } } diff --git a/vuinputd/src/container/remove_from_container_job.rs b/vuinputd/src/container/remove_from_container_job.rs index 88098ae..0c61b25 100644 --- a/vuinputd/src/container/remove_from_container_job.rs +++ b/vuinputd/src/container/remove_from_container_job.rs @@ -2,7 +2,7 @@ // // Author: Johannes Leupolz -use std::{collections::HashMap, future::Future, pin::Pin, time::Duration}; +use std::{collections::HashMap, future::Future, pin::Pin, sync::{Arc, Condvar, Mutex}, time::Duration}; use async_io::Timer; use async_pidfd::AsyncPidFd; @@ -12,6 +12,15 @@ use crate::{ container::{mknod_input_device::{ensure_input_device, remove_input_device}, netlink_message::send_udev_monitor_message_with_properties, runtime_data::{self, delete_udev_data, ensure_udev_structure, read_udev_data, write_udev_data}}, jobs::job::{Job, JobTarget}, monitor_udev::EVENT_STORE, requesting_process::{RequestingProcess, run_in_net_and_mnt_namespace} }; + +#[derive(Clone,Debug,Copy,PartialOrd,PartialEq)] +pub enum State { + Initialized, + Started, + Finished, +} + + #[derive(Clone,Debug)] pub struct RemoveFromContainerJob { requesting_process: RequestingProcess, @@ -20,6 +29,7 @@ pub struct RemoveFromContainerJob { sys_path: String, major: u64, minor: u64, + sync_state: Arc<(Mutex,Condvar)>, } impl RemoveFromContainerJob { @@ -31,8 +41,30 @@ impl RemoveFromContainerJob { sys_path: sys_path, major: major , minor: minor, + sync_state: Arc::new((Mutex::new(State::Initialized), Condvar::new())), } } + fn set_state(&self, new_state: &State) -> () { + let (lock, cvar) = &*self.sync_state; + let mut current_state = lock.lock().unwrap(); + *current_state = *new_state; + // We notify the condvar that the value has changed. + cvar.notify_all(); + } + + pub fn get_awaiter_for_state(&self) -> impl FnOnce(&State) -> () { + // pattern is described on https://doc.rust-lang.org/stable/std/sync/struct.Condvar.html + let sync_state = self.sync_state.clone(); + let awaiter = move | state: &State| { + let (lock, cvar) = &*sync_state; + let mut current_state = lock.lock().unwrap(); + while *state <= *current_state { + current_state = cvar.wait(current_state).unwrap(); + } + }; + awaiter + } + } impl Job for RemoveFromContainerJob { @@ -55,6 +87,8 @@ impl Job for RemoveFromContainerJob { impl RemoveFromContainerJob { async fn remove_from_container(self) { + self.set_state(&State::Started); + let netlink_event = match EVENT_STORE.get().unwrap().lock().unwrap().take(&self.sys_path) { Some(netlink_event) => netlink_event, None => { @@ -73,9 +107,10 @@ impl RemoveFromContainerJob { let mut netlink_data = netlink_data.unwrap().clone(); let major = self.major; let minor=self.minor; + let dev_path = self.dev_path.clone(); let _ = netlink_data.insert("ACTION".to_string(),"remove".to_string()); - let child_pid = run_in_net_and_mnt_namespace(self.requesting_process, Box::new(move || { + let child_pid = run_in_net_and_mnt_namespace(&self.requesting_process, Box::new(move || { // TODO: we should keep the same order as event_execute_rules_on_remove in // https://github.com/systemd/systemd/blob/main/src/udev/udev-event.c @@ -83,14 +118,15 @@ impl RemoveFromContainerJob { if let Err(e) = delete_udev_data(major,minor) { debug!("Error deleting udev data for {}:{}: {e}",major,minor); } - if let Err(e) = remove_input_device(self.dev_path.clone(), self.major, self.minor) { - debug!("Error removing input device {}: {e}",self.dev_path.clone()); + if let Err(e) = remove_input_device(dev_path.clone(), self.major, self.minor) { + debug!("Error removing input device {}: {e}",dev_path.clone()); }; })) .expect("subprocess should work"); let pid_fd = AsyncPidFd::from_pid(child_pid.as_raw()).unwrap(); let _exit_info = pid_fd.wait().await.unwrap(); + self.set_state(&State::Finished); } } diff --git a/vuinputd/src/main.rs b/vuinputd/src/main.rs index 77254c1..e5089ae 100644 --- a/vuinputd/src/main.rs +++ b/vuinputd/src/main.rs @@ -348,7 +348,9 @@ unsafe extern "C" fn vuinput_release( if input_device.is_some() && ! VUINPUTD_NAMESPACES.get().unwrap().equal_mnt_and_net(&vuinput_state.requesting_process.namespaces) { let input_device = input_device.unwrap(); let remove_job=RemoveFromContainerJob::new(vuinput_state.requesting_process.clone(),input_device.devnode.clone(),input_device.syspath.clone(),input_device.major,input_device.minor); + let awaiter = remove_job.get_awaiter_for_state(); JOB_DISPATCHER.get().unwrap().lock().unwrap().dispatch(Box::new(remove_job)); + awaiter(&container::remove_from_container_job::State::Finished); } drop(vuinput_state); @@ -515,16 +517,15 @@ unsafe extern "C" fn vuinput_ioctl( // Create device in container, if the request was really from another namespace if ! VUINPUTD_NAMESPACES.get().unwrap().equal_mnt_and_net(&vuinput_state.requesting_process.namespaces) { let inject_job=InjectInContainerJob::new(vuinput_state.requesting_process.clone(),devnode.clone(),sysname.clone(),major,minor); + let awaiter = inject_job.get_awaiter_for_state(); JOB_DISPATCHER.get().unwrap().lock().unwrap().dispatch(Box::new(inject_job)); + awaiter(&container::inject_in_container_job::State::Finished); } // write a SYN-event (which is just zeros) just for validation let syn_event : [u8; 24] = [0; 24]; vuinput_state.file.write_all(&syn_event).unwrap(); - // hard code 2 second sleep - std::thread::sleep(std::time::Duration::from_secs(2)); - fuse_lowlevel::fuse_reply_ioctl(_req, 0, std::ptr::null(), 0); } UI_DEV_DESTROY => { @@ -535,7 +536,9 @@ unsafe extern "C" fn vuinput_ioctl( if input_device.is_some() && ! VUINPUTD_NAMESPACES.get().unwrap().equal_mnt_and_net(&vuinput_state.requesting_process.namespaces) { let input_device = input_device.unwrap(); let remove_job=RemoveFromContainerJob::new(vuinput_state.requesting_process.clone(),input_device.devnode.clone(),input_device.syspath.clone(),input_device.major,input_device.minor); + let awaiter = remove_job.get_awaiter_for_state(); JOB_DISPATCHER.get().unwrap().lock().unwrap().dispatch(Box::new(remove_job)); + awaiter(&container::remove_from_container_job::State::Finished) } ui_dev_destroy(fd).unwrap(); diff --git a/vuinputd/src/requesting_process.rs b/vuinputd/src/requesting_process.rs index 546199a..5f691f2 100644 --- a/vuinputd/src/requesting_process.rs +++ b/vuinputd/src/requesting_process.rs @@ -243,7 +243,7 @@ pub fn get_requesting_process(pid: Pid) -> RequestingProcess { /// Runs a function inside the given network and mount namespaces. /// Returns the child PID so the caller can `waitpid` on it. -pub fn run_in_net_and_mnt_namespace(ns: RequestingProcess, func: Box) -> nix::Result { +pub fn run_in_net_and_mnt_namespace(ns: &RequestingProcess, func: Box) -> nix::Result { //Note: The child process is created with a single thread—the one that called fork(). match unsafe { fork()? } {