Allow for a job monitor to be reused.

src/activate/services.rs

@@ -56,7 +56,7 @@ pub fn activate(store_path: &StorePath, ephemeral: bool) -> Result<()> {
     // still knows about these units.
     wait_for_jobs(
         &service_manager,
-        job_monitor,
+        &job_monitor,
         stop_services(&service_manager, &services_to_stop)?,
         &timeout,
     )?;
@@ -69,10 +69,9 @@ pub fn activate(store_path: &StorePath, ephemeral: bool) -> Result<()> {
     let active_targets = get_active_targets(&service_manager);
     let services_to_reload = get_services_to_reload(services, old_services);
 
-    let job_monitor = service_manager.monitor_jobs_init()?;
     wait_for_jobs(
         &service_manager,
-        job_monitor,
+        &job_monitor,
         reload_services(&service_manager, &services_to_reload)?
             + start_units(&service_manager, &active_targets?)?,
         &timeout,
@@ -192,7 +191,7 @@ pub fn deactivate() -> Result<()> {
         // still knows about these units.
         wait_for_jobs(
             &service_manager,
-            job_monitor,
+            &job_monitor,
             stop_services(&service_manager, &old_services)?,
             &timeout,
         )?;
@@ -337,7 +336,7 @@ where
 
 fn wait_for_jobs(
     service_manager: &systemd::ServiceManager,
-    job_monitor: systemd::JobMonitor,
+    job_monitor: &systemd::JobMonitor,
     jobs: HashSet<JobId>,
     timeout: &Option<Duration>,
 ) -> Result<()> {

src/systemd.rs

@@ -1,4 +1,4 @@
-// FIXME: Remove this
+// TODO: Remove this
 #![allow(dead_code)]
 
 mod manager;
@@ -75,9 +75,23 @@ pub struct Job<'a> {
     path: Path<'a>,
 }
 
-pub struct JobMonitor {
+pub struct JobMonitor<'a> {
     job_names: Arc<Mutex<im::HashSet<String>>>,
     tokens: im::HashSet<Token>,
+    service_manager: &'a ServiceManager,
+}
+
+impl Drop for JobMonitor<'_> {
+    fn drop(&mut self) {
+        self.tokens.iter().for_each(|t| {
+            self.service_manager
+                .proxy
+                .match_stop(*t, true)
+                .unwrap_or_else(|e|
+                    log::error!("Error while stopping match listener, memory might leak...\n  Caused by: {e}")
+                )
+        });
+    }
 }
 
 impl Drop for ServiceManager {
@@ -147,7 +161,7 @@ impl ServiceManager {
     }
 
     pub fn monitor_jobs_init(&self) -> Result<JobMonitor, Error> {
-        let job_names: Arc<Mutex<im::HashSet<String>>> = Arc::new(Mutex::from(im::HashSet::new()));
+        let job_names = Arc::new(Mutex::from(im::HashSet::<String>::new()));
 
         let job_names_clone = Arc::clone(&job_names);
         let token = self.proxy.match_signal(
@@ -165,16 +179,15 @@ impl ServiceManager {
         Ok(JobMonitor {
             job_names: Arc::clone(&job_names),
             tokens: im::HashSet::unit(token),
+            service_manager: self,
         })
     }
 
     /// Waits for the monitored jobs to finish. Returns `true` if all jobs
     /// finished before the timeout, `false` otherwise.
-    /// It is important that we consume the job monitor, since we remove the signal handler at the
-    /// end of this call, and so the job monitor cannot be re-used.
     pub fn monitor_jobs_finish<I>(
         &self,
-        job_monitor: JobMonitor,
+        job_monitor: &JobMonitor,
         timeout: &Option<Duration>,
         services: I,
     ) -> Result<bool, Error>
@@ -217,11 +230,6 @@ impl ServiceManager {
         if total_jobs > 0 {
             log::info!("All jobs finished.");
         }
-        // Remove the signal handling callback
-        job_monitor
-            .tokens
-            .into_iter()
-            .try_for_each(|t| self.proxy.match_stop(t, true))?;
         Ok(true)
     }