Purge caches and build directories when nothing is running

This prevents deleting files that Cargo is trying to read out from under
it.

Author: Mark-Simulacrum

src/runner/worker.rs

@@ -11,11 +11,8 @@ use crate::utils;
 use rustwide::logging::{self, LogStorage};
 use rustwide::{BuildDirectory, Workspace};
 use std::collections::HashMap;
-use std::sync::Condvar;
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Mutex,
-};
+use std::sync::Mutex;
+use std::sync::{Arc, Condvar, OnceLock};
 use std::time::Duration;
 
 pub trait RecordProgress: Send + Sync {
@@ -51,8 +48,10 @@ pub(super) struct Worker<'a> {
     ex: &'a Experiment,
     config: &'a crate::config::Config,
     api: &'a dyn RecordProgress,
-    target_dir_cleanup: AtomicBool,
     next_crate: &'a (dyn Fn() -> Fallible<Option<Crate>> + Send + Sync),
+
+    // Called by the worker thread between crates, when no global state (namely caches) is in use.
+    between_crates: OnceLock<Box<dyn Fn() + Send + Sync + 'a>>,
 }
 
 impl<'a> Worker<'a> {
@@ -81,7 +80,8 @@ impl<'a> Worker<'a> {
             config,
             next_crate,
             api,
-            target_dir_cleanup: AtomicBool::new(false),
+
+            between_crates: OnceLock::new(),
         }
     }
 
@@ -167,10 +167,12 @@ impl<'a> Worker<'a> {
                 return Ok(());
             };
 
-            self.maybe_cleanup_target_dir()?;
-
             info!("{} processing crate {}", self.name, krate);
 
+            if let Some(cb) = self.between_crates.get() {
+                cb();
+            }
+
             if !self.ex.ignore_blacklist && self.config.should_skip(&krate) {
                 for tc in &self.ex.toolchains {
                     // If a skipped crate is somehow sent to the agent (for example, when a crate was
@@ -338,41 +340,57 @@ impl<'a> Worker<'a> {
             }
         }
     }
-
-    fn maybe_cleanup_target_dir(&self) -> Fallible<()> {
-        if !self.target_dir_cleanup.swap(false, Ordering::SeqCst) {
-            return Ok(());
-        }
-        info!("purging target dir for {}", self.name);
-        for dir in self.build_dir.values() {
-            dir.lock().unwrap().purge()?;
-        }
-
-        Ok(())
-    }
-
-    fn schedule_target_dir_cleanup(&self) {
-        self.target_dir_cleanup.store(true, Ordering::SeqCst);
-    }
 }
 
-pub(super) struct DiskSpaceWatcher<'a> {
+pub(super) struct DiskSpaceWatcher {
     interval: Duration,
     threshold: f32,
-    workers: &'a [Worker<'a>],
     should_stop: Mutex<bool>,
     waiter: Condvar,
+
+    // If the bool is true, that means we're waiting for the cache to reach zero, in which case
+    // workers will wait for it to be false before starting. This gives us a global 'is the cache
+    // in use' synchronization point.
+    cache_in_use: Mutex<(usize, bool)>,
+    cache_waiter: Condvar,
 }
 
-impl<'a> DiskSpaceWatcher<'a> {
-    pub(super) fn new(interval: Duration, threshold: f32, workers: &'a [Worker<'a>]) -> Self {
-        DiskSpaceWatcher {
+impl DiskSpaceWatcher {
+    pub(super) fn new(interval: Duration, threshold: f32, workers: &[Worker<'_>]) -> Arc<Self> {
+        let this = Arc::new(DiskSpaceWatcher {
             interval,
             threshold,
-            workers,
             should_stop: Mutex::new(false),
             waiter: Condvar::new(),
+
+            cache_in_use: Mutex::new((workers.len(), false)),
+            cache_waiter: Condvar::new(),
+        });
+
+        let worker_count = workers.len();
+        for worker in workers {
+            let this = Arc::clone(&this);
+            assert!(worker
+                .between_crates
+                .set(Box::new(move || {
+                    let mut guard = this.cache_in_use.lock().unwrap();
+                    // note that we're not running right now.
+                    guard.0 = guard.0.checked_sub(1).unwrap();
+                    // Wait until the cache doesn't need to be purged...
+                    if guard.1 {
+                        info!("waiting for cache to maybe be purged");
+                    }
+                    this.cache_waiter.notify_all();
+                    let mut guard = this.cache_waiter.wait_while(guard, |c| c.1).unwrap();
+                    // Then set ourselves as running.
+                    guard.0 += 1;
+                    this.cache_waiter.notify_all();
+                    assert!(guard.0 <= worker_count);
+                }))
+                .is_ok());
         }
+
+        this
     }
 
     pub(super) fn stop(&self) {
@@ -406,14 +424,30 @@ impl<'a> DiskSpaceWatcher<'a> {
         };
 
         if usage.is_threshold_reached(self.threshold) {
-            warn!("running the scheduled thread cleanup");
-            for worker in self.workers {
-                worker.schedule_target_dir_cleanup();
-            }
+            warn!("waiting for workers to finish");
+
+            // Set interest in cleaning caches and then wait for cache use to drain to zero.
+            let mut guard = self.cache_in_use.lock().unwrap();
+            guard.1 = true;
+
+            self.cache_waiter.notify_all();
+
+            let mut guard = self.cache_waiter.wait_while(guard, |c| c.0 > 0).unwrap();
+
+            // OK, purging caches, clear interest.
+            guard.1 = false;
+
+            self.cache_waiter.notify_all();
+
+            warn!("purging all build dirs and caches");
 
             if let Err(e) = workspace.purge_all_caches() {
                 warn!("failed to purge caches: {:?}", e);
             }
+
+            if let Err(e) = workspace.purge_all_build_dirs() {
+                warn!("failed to purge build directories: {:?}", e);
+            }
         }
     }
 }