Auto merge of #567 - Mark-Simulacrum:simplify-runner, r=pietroalbini

Skip mpsc channels in report generation and disk cleaning

I doubt this is the cause of our deadlocks, but std's mpsc channels have had some bugs in the past around timeouts IIRC. In any case, the new implementation feels just simpler to me, so an improvement regardless.

Author: bors

src/lib.rs

@@ -8,7 +8,8 @@
     clippy::unnecessary_wraps,
     clippy::needless_question_mark,
     clippy::vec_init_then_push,
-    clippy::upper_case_acronyms
+    clippy::upper_case_acronyms,
+    clippy::mutex_atomic
 )]
 
 pub mod actions;

src/runner/worker.rs

@@ -7,10 +7,10 @@ use crate::runner::{OverrideResult, RunnerState};
 use crate::utils;
 use rustwide::{BuildDirectory, Workspace};
 use std::collections::HashMap;
+use std::sync::Condvar;
 use std::sync::{
     atomic::{AtomicBool, Ordering},
-    mpsc::{self, RecvTimeoutError},
-    Arc, Mutex,
+    Mutex,
 };
 use std::thread;
 use std::time::Duration;
@@ -151,34 +151,38 @@ pub(super) struct DiskSpaceWatcher<'a, DB: WriteResults + Sync> {
     interval: Duration,
     threshold: f32,
     workers: &'a [Worker<'a, DB>],
-    stop_send: Arc<Mutex<mpsc::Sender<()>>>,
-    stop_recv: Arc<Mutex<mpsc::Receiver<()>>>,
+    should_stop: Mutex<bool>,
+    waiter: Condvar,
 }
 
 impl<'a, DB: WriteResults + Sync> DiskSpaceWatcher<'a, DB> {
     pub(super) fn new(interval: Duration, threshold: f32, workers: &'a [Worker<'a, DB>]) -> Self {
-        let (stop_send, stop_recv) = mpsc::channel();
         DiskSpaceWatcher {
             interval,
             threshold,
             workers,
-            stop_send: Arc::new(Mutex::new(stop_send)),
-            stop_recv: Arc::new(Mutex::new(stop_recv)),
+            should_stop: Mutex::new(false),
+            waiter: Condvar::new(),
         }
     }
 
     pub(super) fn stop(&self) {
-        self.stop_send.lock().unwrap().send(()).unwrap();
+        *self.should_stop.lock().unwrap() = true;
+        self.waiter.notify_all();
     }
 
     pub(super) fn run(&self) {
-        loop {
+        let mut should_stop = self.should_stop.lock().unwrap();
+        while !*should_stop {
             self.check();
-            match self.stop_recv.lock().unwrap().recv_timeout(self.interval) {
-                Ok(()) => return,
-                Err(RecvTimeoutError::Timeout) => {}
-                Err(RecvTimeoutError::Disconnected) => panic!("disconnected stop channel"),
-            }
+            // Wait for either the interval to pass or should_stop to get a
+            // write. We don't care if we timed out or not, we can double check
+            // the should_stop regardless.
+            should_stop = self
+                .waiter
+                .wait_timeout(should_stop, self.interval)
+                .unwrap()
+                .0;
         }
     }
 

src/server/reports.rs

@@ -7,8 +7,8 @@ use crate::server::Data;
 use crate::utils;
 use rusoto_core::request::HttpClient;
 use rusoto_s3::S3Client;
-use std::sync::{mpsc, Arc, Mutex};
-use std::thread;
+use std::sync::{Arc, Mutex};
+use std::thread::{self, Thread};
 use std::time::Duration;
 
 // Automatically wake up the reports generator thread every 10 minutes to check for new jobs
@@ -32,7 +32,7 @@ fn generate_report(data: &Data, ex: &Experiment, results: &DatabaseDB) -> Fallib
     Ok(res)
 }
 
-fn reports_thread(data: &Data, wakes: &mpsc::Receiver<()>) -> Fallible<()> {
+fn reports_thread(data: &Data) -> Fallible<()> {
     let timeout = Duration::from_secs(AUTOMATIC_THREAD_WAKEUP);
     let results = DatabaseDB::new(&data.db);
 
@@ -41,9 +41,7 @@ fn reports_thread(data: &Data, wakes: &mpsc::Receiver<()>) -> Fallible<()> {
             Some(ex) => ex,
             None => {
                 // This will sleep AUTOMATIC_THREAD_WAKEUP seconds *or* until a wake is received
-                if let Err(mpsc::RecvTimeoutError::Disconnected) = wakes.recv_timeout(timeout) {
-                    thread::sleep(timeout);
-                }
+                std::thread::park_timeout(timeout);
 
                 continue;
             }
@@ -128,43 +126,28 @@ fn reports_thread(data: &Data, wakes: &mpsc::Receiver<()>) -> Fallible<()> {
 }
 
 #[derive(Clone, Default)]
-pub struct ReportsWorker(Arc<Mutex<Option<mpsc::Sender<()>>>>);
+pub struct ReportsWorker(Arc<Mutex<Option<Thread>>>);
 
 impl ReportsWorker {
     pub fn new() -> Self {
         ReportsWorker(Arc::new(Mutex::new(None)))
     }
 
     pub fn spawn(&self, data: Data) {
-        let waker = self.0.clone();
-        thread::spawn(move || {
-            // Set up a new waker channel
-            let (wake_send, wake_recv) = mpsc::channel();
-            {
-                let mut waker = waker.lock().unwrap();
-                *waker = Some(wake_send);
-            }
-
-            loop {
-                let result = reports_thread(&data.clone(), &wake_recv)
-                    .with_context(|_| "the reports generator thread crashed");
-                if let Err(e) = result {
-                    utils::report_failure(&e);
-                }
-
-                warn!("the reports generator thread will be respawned in one minute");
-                thread::sleep(Duration::from_secs(60));
+        let joiner = thread::spawn(move || loop {
+            let result = reports_thread(&data.clone())
+                .with_context(|_| "the reports generator thread crashed");
+            if let Err(e) = result {
+                utils::report_failure(&e);
             }
         });
+        *self.0.lock().unwrap_or_else(|e| e.into_inner()) = Some(joiner.thread().clone());
     }
 
     pub fn wake(&self) {
-        // We don't really care if the wake fails: the reports generator thread wakes up on its own
-        // every few minutes, so this just speeds up the process
-        if let Some(waker) = self.0.lock().ok().as_ref().and_then(|opt| opt.as_ref()) {
-            if waker.send(()).is_err() {
-                warn!("can't wake the reports generator, will have to wait");
-            }
+        let guard = self.0.lock().unwrap_or_else(|e| e.into_inner());
+        if let Some(thread) = &*guard {
+            thread.unpark();
         } else {
             warn!("no report generator to wake up!");
         }