fix: issue #36.

Author: BinChengZhao

benches/body.rs

@@ -63,6 +63,6 @@ fn bench_try_wait(b: &mut Bencher) {
 }
 
 #[bench]
-fn bench_get_timestamp(b: &mut Bencher) {
-    b.iter(|| get_timestamp());
+fn bench_timestamp(b: &mut Bencher) {
+    b.iter(|| timestamp());
 }

examples/cycle_tokio_task.rs

@@ -57,7 +57,7 @@ fn build_wake_task() -> Result<Task> {
 
 pub async fn generate_closure_template() {
     let name: String = "'delay_timer-is-easy-to-use.'".into();
-    let future_inner = async_template(get_timestamp() as i32, name.clone());
+    let future_inner = async_template(timestamp() as i32, name.clone());
     future_inner.await.ok();
 }
 

examples/demo.rs

@@ -103,7 +103,7 @@ fn build_task_customized_async_task() -> Result<Task, TaskError> {
     let body = move || {
         let name = name.clone();
         async move {
-            async_template(get_timestamp() as i32, name).await;
+            async_template(timestamp() as i32, name).await;
         }
     };
     #[allow(deprecated)]

examples/increase.rs

@@ -90,7 +90,7 @@ fn get_wake_fn(
         unsafe {
             println!(
                 "end time {}, result {}",
-                get_timestamp(),
+                timestamp(),
                 (*local_run_flag).load(SeqCst)
             );
         }

src/entity.rs

@@ -177,7 +177,7 @@ impl Default for SharedHeader {
         let wheel_queue = EventHandle::init_task_wheel(DEFAULT_TIMER_SLOT_COUNT);
         let task_flag_map = Arc::new(DashMap::new());
         let second_hand = Arc::new(AtomicU64::new(0));
-        let global_time = Arc::new(AtomicU64::new(get_timestamp()));
+        let global_time = Arc::new(AtomicU64::new(timestamp()));
         let shared_motivation = Arc::new(AtomicBool::new(true));
         let runtime_instance = RuntimeInstance::default();
         let id_generator = Arc::new(AsyncMutex::new(SnowflakeIdGenerator::new(1, 1)));
@@ -510,7 +510,7 @@ cfg_status_report!(
 // to get rid of system interference,
 // and this change can also be applied to snowflake-rs.
 /// get current OS SystemTime.
-pub fn get_timestamp() -> u64 {
+pub fn timestamp() -> u64 {
     match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
         Ok(n) => n.as_secs(),
         Err(_) => panic!("SystemTime before UNIX EPOCH!"),
@@ -522,7 +522,7 @@ pub fn get_timestamp() -> u64 {
 // to get rid of system interference,
 // and this change can also be applied to snowflake-rs.
 /// get current OS SystemTime.
-pub fn get_timestamp_micros() -> u128 {
+pub fn timestamp_micros() -> u128 {
     match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
         Ok(n) => n.as_micros(),
         Err(_) => panic!("SystemTime before UNIX EPOCH!"),

src/prelude.rs

@@ -10,7 +10,7 @@
 //!
 //! The prelude may grow over time as additional items see ubiquitous use.
 
-pub use crate::entity::{get_timestamp, get_timestamp_micros, DelayTimer, DelayTimerBuilder};
+pub use crate::entity::{timestamp, timestamp_micros, DelayTimer, DelayTimerBuilder};
 pub use crate::error::*;
 pub use crate::timer::runtime_trace::state::instance;
 pub use crate::timer::runtime_trace::task_handle::DelayTaskHandler;

src/timer/runtime_trace/sweeper.rs

@@ -98,7 +98,7 @@ impl RecyclingBins {
         loop {
             let mut recycle_unit_heap = self.recycle_unit_heap.lock().await;
 
-            let now: u64 = get_timestamp();
+            let now: u64 = timestamp();
             let mut duration: Option<Duration> = None;
             for _ in 0..200 {
                 if let Some(recycle_flag) = (&recycle_unit_heap).peek().map(|r| r.0.deadline <= now)
@@ -185,7 +185,7 @@ mod tests {
 
     #[test]
     fn test_task_valid() -> AnyResult<()> {
-        use super::{get_timestamp, RecycleUnit, RecyclingBins, RuntimeKind, TimerEvent};
+        use super::{timestamp, RecycleUnit, RecyclingBins, RuntimeKind, TimerEvent};
         use smol::{
             block_on,
             channel::{unbounded, TryRecvError},
@@ -216,7 +216,7 @@ mod tests {
             })
         });
 
-        let deadline = get_timestamp() + 5;
+        let deadline = timestamp() + 5;
 
         for i in 1..10 {
             recycle_unit_sender.try_send(RecycleUnit::new(deadline, i, (i * i) as i64))?;

src/timer/task.rs

@@ -228,15 +228,15 @@ impl<'a> TryFrom<(FrequencyUnify<'a>, ScheduleIteratorTimeZone)> for FrequencyIn
                     return Err(FrequencyAnalyzeError::DisInitTime);
                 }
 
-                let seconds_state: SecondsState = (get_timestamp()..).step_by(seconds as usize);
+                let seconds_state: SecondsState = (timestamp()..).step_by(seconds as usize);
                 FrequencyInner::SecondsCountDown(1, seconds_state)
             }
             FrequencyUnify::FrequencySeconds(FrequencySeconds::Repeated(seconds)) => {
                 if seconds == 0 {
                     return Err(FrequencyAnalyzeError::DisInitTime);
                 }
 
-                let seconds_state: SecondsState = (get_timestamp()..).step_by(seconds as usize);
+                let seconds_state: SecondsState = (timestamp()..).step_by(seconds as usize);
 
                 FrequencyInner::SecondsRepeated(seconds_state)
             }
@@ -245,7 +245,7 @@ impl<'a> TryFrom<(FrequencyUnify<'a>, ScheduleIteratorTimeZone)> for FrequencyIn
                     return Err(FrequencyAnalyzeError::DisInitTime);
                 }
 
-                let seconds_state: SecondsState = (get_timestamp()..).step_by(seconds as usize);
+                let seconds_state: SecondsState = (timestamp()..).step_by(seconds as usize);
                 FrequencyInner::SecondsCountDown(count_down, seconds_state)
             }
         };
@@ -513,7 +513,7 @@ impl TaskContext {
                 .send(TimerEvent::FinishTask(FinishTaskBody {
                     task_id: self.task_id,
                     record_id: self.record_id,
-                    finish_time: get_timestamp(),
+                    finish_time: timestamp(),
                     finish_output,
                 }))
                 .await
@@ -1008,7 +1008,7 @@ impl<'a> TaskBuilder<'a> {
 
     pub fn set_frequency_once_by_timestamp_seconds(&mut self, timestamp_seconds: u64) -> &mut Self {
         let duration = timestamp_seconds
-            .checked_sub(get_timestamp())
+            .checked_sub(timestamp())
             .unwrap_or(ONE_SECOND);
 
         self.frequency = FrequencyUnify::FrequencySeconds(FrequencySeconds::Once(duration));
@@ -1148,7 +1148,7 @@ mod tests {
             .map(|i| {
                 debug_assert_eq!(
                     task.get_next_exec_timestamp().unwrap(),
-                    get_timestamp() + (init_seconds * (i - 1))
+                    timestamp() + (init_seconds * (i - 1))
                 );
             })
             .for_each(drop);
@@ -1180,7 +1180,7 @@ mod tests {
             .map(|i| {
                 debug_assert_eq!(
                     task.get_next_exec_timestamp().unwrap(),
-                    get_timestamp() + (init_minutes * (i - 1) * ONE_MINUTE)
+                    timestamp() + (init_minutes * (i - 1) * ONE_MINUTE)
                 );
             })
             .for_each(drop);
@@ -1201,7 +1201,7 @@ mod tests {
             .map(|i| {
                 debug_assert_eq!(
                     task.get_next_exec_timestamp().unwrap(),
-                    get_timestamp() + (init_hours * (i - 1) * ONE_HOUR)
+                    timestamp() + (init_hours * (i - 1) * ONE_HOUR)
                 );
             })
             .for_each(drop);
@@ -1222,7 +1222,7 @@ mod tests {
             .map(|i| {
                 debug_assert_eq!(
                     task.get_next_exec_timestamp().unwrap(),
-                    get_timestamp() + (init_days * (i - 1) * ONE_DAY)
+                    timestamp() + (init_days * (i - 1) * ONE_DAY)
                 );
             })
             .for_each(drop);

src/timer/timer_core.rs

@@ -2,7 +2,7 @@
 //! It is the core of the entire cycle scheduling task.
 use crate::prelude::*;
 
-use crate::entity::get_timestamp;
+use crate::entity::timestamp;
 use crate::entity::RuntimeKind;
 
 use std::mem::replace;
@@ -15,10 +15,12 @@ use smol::Timer as smolTimer;
 pub(crate) const DEFAULT_TIMER_SLOT_COUNT: u64 = 3600;
 
 /// The clock of timer core.
+#[derive(Debug)]
 struct Clock {
     inner: ClockInner,
 }
 
+#[derive(Debug)]
 enum ClockInner {
     Sc(SmolClock),
 
@@ -52,7 +54,7 @@ impl Clock {
             ClockInner::Sc(ref mut smol_clock) => smol_clock.tick().await,
 
             ClockInner::Tc(ref mut tokio_clock) => tokio_clock.tick().await,
-        }
+        };
     }
 }
 use tokio::time::{self, interval_at, Interval};
@@ -142,14 +144,15 @@ pub enum TimerEvent {
     /// Take the initiative to perform once Task.
     AdvanceTask(u64),
 }
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 /// delay-timer internal timer wheel core.
 pub struct Timer {
     /// Event sender that provides events to `EventHandle` processing.
-    pub(crate) timer_event_sender: TimerEventSender,
+    timer_event_sender: TimerEventSender,
     #[allow(dead_code)]
     status_report_sender: Option<AsyncSender<i32>>,
-    pub(crate) shared_header: SharedHeader,
+    shared_header: SharedHeader,
+    clock: Clock,
 }
 
 // In any case, the task is not executed in the Scheduler,
@@ -158,10 +161,14 @@ pub struct Timer {
 impl Timer {
     /// Initialize a timer wheel core.
     pub fn new(timer_event_sender: TimerEventSender, shared_header: SharedHeader) -> Self {
+        let runtime_kind = shared_header.runtime_instance.kind;
+        let clock = Clock::new(runtime_kind);
+
         Timer {
             timer_event_sender,
             status_report_sender: None,
             shared_header,
+            clock,
         }
     }
 
@@ -181,25 +188,29 @@ impl Timer {
             .unwrap_or_else(|e| e)
     }
 
+    /// Time goes on, the clock ticks.
+    pub(crate) async fn lapse(&mut self) {
+        self.clock.tick().await;
+        self.next_position();
+    }
+
     /// Return a future can pool it for Schedule all cycles task.
     pub(crate) async fn async_schedule(&mut self) {
-        // if that overtime , i run it not block
-        let mut second_hand;
-        let mut next_second_hand;
-        let mut timestamp;
+        // if that overtime , run it not block
 
-        let runtime_kind = self.shared_header.runtime_instance.kind;
-        let mut clock = Clock::new(runtime_kind);
+        let mut second_hand = self.second_hand();
+        let mut next_second_hand = second_hand + 1;
+        let mut current_timestamp = timestamp();
 
         loop {
             //TODO: replenish ending single, for stop current jod and thread.
             if !self.shared_header.shared_motivation.load(Acquire) {
                 return;
             }
 
-            second_hand = self.next_position();
-            timestamp = get_timestamp();
-            self.shared_header.global_time.store(timestamp, Release);
+            self.shared_header
+                .global_time
+                .store(current_timestamp, Release);
             let task_ids;
 
             {
@@ -217,7 +228,7 @@ impl Timer {
                 }
             }
 
-            trace!("timestamp: {}, task_ids: {:?}", timestamp, task_ids);
+            trace!("timestamp: {}, task_ids: {:?}", current_timestamp, task_ids);
 
             // Centralize task processing to avoid duplicate lock requests and releases.
             // FIXME: https://github.com/BinChengZhao/delay-timer/issues/29
@@ -234,8 +245,7 @@ impl Timer {
                 }
 
                 if let Some(task) = task_option {
-                    next_second_hand = (second_hand + 1) % DEFAULT_TIMER_SLOT_COUNT;
-                    self.maintain_task(task, timestamp, next_second_hand)
+                    self.maintain_task(task, current_timestamp, next_second_hand)
                         .await
                         .map_err(|e| error!("{}", e))
                         .ok();
@@ -250,10 +260,20 @@ impl Timer {
                 }
             }
 
-            clock.tick().await;
+            self.lapse().await;
+
+            second_hand = self.second_hand();
+            next_second_hand = (second_hand + 1) % DEFAULT_TIMER_SLOT_COUNT;
+            current_timestamp = timestamp();
         }
     }
 
+    /// Access to the second-hand
+    pub(crate) fn second_hand(&self) -> u64 {
+        self.shared_header.second_hand.load(Acquire)
+    }
+
+    /// Send timer-event to event-handle.
     pub(crate) async fn send_timer_event(
         &mut self,
         task_id: u64,
@@ -347,6 +367,11 @@ impl Timer {
             .get_next_exec_timestamp()
             .ok_or_else(|| anyhow!("can't get_next_exec_timestamp in task :{}", task.task_id))?;
 
+        // cylinder_line = 24
+        // slot_seed = 60
+        // when-init: slot_seed+=1 == 61
+        // when-on-slot61-exec: (task_excute_timestamp - timestamp + next_second_hand) % slot_seed == 61
+
         // Time difference + next second hand % DEFAULT_TIMER_SLOT_COUNT
         let step = task_excute_timestamp.checked_sub(timestamp).unwrap_or(1) + next_second_hand;
         let cylinder_line = step / DEFAULT_TIMER_SLOT_COUNT;

src/utils/parse.rs

@@ -63,6 +63,7 @@ pub mod shell_command {
     /// Abstraction of command methods in multiple libraries.
     pub trait CommandUnify<Child: ChildUnify>: Sized {
         /// Constructs a new Command for launching the program at path program.
+        // FIXME:
         fn new<S: AsRef<OsStr>>(program: S) -> Self {
             Self::new(program.as_ref())
         }