feat: Optimized use of internal time-wheel memory

CHANGELOG.md

@@ -1,3 +1,15 @@
+# Version 0.10.0 
+
+## Changed
+Optimized the use of internal `time-wheel` memory. ([#28](https://github.com/BinChengZhao/delay-timer/issues/28)), thanks `elderbig` !
+
+### Details
+There is a `time-wheel` in `delay-timer`, which is the carrier of all tasks.
+
+The time wheel uses slots (time scales) as units, each slot corresponds to a hash table, when a slot is rotated to it will execute the task that is already ready internally, when the task is executed it will move from one slot to another. In order to have enough capacity to store the tasks, there may be a memory allocation here, so that by the time the whole time wheel is traversed, each internal time wheel-slot will have rich memory capacity, and when there are many tasks the memory occupied by the whole time wheel will be very large. So it will be necessary to shrink the memory in time.
+
+This change is to shrink the memory in time after each round of training slots and executing tasks to ensure that the slots have a basic and compact capacity.
+
 # Version 0.9.2
 
 ## Changed

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "delay_timer"
-version = "0.9.2"
+version = "0.10.0"
 authors = ["binchengZhao <[email protected]>"]
 edition = "2018"
 repository = "https://github.com/BinChengZhao/delay-timer"
@@ -49,6 +49,8 @@ tokio = { version = "^1.3.0", features = ["full"] , optional = true }
 [dev-dependencies]
 rand = "0.8.4"
 surf = "^2.1.0"
+tracing = "0.1.29"
+tracing-subscriber = "0.2.0"
 tokio = { version = "^1.3.0", features = ["full"] }
 hyper= {version  = "^0.14.2" , features = ["full"] }
 pretty_env_logger = "^0.4"

README.md

@@ -194,7 +194,7 @@ pub fn generate_closure_template(
 
         let future = async move {
             future_inner.await;
-            context.finishe_task().await;
+            context.finish_task().await;
         };
         create_delay_task_handler(async_spawn(future))
     }

examples/cycle_tokio_task.rs

@@ -64,7 +64,7 @@ pub fn generate_closure_template(
 
         let future = async move {
             future_inner.await.ok();
-            context.finishe_task(None).await;
+            context.finish_task(None).await;
         };
 
         create_delay_task_handler(async_spawn_by_tokio(future))

examples/demo.rs

@@ -110,7 +110,7 @@ pub fn generate_closure_template(
 
         let future = async move {
             future_inner.await;
-            context.finishe_task(None).await;
+            context.finish_task(None).await;
         };
         create_delay_task_handler(async_spawn(future))
     }

examples/demo_async_std_and_tracing.rs

@@ -0,0 +1,42 @@
+#![allow(deprecated)]
+
+use anyhow::Result;
+use delay_timer::prelude::*;
+use smol::Timer;
+use std::time::Duration;
+use tracing::{info, Level};
+use tracing_subscriber::FmtSubscriber;
+
+#[async_std::main]
+async fn main() -> Result<()> {
+    // a builder for `FmtSubscriber`.
+    FmtSubscriber::builder()
+        // all spans/events with a level higher than TRACE (e.g, debug, info, warn, etc.)
+        // will be written to stdout.
+        .with_max_level(Level::DEBUG)
+        // completes the builder.
+        .init();
+
+    let delay_timer = DelayTimerBuilder::default().build();
+    for i in 0..1000 {
+        delay_timer.add_task(build_task_async_execute_process(i)?)?;
+    }
+
+    info!("==== All job is be init! ====");
+    for _ in 0..120 {
+        Timer::after(Duration::from_secs(60)).await;
+    }
+    Ok(delay_timer.stop_delay_timer()?)
+}
+
+fn build_task_async_execute_process(task_id: u64) -> Result<Task, TaskError> {
+    let mut task_builder = TaskBuilder::default();
+
+    let body = unblock_process_task_fn("echo hello".into());
+    task_builder
+        .set_frequency_by_candy(CandyFrequency::Repeated(CandyCron::Secondly))
+        .set_task_id(task_id)
+        .set_maximum_running_time(10)
+        .set_maximum_parallel_runnable_num(1)
+        .spawn(body)
+}

src/lib.rs

@@ -207,7 +207,7 @@
 //!
 //!         let future = async move {
 //!             future_inner.await;
-//!             context.finishe_task(None).await;
+//!             context.finish_task(None).await;
 //!         };
 //!
 //!         create_delay_task_handler(async_spawn(future))

src/macros/generate_fn_macro.rs

@@ -14,7 +14,7 @@ macro_rules! create_async_fn_body {
                 let future_inner = async move { $async_body };
                 future_inner.await;
 
-                context.finishe_task(None).await;
+                context.finish_task(None).await;
             };
             let handle = async_spawn(f);
             create_delay_task_handler(handle)
@@ -34,7 +34,7 @@ macro_rules! create_async_fn_body {
                 let future_inner = async move { $async_body };
                 future_inner.await;
 
-                context.finishe_task(None).await;
+                context.finish_task(None).await;
             };
             let handle = async_spawn(f);
             create_delay_task_handler(handle)
@@ -53,7 +53,7 @@ cfg_tokio_support!(
                     let future_inner = async move { $async_body };
                     future_inner.await;
 
-                    context.finishe_task().await;
+                    context.finish_task().await;
                 });
                 create_delay_task_handler(handle)
             }
@@ -71,7 +71,7 @@ cfg_tokio_support!(
                         let future_inner = async move { $async_body };
                         future_inner.await;
 
-                        context.finishe_task().await;
+                        context.finish_task().await;
                     };
                     let handle = async_spawn(f);
                     create_delay_task_handler(handle)

src/timer/slot.rs

@@ -38,8 +38,8 @@ impl Slot {
         self.task_map.remove(&task_id)
     }
 
-    //Check and reduce cylinder_line，
-    //Returns a Vec. containing all task ids to be executed.(cylinder_line == 0)
+    // Check and reduce cylinder_line，
+    // Returns a Vec. containing all task ids to be executed.(cylinder_line == 0)
     pub(crate) fn arrival_time_tasks(&mut self) -> Vec<u64> {
         let mut task_id_vec = vec![];
 
@@ -51,4 +51,11 @@ impl Slot {
 
         task_id_vec
     }
+
+    // When the operation is finished with the task, shrink the container in time
+    // To avoid the overall time-wheel from occupying too much memory.
+    // FIX: https://github.com/BinChengZhao/delay-timer/issues/28
+    pub(crate) fn shrink(&mut self) {
+        self.task_map.shrink_to(64);
+    }
 }

src/timer/task.rs

@@ -498,7 +498,7 @@ impl TaskContext {
     }
 
     /// Send a task-Finish signal to EventHandle.
-    pub async fn finishe_task(self, finish_output: Option<FinishOutput>) {
+    pub async fn finish_task(self, finish_output: Option<FinishOutput>) {
         if let Some(timer_event_sender) = self.timer_event_sender {
             timer_event_sender
                 .send(TimerEvent::FinishTask(FinishTaskBody {

src/timer/timer_core.rs

@@ -151,6 +151,7 @@ pub enum TimerEvent {
 pub struct Timer {
     /// Event sender that provides events to `EventHandle` processing.
     pub(crate) timer_event_sender: TimerEventSender,
+    #[allow(dead_code)]
     status_report_sender: Option<AsyncSender<i32>>,
     pub(crate) shared_header: SharedHeader,
 }
@@ -237,6 +238,14 @@ impl Timer {
                 }
             }
 
+            {
+                // When the operation is finished with the task, shrink the container in time
+                // To avoid the overall time-wheel from occupying too much memory.
+                if let Some(mut slot_mut) = self.shared_header.wheel_queue.get_mut(&second_hand) {
+                    slot_mut.shrink();
+                }
+            }
+
             clock.tick().await;
         }
     }

src/utils/convenience.rs

@@ -25,13 +25,25 @@ pub mod functions {
     ) -> impl Fn(TaskContext) -> Box<dyn DelayTaskHandler> + 'static + Send + Sync {
         use smol::process::{Child, Command};
         move |context: TaskContext| {
+            debug!(
+                "Unblock-Process Task: {}, Record: {} Start",
+                context.task_id, context.record_id
+            );
+
             let shell_command_clone = shell_command.clone();
             create_delay_task_handler(async_spawn(async move {
                 let childs = parse_and_run::<Child, Command>(&shell_command_clone).await;
 
                 if let Err(err) = childs {
+                    debug!(
+                        "Unblock-Process Task: {}, Record: {} fail for: {}",
+                        context.task_id,
+                        context.record_id,
+                        err.to_string()
+                    );
+
                     context
-                        .finishe_task(Some(FinishOutput::ExceptionOutput(err.to_string())))
+                        .finish_task(Some(FinishOutput::ExceptionOutput(err.to_string())))
                         .await;
                     return Err(anyhow!(err.to_string()));
                 }
@@ -40,8 +52,14 @@ pub mod functions {
 
                 let last_child = childs.pop_back().ok_or_else(|| anyhow!("Without child."))?;
                 let output = last_child.wait_with_output().await?;
+
+                debug!(
+                    "Unblock-Process Task: {}, Record: {} finished",
+                    context.task_id, context.record_id
+                );
+
                 context
-                    .finishe_task(Some(FinishOutput::ProcessOutput(output)))
+                    .finish_task(Some(FinishOutput::ProcessOutput(output)))
                     .await;
 
                 Ok(())
@@ -63,7 +81,7 @@ pub mod functions {
 
                     if let Err(err) = childs {
                         context
-                            .finishe_task(Some(FinishOutput::ExceptionOutput(err.to_string())))
+                            .finish_task(Some(FinishOutput::ExceptionOutput(err.to_string())))
                             .await;
                         return Err(anyhow!(err.to_string()));
                     }
@@ -73,7 +91,7 @@ pub mod functions {
                     let last_child = childs.pop_back().ok_or_else(|| anyhow!("Without child."))?;
                     let output = last_child.wait_with_output().await?;
                     context
-                        .finishe_task(Some(FinishOutput::ProcessOutput(output)))
+                        .finish_task(Some(FinishOutput::ProcessOutput(output)))
                         .await;
 
                     Ok(())