From 320a77ef63fe4793f037d8122fdabd1480747f5c Mon Sep 17 00:00:00 2001 From: Johannes Leupolz Date: Mon, 8 Dec 2025 23:03:34 +0000 Subject: [PATCH] Files renamed for more clarity --- vuinputd/src/job_engine/job.rs | 2 +- vuinputd/src/job_engine/tests.rs | 310 ++++++++++++++++++ vuinputd/src/jobs/inject_in_container_job.rs | 2 +- .../src/jobs/remove_from_container_job.rs | 2 +- vuinputd/src/main.rs | 6 +- .../mod.rs} | 0 6 files changed, 317 insertions(+), 5 deletions(-) rename vuinputd/src/{requesting_process.rs => process_tools/mod.rs} (100%) diff --git a/vuinputd/src/job_engine/job.rs b/vuinputd/src/job_engine/job.rs index 051bce8..d03bc27 100644 --- a/vuinputd/src/job_engine/job.rs +++ b/vuinputd/src/job_engine/job.rs @@ -11,7 +11,7 @@ use std::sync::Mutex; use std::thread::{self, JoinHandle}; use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc}; -use crate::requesting_process::RequestingProcess; +use crate::process_tools::RequestingProcess; // To discuss: // what we handle here, could also be named Task. The decision for job was more or less diff --git a/vuinputd/src/job_engine/tests.rs b/vuinputd/src/job_engine/tests.rs index db87e02..77276db 100644 --- a/vuinputd/src/job_engine/tests.rs +++ b/vuinputd/src/job_engine/tests.rs @@ -56,3 +56,313 @@ fn test_job_ordering() { assert_eq!(*c.lock().unwrap(), 6); } + +/* + +// +// 2. Cancellation test (default: do NOT run cancelled job) +// +#[test] +fn test_cancellation_stops_jobs() { + let (mut dispatcher, mut pool) = test_dispatcher(); + let c = shared_counter(); + + // First job runs + let c1 = c.clone(); + dispatcher.queue(Job::new(move |_ctx| { + *c1.borrow_mut() += 1; + Poll::Ready(()) + })); + + // Second job should NOT run (will be cancelled) + let c2 = c.clone(); + dispatcher.queue(Job::with_flags( + move |_ctx| { + *c2.borrow_mut() += 1; + Poll::Ready(()) + }, + JobFlags::default(), // execute_after_cancellation = false + )); + + // Immediately close before job #2 runs + dispatcher.close(); + run_pool_to_completion(&mut pool); + + assert_eq!(*c.borrow(), 1); +} + +// +// 3. Cancellation *with* execute_after_cancellation +// +#[test] +fn test_cancellation_runs_cleanup_jobs() { + let (mut dispatcher, mut pool) = test_dispatcher(); + let c = shared_counter(); + + // First job runs + let c1 = c.clone(); + dispatcher.queue(Job::new(move |_ctx| { + *c1.borrow_mut() += 1; + Poll::Ready(()) + })); + + // Cleanup job must run even after cancellation + let c2 = c.clone(); + dispatcher.queue(Job::with_flags( + move |_ctx| { + *c2.borrow_mut() += 10; + Poll::Ready(()) + }, + JobFlags { + execute_after_cancellation: true, + }, + )); + + dispatcher.close(); + run_pool_to_completion(&mut pool); + + assert_eq!(*c.borrow(), 11); +} + +// +// 4. SIGTERM-like shutdown behaviour +// +#[test] +fn test_shutdown_wait_until_finished() { + let (mut dispatcher, mut pool) = test_dispatcher(); + let c = shared_counter(); + + // Simulate some work + let c1 = c.clone(); + dispatcher.queue(Job::new(move |_ctx| { + *c1.borrow_mut() += 1; + Poll::Ready(()) + })); + + dispatcher.close(); + run_pool_to_completion(&mut pool); + dispatcher.wait_until_finished(); + + assert_eq!(*c.borrow(), 1); +} + +// +// 5. Failure propagation +// +#[test] +fn test_job_failure_does_not_crash_dispatcher() { + let (mut dispatcher, mut pool) = test_dispatcher(); + let c = shared_counter(); + + dispatcher.queue(Job::new(|_ctx| { + panic!("intentional test panic"); + #[allow(unreachable_code)] + Poll::Ready(()) + })); + + // Following job should still run if the system recovers + let c2 = c.clone(); + dispatcher.queue(Job::new(move |_ctx| { + *c2.borrow_mut() += 1; + Poll::Ready(()) + })); + + dispatcher.close(); + + // Some designs need catch_unwind here; yours might not. + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run_pool_to_completion(&mut pool); + })); + + assert_eq!(*c.borrow(), 1); +} + +// +// 6. Stress: many short jobs +// +#[test] +fn test_many_small_jobs() { + let (mut dispatcher, mut pool) = test_dispatcher(); + let count = shared_counter(); + + for _ in 0..200 { + let c = count.clone(); + dispatcher.queue(Job::new(move |_ctx| { + *c.borrow_mut() += 1; + Poll::Ready(()) + })); + } + + dispatcher.close(); + run_pool_to_completion(&mut pool); + + assert_eq!(*count.borrow(), 200); +} + +// +// 7. Ensure no jobs run after final close +// +#[test] +fn test_no_jobs_after_close() { + let (mut dispatcher, mut pool) = test_dispatcher(); + dispatcher.close(); + + let executed = shared_counter(); + let e = executed.clone(); + + dispatcher.queue(Job::new(move |_ctx| { + *e.borrow_mut() += 1; + Poll::Ready(()) + })); + + // Dispatcher should ignore queued jobs after close() + run_pool_to_completion(&mut pool); + + assert_eq!(*executed.borrow(), 0); +} +#[test] +fn test_stress_light_multi_target() { + use std::sync::{Arc, Mutex}; + use crate::jobs::{Dispatcher, JobTarget}; + use crate::jobs::closure_job::ClosureJob; + + // Shared result buffer + let results = Arc::new(Mutex::new(Vec::new())); + + let mut dispatcher = Dispatcher::new(); + + // Create a few job targets + let target_a = JobTarget::Container("A".into()); + let target_b = JobTarget::Container("B".into()); + let host = JobTarget::Host; + + // Create 10 jobs per target + for i in 0..10 { + let results_a = results.clone(); + dispatcher.dispatch(Box::new(ClosureJob::new( + format!("A-{i}"), + target_a.clone(), + false, + Box::new(move |_t| { + Box::pin(async move { + results_a.lock().unwrap().push(format!("A-{i}")); + }) + }), + ))); + + let results_b = results.clone(); + dispatcher.dispatch(Box::new(ClosureJob::new( + format!("B-{i}"), + target_b.clone(), + false, + Box::new(move |_t| { + Box::pin(async move { + results_b.lock().unwrap().push(format!("B-{i}")); + }) + }), + ))); + + let results_h = results.clone(); + dispatcher.dispatch(Box::new(ClosureJob::new( + format!("H-{i}"), + host.clone(), + false, + Box::new(move |_t| { + Box::pin(async move { + results_h.lock().unwrap().push(format!("H-{i}")); + }) + }), + ))); + } + + dispatcher.close(); + dispatcher.wait_until_finished(); + + let buf = results.lock().unwrap(); + + // Each target must preserve *its* order: + assert_eq!( + buf.iter().filter(|s| s.starts_with("A-")).cloned().collect::>(), + (0..10).map(|i| format!("A-{i}")).collect::>() + ); + + assert_eq!( + buf.iter().filter(|s| s.starts_with("B-")).cloned().collect::>(), + (0..10).map(|i| format!("B-{i}")).collect::>() + ); + + assert_eq!( + buf.iter().filter(|s| s.starts_with("H-")).cloned().collect::>(), + (0..10).map(|i| format!("H-{i}")).collect::>() + ); + + // But between targets, the order may be interleaved (expected) + // So we don't assert cross-order. +} + +#[test] +fn test_cleanup_when_target_disappears() { + use std::sync::{Arc, Mutex}; + use crate::jobs::{Dispatcher, JobTarget}; + use crate::jobs::closure_job::ClosureJob; + + let results = Arc::new(Mutex::new(Vec::new())); + let mut dispatcher = Dispatcher::new(); + + let target_dead = JobTarget::Container("dead".into()); + let target_alive = JobTarget::Container("alive".into()); + + // Two jobs for dead target: one normal, one allowed after cancel + { + let r = results.clone(); + dispatcher.dispatch(Box::new(ClosureJob::new( + "dead-normal", + target_dead.clone(), + false, + Box::new(move |_| Box::pin(async move { + r.lock().unwrap().push("dead-normal".into()); + })) + ))); + + let r = results.clone(); + dispatcher.dispatch(Box::new(ClosureJob::new( + "dead-cleanup", + target_dead.clone(), + true, // allowed after cancellation + Box::new(move |_| Box::pin(async move { + r.lock().unwrap().push("dead-cleanup".into()); + })) + ))); + } + + // One job for a live target + { + let r = results.clone(); + dispatcher.dispatch(Box::new(ClosureJob::new( + "alive-job", + target_alive.clone(), + false, + Box::new(move |_| Box::pin(async move { + r.lock().unwrap().push("alive-job".into()); + })) + ))); + } + + // Simulate container removal + dispatcher.target_gone(&target_dead); + + dispatcher.close(); + dispatcher.wait_until_finished(); + + let buf = results.lock().unwrap(); + + // Should NOT run: + assert!(!buf.contains(&"dead-normal".into())); + + // Should run because execute_after_cancellation = true + assert!(buf.contains(&"dead-cleanup".into())); + + // Should run normally + assert!(buf.contains(&"alive-job".into())); +} + */ \ No newline at end of file diff --git a/vuinputd/src/jobs/inject_in_container_job.rs b/vuinputd/src/jobs/inject_in_container_job.rs index 19abaea..eca8fb2 100644 --- a/vuinputd/src/jobs/inject_in_container_job.rs +++ b/vuinputd/src/jobs/inject_in_container_job.rs @@ -7,7 +7,7 @@ use std::{collections::HashMap, future::Future, pin::Pin, sync::{Arc, Condvar, M use async_io::Timer; use log::debug; -use crate::{job_engine::job::{Job, JobTarget}, jobs::{mknod_input_device::ensure_input_device, monitor_udev_job::EVENT_STORE, netlink_message::send_udev_monitor_message_with_properties, runtime_data::{ensure_udev_structure, read_udev_data, write_udev_data}}, requesting_process::{Pid, RequestingProcess, await_process, run_in_net_and_mnt_namespace}}; +use crate::{job_engine::job::{Job, JobTarget}, jobs::{mknod_input_device::ensure_input_device, monitor_udev_job::EVENT_STORE, netlink_message::send_udev_monitor_message_with_properties, runtime_data::{ensure_udev_structure, read_udev_data, write_udev_data}}, process_tools::{Pid, RequestingProcess, await_process, run_in_net_and_mnt_namespace}}; #[derive(Clone,Debug,Copy,PartialOrd,PartialEq)] diff --git a/vuinputd/src/jobs/remove_from_container_job.rs b/vuinputd/src/jobs/remove_from_container_job.rs index a08bd81..1de6865 100644 --- a/vuinputd/src/jobs/remove_from_container_job.rs +++ b/vuinputd/src/jobs/remove_from_container_job.rs @@ -7,7 +7,7 @@ use std::{collections::HashMap, future::Future, pin::Pin, sync::{Arc, Condvar, M use async_io::Timer; use log::debug; -use crate::{job_engine::job::{Job, JobTarget}, jobs::{mknod_input_device::remove_input_device, monitor_udev_job::EVENT_STORE, netlink_message::send_udev_monitor_message_with_properties, runtime_data::{delete_udev_data, ensure_udev_structure, read_udev_data, write_udev_data}}, requesting_process::{Pid, RequestingProcess, await_process, run_in_net_and_mnt_namespace}}; +use crate::{job_engine::job::{Job, JobTarget}, jobs::{mknod_input_device::remove_input_device, monitor_udev_job::EVENT_STORE, netlink_message::send_udev_monitor_message_with_properties, runtime_data::{delete_udev_data, ensure_udev_structure, read_udev_data, write_udev_data}}, process_tools::{Pid, RequestingProcess, await_process, run_in_net_and_mnt_namespace}}; diff --git a/vuinputd/src/main.rs b/vuinputd/src/main.rs index 59f1ef8..a67067e 100644 --- a/vuinputd/src/main.rs +++ b/vuinputd/src/main.rs @@ -37,16 +37,18 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex, OnceLock, RwLock}; use uinput_ioctls::*; -pub mod requesting_process; pub mod compat; + use crate::compat::input_event_compat; use crate::jobs::inject_in_container_job::InjectInContainerJob; use crate::jobs::monitor_udev_job::MonitorBackgroundLoop; use crate::jobs::remove_from_container_job::RemoveFromContainerJob; -use crate::requesting_process::*; + +pub mod process_tools; pub mod job_engine; use crate::job_engine::job::*; +use crate::process_tools::*; pub mod jobs; diff --git a/vuinputd/src/requesting_process.rs b/vuinputd/src/process_tools/mod.rs similarity index 100% rename from vuinputd/src/requesting_process.rs rename to vuinputd/src/process_tools/mod.rs