feat(backend::scheduler) deadlock spotted

backend/server/src/scheduler/mod.rs

@@ -1,6 +1,6 @@
 use std::{
     collections::{HashMap, HashSet},
-    sync::{Arc, RwLock},
+    sync::Arc,
 };
 
 use crate::scheduler::{
@@ -9,12 +9,13 @@ use crate::scheduler::{
     tasks::CronTask,
 };
 
+use tokio::sync::RwLock;
 use uuid::Uuid;
 
 pub mod error;
 mod runner;
 mod sleeper;
-mod task_func;
+pub mod task_func;
 pub mod tasks;
 
 #[derive(Default)]
@@ -26,43 +27,42 @@ pub struct Scheduler {
 
 impl Scheduler {
     /// Start all tasks in the scheduler
-    pub fn start(&self) -> Result<(), SchedulerError> {
-        self.runner.run_all()
+    pub async fn start(&self) -> Result<(), SchedulerError> {
+        self.runner.run_all().await
     }
 
     /// Terminate all tasks in the scheduler
-    pub fn halt(&self) -> Result<(), SchedulerError> {
-        for id in self.running.read()?.iter() {
-            let func = &self.runner.terminate_remove(*id)?;
-            self.running.write()?.remove(id);
-            self.parking_pool.write()?.insert(*id, func.clone());
+    pub async fn halt(&self) -> Result<(), SchedulerError> {
+        for id in self.running.read().await.iter() {
+            let func = &self.runner.terminate_remove(*id).await?;
+            self.running.write().await.remove(id);
+            self.parking_pool.write().await.insert(*id, func.clone());
         }
         Ok(())
     }
 
     /// Run a task with `id`
-    pub fn run(&self, id: &Uuid) -> Result<(), SchedulerError> {
-        if self.running.read()?.contains(id) {
+    pub async fn run(&self, id: &Uuid) -> Result<(), SchedulerError> {
+        if self.running.read().await.contains(id) {
             Ok(())
-        } else if self.parking_pool.read()?.contains_key(id) {
-            let func = self.parking_pool.write()?.remove(id).unwrap();
-            self.running.write()?.insert(*id);
-            self.runner
-                .add_with_id(func, id)
-                .and_then(|_| self.runner.run_id(id))
+        } else if self.parking_pool.read().await.contains_key(id) {
+            let func = self.parking_pool.write().await.remove(id).unwrap();
+            self.running.write().await.insert(*id);
+            self.runner.add_with_id(func, id).await?;
+            self.runner.run_id(id).await
         } else {
             Err(IdNotFound)
         }
     }
 
     /// Stop a task with `id`
-    pub fn stop(&self, id: &Uuid) -> Result<(), SchedulerError> {
-        if self.running.read()?.contains(id) {
-            let func = self.runner.terminate_remove(*id)?;
-            self.running.write()?.remove(id);
-            self.parking_pool.write()?.insert(*id, func.clone());
+    pub async fn stop(&self, id: &Uuid) -> Result<(), SchedulerError> {
+        if self.running.read().await.contains(id) {
+            let func = self.runner.terminate_remove(*id).await?;
+            self.running.write().await.remove(id);
+            self.parking_pool.write().await.insert(*id, func.clone());
             Ok(())
-        } else if self.parking_pool.read()?.get(id).is_some() {
+        } else if self.parking_pool.read().await.get(id).is_some() {
             Ok(())
         } else {
             Err(IdNotFound)
@@ -72,17 +72,21 @@ impl Scheduler {
     // pub fn now(&self, id: &Uuid) -> Result<(), SchedulerError> {
     // }
 
-    pub fn insert(&self, task: Box<dyn CronTask>) -> Result<Uuid, SchedulerError> {
-        let id = self.runner.add(task.into())?;
-        self.running.write()?.insert(id);
+    pub async fn insert(&self, task: Box<dyn CronTask>) -> Result<Uuid, SchedulerError> {
+        let id = self.runner.add(task.into()).await?;
+        self.running.write().await.insert(id);
         Ok(id)
     }
 
-    pub fn insert_list(&self, tasks: Vec<Box<dyn CronTask>>) -> Result<Vec<Uuid>, SchedulerError> {
+    pub async fn insert_list(
+        &self,
+        tasks: Vec<Box<dyn CronTask>>,
+    ) -> Result<Vec<Uuid>, SchedulerError> {
         let ids = self
             .runner
-            .add_list(tasks.into_iter().map(|t| t.into()).collect())?;
-        self.running.write()?.extend(&ids);
+            .add_list(tasks.into_iter().map(|t| t.into()).collect())
+            .await?;
+        self.running.write().await.extend(&ids);
         Ok(ids)
     }
 
@@ -91,51 +95,51 @@ impl Scheduler {
     }
 }
 
-#[cfg(test)]
-mod tests {
-    use std::{error::Error, thread, time::Duration};
-
-    use crate::scheduler::{
-        tasks::{async_basic::AsyncBasic, basic::BasicTask},
-        Scheduler,
-    };
-
-    use uuid::Uuid;
-
-    fn execution(scheduler: Scheduler, id: Uuid) -> Result<(), Box<dyn Error>> {
-        for i in 1..=2 {
-            println!("Loop {} start", i);
-            println!("Run {}", i);
-            scheduler.run(&id)?;
-            println!("Sleep {}", i);
-            thread::sleep(Duration::from_secs(4));
-            println!("Sleep {} finished", i);
-            scheduler.stop(&id)?;
-            println!("Stop {}", i);
-            thread::sleep(Duration::from_secs(2));
-            println!("Loop {} finished", i);
-        }
-        Ok(())
-    }
-
-    #[test]
-    fn test_run() -> Result<(), Box<dyn Error>> {
-        let scheduler = Scheduler::default();
-        let task = BasicTask::new(|| println!("test print"), "* * * * * *")?;
-        let id = scheduler.insert(Box::new(task))?;
-        scheduler.start()?;
-        execution(scheduler, id)
-    }
-
-    #[test]
-    fn test_run_async() -> Result<(), Box<dyn Error>> {
-        let scheduler = Scheduler::default();
-        let task = AsyncBasic::new(
-            || Box::pin(async { println!("test async print") }),
-            "* * * * * *",
-        )?;
-        let id = scheduler.insert(Box::new(task))?;
-        scheduler.start()?;
-        execution(scheduler, id)
-    }
-}
+// #[cfg(test)]
+// mod tests {
+//     use std::{error::Error, thread, time::Duration};
+//
+//     use crate::scheduler::{
+//         tasks::{async_basic::AsyncBasic, basic::BasicTask},
+//         Scheduler,
+//     };
+//
+//     use uuid::Uuid;
+//
+//     async fn execution(scheduler: Scheduler, id: Uuid) -> Result<(), Box<dyn Error>> {
+//         for i in 1..=2 {
+//             println!("Loop {} start", i);
+//             println!("Run {}", i);
+//             scheduler.run(&id).await?;
+//             println!("Sleep {}", i);
+//             thread::sleep(Duration::from_secs(4));
+//             println!("Sleep {} finished", i);
+//             scheduler.stop(&id)?;
+//             println!("Stop {}", i);
+//             thread::sleep(Duration::from_secs(2));
+//             println!("Loop {} finished", i);
+//         }
+//         Ok(())
+//     }
+//
+//     #[tokio::test]
+//     async fn test_run() -> Result<(), Box<dyn Error>> {
+//         let scheduler = Scheduler::default();
+//         let task = BasicTask::new(|| println!("test print"), "* * * * * *")?;
+//         let id = scheduler.insert(Box::new(task)).await?;
+//         scheduler.start().await?;
+//         execution(scheduler, id)
+//     }
+//
+//     #[tokio::test]
+//     async fn test_run_async() -> Result<(), Box<dyn Error>> {
+//         let scheduler = Scheduler::default();
+//         let task = AsyncBasic::new(
+//             || Box::pin(async { println!("test async print") }),
+//             "* * * * * *",
+//         )?;
+//         let id = scheduler.insert(Box::new(task)).await?;
+//         scheduler.start().await?;
+//         execution(scheduler, id)
+//     }
+// }

backend/server/src/scheduler/runner.rs

@@ -1,20 +1,22 @@
-use std::{
-    collections::HashMap,
-    sync::{
-        atomic::{AtomicBool, Ordering}, Arc,
-        RwLock,
-    },
-};
-
 use crate::scheduler::{
     error::{SchedulerError, SchedulerError::IdNotFound},
     sleeper::SkippableSleeper,
     tasks::CronTask,
 };
+use futures::future;
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+};
 
 use tokio::runtime::Runtime;
+use tokio::sync::RwLock;
 use uuid::Uuid;
 
+#[derive(Debug)]
 struct TaskStatus {
     sleeper: RwLock<SkippableSleeper>,
     is_running: AtomicBool,
@@ -40,13 +42,22 @@ impl TaskEntity {
     }
 
     pub async fn run_blocking(&self) {
+        dbg!("entered");
+        dbg!("chkpt 1");
         loop {
-            self.status.sleeper.read().unwrap().start();
+            dbg!("chkpt 2");
+            // deadlock spotted
+            self.status.sleeper.read().await.start().await;
+            dbg!("chkpt 3");
             if self.status.expected_running.load(Ordering::SeqCst) {
+                dbg!("chkpt 4");
                 self.func.call().await;
+                dbg!("chkpt 5");
                 if let Some(next_execution) = self.func.get_next_execution() {
-                    *(self.status.sleeper.write().unwrap()) = next_execution.into();
+                    dbg!("chkpt 6a");
+                    *(self.status.sleeper.write().await) = next_execution.into();
                 } else {
+                    dbg!("chkpt 6b");
                     self.status.is_running.store(false, Ordering::SeqCst);
                     break;
                 }
@@ -56,10 +67,10 @@ impl TaskEntity {
         }
     }
 
-    pub fn terminate(&self) {
+    pub async fn terminate(&self) {
         self.status.expected_running.store(false, Ordering::SeqCst);
         self.status.is_running.store(false, Ordering::SeqCst);
-        self.status.sleeper.read().unwrap().skip();
+        self.status.sleeper.read().await.skip();
     }
 }
 
@@ -76,8 +87,8 @@ impl Runner {
         }
     }
 
-    pub fn run_all(&self) -> Result<(), SchedulerError> {
-        let list = &self.task_list.read()?;
+    pub async fn run_all(&self) -> Result<(), SchedulerError> {
+        let list = &self.task_list.read().await;
         self.rt.block_on(async {
             for (_, entity) in list.iter() {
                 self.run_entity(entity.clone()).await
@@ -86,8 +97,8 @@ impl Runner {
         Ok(())
     }
 
-    pub fn run_id(&self, id: &Uuid) -> Result<(), SchedulerError> {
-        if let Some(entity) = self.task_list.read()?.get(id).cloned() {
+    pub async fn run_id(&self, id: &Uuid) -> Result<(), SchedulerError> {
+        if let Some(entity) = self.task_list.read().await.get(id).cloned() {
             self.rt.block_on(async { self.run_entity(entity).await });
             Ok(())
         } else {
@@ -100,60 +111,67 @@ impl Runner {
             && entity.status.expected_running.load(Ordering::SeqCst)
         {
             entity.status.is_running.store(true, Ordering::SeqCst);
-            tokio::spawn({
+            let fut = {
                 let entity_clone = entity.clone();
                 async move { entity_clone.run_blocking().await }
-            })
-            .await
-            .expect("tokio runtime error");
+            };
+            dbg!("spawn");
+            tokio::spawn(fut);
         }
     }
 
-    pub fn add_with_id(
+    pub async fn add_with_id(
         &self,
         task: Arc<Box<dyn CronTask>>,
         id: &Uuid,
     ) -> Result<(), SchedulerError> {
         self.task_list
-            .write()?
+            .write()
+            .await
             .insert(*id, TaskEntity::new(task).into());
         Ok(())
     }
 
-    pub fn add(&self, task: Arc<Box<dyn CronTask>>) -> Result<Uuid, SchedulerError> {
+    pub async fn add(&self, task: Arc<Box<dyn CronTask>>) -> Result<Uuid, SchedulerError> {
         let id = Uuid::new_v4();
-        self.add_with_id(task, &id)?;
+        self.add_with_id(task, &id).await?;
         Ok(id)
     }
 
-    pub fn add_list_with_id(
+    pub async fn add_list_with_id(
         &self,
         tasks: Vec<(&Uuid, Arc<Box<dyn CronTask>>)>,
     ) -> Result<(), SchedulerError> {
-        tasks
+        let fut = tasks
             .into_iter()
-            .try_for_each(|(id, task)| self.add_with_id(task, id))
+            .map(async |(id, entity)| self.add_with_id(entity, id).await)
+            .collect::<Vec<_>>();
+        future::join_all(fut).await.into_iter().collect()
     }
 
-    pub fn add_list(
+    pub async fn add_list(
         &self,
         tasks: Vec<Arc<Box<dyn CronTask>>>,
     ) -> Result<Vec<Uuid>, SchedulerError> {
-        tasks
+        let fut = tasks
             .into_iter()
-            .map(|task| {
+            .map(async |task| {
                 let id = Uuid::new_v4();
-                self.add_with_id(task, &id).map(move |_| id)
+                self.add_with_id(task, &id).await.map(move |_| id)
             })
-            .collect::<Result<Vec<_>, _>>()
+            .collect::<Vec<_>>();
+        future::join_all(fut).await.into_iter().collect()
     }
 
-    pub fn terminate_remove(&self, id: Uuid) -> Result<Arc<Box<dyn CronTask>>, SchedulerError> {
-        let mut list = self.task_list.write()?;
+    pub async fn terminate_remove(
+        &self,
+        id: Uuid,
+    ) -> Result<Arc<Box<dyn CronTask>>, SchedulerError> {
+        let mut list = self.task_list.write().await;
         match list.get(&id) {
             None => Err(IdNotFound),
             Some(task) => {
-                task.terminate();
+                task.terminate().await;
                 let func = list.remove(&id).unwrap().func.clone();
                 Ok(func.clone())
             }

backend/server/src/scheduler/sleeper.rs

@@ -1,80 +1,81 @@
-use std::{
-    sync::{Arc, Condvar, Mutex},
-    thread,
-    time::Duration,
-};
-
 use chrono::{DateTime, Utc};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::{sync::Arc, time::Duration};
+use tokio::sync::Notify;
 
-#[derive(Clone, Debug, Default)]
+#[derive(Debug, Clone)]
 pub struct SkippableSleeper {
-    pub duration: Duration,
-    state: Arc<(Mutex<bool>, Condvar)>,
+    duration: Duration,
+    skipped: Arc<AtomicBool>,
+    notify: Arc<Notify>,
 }
 
 impl SkippableSleeper {
     pub fn new(duration: Duration) -> Self {
-        SkippableSleeper {
+        Self {
             duration,
-            state: Arc::new((Mutex::new(false), Condvar::new())),
+            skipped: Arc::new(AtomicBool::new(false)),
+            notify: Arc::new(Notify::new()),
         }
     }
 
-    // Blocks until the duration has passed or skip() is called
-    pub fn start(&self) {
-        let (lock, cvar) = &*self.state;
-        let mut received = lock.lock().unwrap();
-        let state_clone = Arc::clone(&self.state);
-        let duration_clone = self.duration;
-        thread::spawn(move || {
-            thread::sleep(duration_clone);
-            let (lock, cvar) = &*state_clone;
-            let mut received = lock.lock().unwrap();
-            *received = true;
-            cvar.notify_one();
-        });
-        while !*received {
-            received = cvar.wait(received).unwrap();
+    pub async fn start(&self) {
+        if self.skipped.load(Ordering::Acquire) {
+            return;
+        }
+
+        let sleep = tokio::time::sleep(self.duration);
+        let notify = self.notify.notified();
+
+        tokio::select! {
+            _ = sleep => {}
+            _ = notify => {
+                self.skipped.store(true, Ordering::Release);
+            }
         }
     }
 
-    // Skip the blocking delay
     pub fn skip(&self) {
-        let (lock, cvar) = &*self.state;
-        let mut received = lock.lock().unwrap();
-        *received = true;
-        cvar.notify_one();
+        self.notify.notify_waiters();
     }
 }
 
 impl From<DateTime<Utc>> for SkippableSleeper {
     fn from(value: DateTime<Utc>) -> Self {
-        let from_utc_timestamp = value.timestamp() as u64;
-        let curr_utc_timestamp = Utc::now().timestamp() as u64;
-        SkippableSleeper::new(Duration::from_secs(from_utc_timestamp - curr_utc_timestamp))
+        let now = Utc::now();
+        let duration = value.signed_duration_since(now);
+        let duration = duration.to_std().unwrap_or(Duration::ZERO);
+        Self::new(duration)
     }
 }
 
 #[cfg(test)]
-mod test {
-    use crate::scheduler::sleeper::SkippableSleeper;
-    use std::thread;
-    use std::time::Duration;
+mod tests {
+    use super::*;
+    use tokio::time::{sleep, Duration};
 
-    #[test]
-    fn usage() {
-        let sleeper = SkippableSleeper::new(Duration::new(10, 0));
-        thread::spawn({
-            let sleeper = sleeper.clone();
-            move || {
-                println!("Sleeping...");
-                sleeper.start();
-                println!("Woke up!");
-            }
+    #[tokio::test]
+    async fn skip_sleeper() {
+        let sleeper = SkippableSleeper::new(Duration::from_secs(10));
+        let sleeper_clone = sleeper.clone();
+
+        let handle = tokio::spawn(async move {
+            println!("Sleeping...");
+            sleeper_clone.start().await;
+            println!("Woke up!");
         });
-        println!("Start Sleeping...");
-        thread::sleep(Duration::from_secs(4));
+
+        println!("Start waiting...");
+        sleep(Duration::from_secs(4)).await;
         println!("Skipping sleep...");
         sleeper.skip();
+
+        handle.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn normal_sleeper() {
+        let sleeper = SkippableSleeper::new(Duration::from_secs(4));
+        sleeper.start().await;
     }
 }