Merge pull request #19 from BinChengZhao/feature/stable

Version 0.8.0

Author: BinChengZhao

.gitignore

@@ -3,7 +3,7 @@
 /target/
 /examples/target/
 /examples/*.lock
-
+.DS_Store
 
 # Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
 # More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html

CHANGELOG.md

@@ -1,3 +1,17 @@
+# Version 0.8.0
+
+v0.8.0 New features:
+
+    1. Optimized the api for custom tokio runtime, better api definition, better experience.
+
+      Api changes.
+        1.  Added: `tokio_runtime_by_default` & `tokio_runtime_by_custom` & `tokio_runtime_shared_by_custom` .
+
+        2. Private: `tokio_runtime`.
+  
+    2. Optimized the method of canceling task instances, no error log is recorded when canceling a task failure after timeout.
+
+    3. Fix the compile error about `quit_one_task_handler` under nightly version.
 # Version 0.7.0
 
 v0.7.0 New features:

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "delay_timer"
-version = "0.7.2"
+version = "0.8.0"
 authors = ["binchengZhao <binchengZhao@outlook.com>"]
 edition = "2018"
 repository = "https://github.com/BinChengZhao/delay-timer"

examples/cycle_tokio_task.rs

@@ -8,7 +8,9 @@ use std::thread::{current, park, Thread};
 
 fn main() -> Result<()> {
     // Build an DelayTimer that uses the default configuration of the Tokio runtime internally.
-    let delay_timer = DelayTimerBuilder::default().tokio_runtime(None).build();
+    let delay_timer = DelayTimerBuilder::default()
+        .tokio_runtime_by_default()
+        .build();
 
     // Develop a task that runs in an asynchronous cycle (using a custom asynchronous template).
     delay_timer.add_task(build_task_customized_async_task()?)?;

examples/dynamic_cancel.rs

@@ -37,7 +37,7 @@ fn async_cancel() -> Result<()> {
 
     // Build an DelayTimer that uses the Customize a tokio runtime.
     let delay_timer = DelayTimerBuilder::default()
-        .tokio_runtime(Some(tokio_rt.clone()))
+        .tokio_runtime_shared_by_custom(tokio_rt.clone())
         .build();
 
     tokio_rt.block_on(async {

examples/generic.rs

@@ -0,0 +1,69 @@
+use anyhow::Result;
+use delay_timer::prelude::*;
+use smol::Timer;
+use std::any::{type_name, Any};
+use std::thread::park_timeout;
+use std::time::Duration;
+use surf;
+
+// cargo run --package delay_timer --example generic --features=full
+
+fn main() -> Result<()> {
+    let delay_timer = DelayTimerBuilder::default().enable_status_report().build();
+
+    // Develop tasks with generic parameters. that runs in an asynchronous cycle.
+    delay_timer.add_task(build_generic_task_async_request(Dog)?)?;
+
+    // Give a little time to observe the output.
+    park_timeout(Duration::from_secs(30));
+
+    // No new tasks are accepted; running tasks are not affected.
+    delay_timer.stop_delay_timer()?;
+
+    Ok(())
+}
+
+fn build_generic_task_async_request<T: Animal>(animal: T) -> Result<Task, TaskError> {
+    let mut task_builder = TaskBuilder::default();
+
+    let other_animal = Cat;
+    let int_animal = 1;
+
+    let body = create_async_fn_body!((animal, other_animal, int_animal){
+        if let Ok(mut res) = surf::get("https://httpbin.org/get").await {
+            dbg!(res.body_string().await.unwrap_or_default());
+            animal_ref.call();
+            other_animal_ref.call();
+            <i32 as Animal>::call(&int_animal_ref);
+
+
+            Timer::after(Duration::from_secs(3)).await;
+            dbg!("Task2 is done.");
+        }
+    });
+
+    task_builder
+        .set_frequency(Frequency::CountDown(15, "* * * * * * *"))
+        .set_task_id(2)
+        .set_maximum_running_time(5)
+        .spawn(body)
+}
+
+trait ThreadSafe: Any + Sized + Clone + Send + Sync + 'static {}
+impl<T: Any + Sized + Clone + Send + Sync + 'static> ThreadSafe for T {}
+
+trait Animal: ThreadSafe {
+    fn call(&self);
+}
+
+impl<T: ThreadSafe> Animal for T {
+    fn call(&self) {
+        println!("this is {}", type_name::<T>());
+    }
+}
+
+#[derive(Clone, Copy)]
+struct Dog;
+
+#[derive(Clone, Copy)]
+struct Cat;

src/entity.rs

@@ -116,7 +116,7 @@ impl fmt::Debug for SharedHeader {
 
 #[derive(Clone, Default, Debug)]
 pub(crate) struct RuntimeInstance {
-    //smol have no instance.
+    // smol have no instance.
     #[cfg(feature = "tokio-support")]
     pub(crate) inner: Option<Arc<Runtime>>,
     pub(crate) kind: RuntimeKind,
@@ -335,29 +335,27 @@ cfg_tokio_support!(
 ///
 /// This function requires the `tokio-support` feature of the `delay_timer`
 /// crate to be enabled.
-   impl DelayTimerBuilder{
-
-    fn assign_task_by_tokio(&mut self, timer: Timer,event_handle: EventHandle) {
+impl DelayTimerBuilder {
+    fn assign_task_by_tokio(&mut self, timer: Timer, event_handle: EventHandle) {
         self.run_async_schedule_by_tokio(timer);
         self.run_event_handle_by_tokio(event_handle);
     }
 
-    fn run_async_schedule_by_tokio(&self, mut timer: Timer){
-       if let Some(ref tokio_runtime_ref) = self.shared_header.runtime_instance.inner{
-           let tokio_runtime = tokio_runtime_ref.clone();
-        Builder::new()
-        .name("async_schedule_tokio".into())
-        .spawn(move || {
-            tokio_runtime.block_on(async {
-                timer.async_schedule().await;
-            })
-        })
-        .expect("async_schedule can't start.");
-       }
-
+    fn run_async_schedule_by_tokio(&self, mut timer: Timer) {
+        if let Some(ref tokio_runtime_ref) = self.shared_header.runtime_instance.inner {
+            let tokio_runtime = tokio_runtime_ref.clone();
+            Builder::new()
+                .name("async_schedule_tokio".into())
+                .spawn(move || {
+                    tokio_runtime.block_on(async {
+                        timer.async_schedule().await;
+                    })
+                })
+                .expect("async_schedule can't start.");
+        }
     }
 
-    fn run_event_handle_by_tokio(&self, mut event_handle: EventHandle){
+    fn run_event_handle_by_tokio(&self, mut event_handle: EventHandle) {
         if let Some(ref tokio_runtime_ref) = self.shared_header.runtime_instance.inner {
             let tokio_runtime = tokio_runtime_ref.clone();
             Builder::new()
@@ -369,44 +367,58 @@ cfg_tokio_support!(
                 })
                 .expect("event_handle_handle_by_tokio can't start.");
         }
+    }
 
-     }
-
-     /// With this API, let DelayTimer internally use the user custom TokioRuntime.
-     /// If None is given, a custom TokioRuntime is generated internally.
-    pub fn tokio_runtime(mut self, rt:Option<Arc<Runtime>>) ->Self {
-        self.shared_header.register_tokio_runtime(rt);
-        self
+    /// With this API, `DelayTimer` use default `TokioRuntime` is generated internally.
+    pub fn tokio_runtime_by_default(self) -> Self {
+        self.tokio_runtime(None)
     }
-   }
 
-   impl SharedHeader {
-        pub(crate) fn tokio_support() -> Option<Runtime> {
-            TokioBuilder::new_multi_thread().enable_all()
-                .thread_name_fn(|| {
-                    static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
-                    let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
-                    format!("tokio-{}", id)
-                })
-                .on_thread_start(|| {
-                    info!("tokio-thread started");
-                })
-                .build()
-                .ok()
-        }
+    /// With this API, `DelayTimer` internally use the user customized and independent `TokioRuntime`.
+    pub fn tokio_runtime_by_custom(self, rt: Runtime) -> Self {
+        self.tokio_runtime(Some(Arc::new(rt)))
+    }
 
-        pub(crate) fn register_tokio_runtime(&mut self,mut rt:Option<Arc<Runtime>>) {
+    /// With this api, `DelayTimer` internal will share a `TokioRuntime` with the user .
+    pub fn tokio_runtime_shared_by_custom(self, rt: Arc<Runtime>) -> Self {
+        self.tokio_runtime(Some(rt))
+    }
 
-            if rt.is_none(){
-              rt = Some(Arc::new(Self::tokio_support().expect("init tokioRuntime is fail.")));
-            }
+    /// With this API, `DelayTimer` internally use the user custom TokioRuntime.
+    /// If None is given, a default TokioRuntime is generated internally.
+    pub(crate) fn tokio_runtime(mut self, rt: Option<Arc<Runtime>>) -> Self {
+        self.shared_header.register_tokio_runtime(rt);
+        self
+    }
+}
 
-            self.runtime_instance.inner = rt;
-            self.runtime_instance.kind = RuntimeKind::Tokio;
-        }
+impl SharedHeader {
+    pub(crate) fn tokio_support() -> Option<Runtime> {
+        TokioBuilder::new_multi_thread()
+            .enable_all()
+            .thread_name_fn(|| {
+                static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
+                let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
+                format!("tokio-{}", id)
+            })
+            .on_thread_start(|| {
+                info!("tokio-thread started");
+            })
+            .build()
+            .ok()
     }
 
+    pub(crate) fn register_tokio_runtime(&mut self, mut rt: Option<Arc<Runtime>>) {
+        if rt.is_none() {
+            rt = Some(Arc::new(
+                Self::tokio_support().expect("init tokioRuntime is fail."),
+            ));
+        }
 
+        self.runtime_instance.inner = rt;
+        self.runtime_instance.kind = RuntimeKind::Tokio;
+    }
+}
 
 );
 

src/timer/event_handle.rs

@@ -148,8 +148,8 @@ impl EventHandle {
         }
     );
 
-    //handle all event.
-    //TODO:Add TestUnit.
+    // handle all event.
+    // TODO: Add TestUnit.
     pub(crate) async fn lauch(&mut self) {
         self.init_sub_workers();
         self.handle_event().await;
@@ -231,11 +231,11 @@ impl EventHandle {
                 remove_result
             }
             TimerEvent::CancelTask(task_id, record_id) => {
-                self.cancel_task(task_id, record_id, state::instance::CANCELLED)
+                self.cancel_task::<true>(task_id, record_id, state::instance::CANCELLED)
             }
 
             TimerEvent::TimeoutTask(task_id, record_id) => {
-                self.cancel_task(task_id, record_id, state::instance::TIMEOUT)
+                self.cancel_task::<false>(task_id, record_id, state::instance::TIMEOUT)
             }
 
             TimerEvent::AppendTaskHandle(task_id, delay_task_handler_box) => {
@@ -384,13 +384,37 @@ impl EventHandle {
         ))
     }
 
-    pub(crate) fn cancel_task(&mut self, task_id: u64, record_id: i64, state: usize) -> Result<()> {
+    // The `INITIATIVE` mark indicates whether the cancellation was initiated by an outside party.
+
+    // `INITIATIVE` = true
+    // External initiative to cancel the action,
+    // If the cancellation fails then the error log record needs to be kept.
+
+    // `INITIATIVE` = false
+    // Passive cancellation at runtime (e.g., timeout) indicates that
+    // The task instance has completed or has been actively cancelled,
+    // And no error logging is required.
+    pub(crate) fn cancel_task<const INITIATIVE: bool>(
+        &mut self,
+        task_id: u64,
+        record_id: i64,
+        state: usize,
+    ) -> Result<()> {
         if let Some(mut task_mark_ref_mut) = self.shared_header.task_flag_map.get_mut(&task_id) {
             let task_mark = task_mark_ref_mut.value_mut();
 
             // The cancellation operation is executed first, and then the outside world is notified of the cancellation event.
             // If the operation object does not exist in the middle, it should return early.
-            self.task_trace.quit_one_task_handler(task_id, record_id)?;
+
+            let quit_result = self.task_trace.quit_one_task_handler(task_id, record_id);
+
+            if quit_result.is_err() {
+                if INITIATIVE {
+                    quit_result?;
+                } else {
+                    return Ok(());
+                }
+            }
 
             if task_mark.task_instances_chain_maintainer.is_some() {
                 // Here the user can be notified that the task instance has disappeared via `Instance`.
@@ -401,6 +425,7 @@ impl EventHandle {
 
             return Ok(());
         }
+
         Err(anyhow!(
             "Fn : `cancel_task`, Without the `task_mark_ref_mut` for task_id :{}, record_id : {}, state : {}",
             task_id,

src/timer/runtime_trace/sweeper.rs

@@ -145,6 +145,7 @@ impl RecyclingBins {
                 match self.recycle_unit_sources.recv().await {
                     Ok(recycle_unit) => {
                         let mut recycle_unit_heap = self.recycle_unit_heap.lock().await;
+
                         (&mut recycle_unit_heap).push(Reverse(recycle_unit));
                     }
 

src/timer/runtime_trace/task_handle.rs

@@ -48,8 +48,8 @@ impl TaskTrace {
                 task_id, record_id
             )
         })?;
-
-        let mut list_mut_cursor = task_handler_list.cursor_back_mut();
+      
+        let mut list_mut_cursor = task_handler_list.cursor_front_mut();
 
         let mut task_handler_box_ref: &mut DelayTaskHandlerBox;
         loop {
@@ -74,7 +74,7 @@ impl TaskTrace {
             list_mut_cursor.move_next();
         }
 
-        //remove current task_handler_box.
+        // remove current task_handler_box.
         list_mut_cursor
             .remove_current()
             .map(|mut task_handler_box| task_handler_box.quit())