fix(#18): have JobStore.next_job takes a timeout and block

src/job_store.rs

@@ -35,15 +35,31 @@ impl JobStore {
     }
   }
 
-  pub fn next_job(&self) -> Option<Arc<Job>> {
-    let mut guard = self.data.lock().unwrap();
-    guard.pop_first().map(|wrapper| {
-      let (job, next) = wrapper.compute_next();
-      if let Some(next) = next {
-        guard.insert(next);
-      }
-      job
-    })
+  pub fn next_job(&self, timeout: Duration) -> Option<Arc<Job>> {
+    let guard = self.data.lock().unwrap();
+    let timeout = guard
+      .first()
+      .map(|wrapper| {
+        wrapper
+          .next_fire()
+          .duration_since(SystemTime::now())
+          .unwrap_or_default()
+      })
+      .unwrap_or(timeout)
+      .min(timeout);
+
+    let (mut guard, _) = self.signal.wait_timeout(guard, timeout).unwrap();
+    if !guard.is_empty() && guard.first().unwrap().next_fire() <= SystemTime::now() + Duration::from_micros(10) {
+      guard.pop_first().map(|wrapper| {
+        let (job, next) = wrapper.compute_next();
+        if let Some(next) = next {
+          guard.insert(next);
+        }
+        job
+      })
+    } else {
+      None
+    }
   }
 
   pub fn add(&self, job: Job, trigger: Trigger) {
@@ -52,12 +68,8 @@ impl JobStore {
     self.signal.notify_one()
   }
 
-  pub fn next_fire(&self) -> Option<Duration> {
-    self.data.lock().unwrap().first().map(|j| {
-      j.next_fire()
-        .duration_since(SystemTime::now())
-        .unwrap_or(Duration::ZERO)
-    })
+  pub fn shutdown(&self) {
+    self.signal.notify_one();
   }
 }
 

src/threading/mod.rs

@@ -28,7 +28,7 @@ use std::thread::JoinHandle;
 use std::time::Duration;
 use thread_pool::WorkerPool;
 
-const DEFAULT_WAIT_NO_WORK: Duration = Duration::from_secs(1);
+const DEFAULT_WAIT_NO_WORK: Duration = Duration::from_secs(2);
 
 pub(super) struct SchedulerThread {
   halted: Arc<AtomicBool>,
@@ -51,11 +51,7 @@ impl SchedulerThread {
         .name("Quartz Scheduler Thread".to_string())
         .spawn(move || {
           while !halted.load(Acquire) {
-            let next_fire = store.next_fire().unwrap_or(DEFAULT_WAIT_NO_WORK);
-            if !next_fire.is_zero() {
-              thread::sleep(next_fire);
-            }
-            if let Some(job) = store.next_job() {
+            if let Some(job) = store.next_job(DEFAULT_WAIT_NO_WORK) {
               #[allow(clippy::unit_arg)]
               if let Err(_task) = workers.submit(job) {
                 // FIXME: no worker available! reschedule task
@@ -76,6 +72,7 @@ impl SchedulerThread {
 
   pub fn shutdown(self) {
     self.halted.store(true, std::sync::atomic::Ordering::Release);
+    self.store.shutdown();
     self.handle.join().expect("Scheduler thread panicked");
     Arc::try_unwrap(self.workers)
       .expect("worker pool is still being used!")

src/threading/thread_pool.rs

@@ -157,7 +157,7 @@ impl<T: Executable + Debug> Worker<T> {
     let l = self.task.lock().unwrap();
     self.busy.store(true, Ordering::Release);
     self.cvar.notify_one();
-    std::mem::drop(l);
+    drop(l);
   }
 }