chore: fix all tests case.

Author: BinChengZhao

src/entity.rs

@@ -38,15 +38,32 @@ cfg_status_report!(
 pub(crate) type SharedMotivation = Arc<AtomicBool>;
 // Global IdGenerator.
 pub(crate) type SharedIdGenerator = Arc<AsyncMutex<SnowflakeIdGenerator>>;
-// Global sencond hand.
-pub(crate) type SencondHand = Arc<AtomicU64>;
 // Global Timestamp.
 pub(crate) type GlobalTime = Arc<AtomicU64>;
 // Shared task-wheel for operate.
 pub(crate) type SharedTaskWheel = Arc<DashMap<u64, Slot>>;
 // The slot currently used for storing global tasks.
 pub(crate) type SharedTaskFlagMap = Arc<DashMap<u64, TaskMark>>;
 
+/// Global sencond hand.
+#[derive(Debug, Clone, Default)]
+pub(crate) struct SencondHand {
+    pub(crate) inner: Arc<AtomicU64>,
+}
+
+impl SencondHand {
+    pub(crate) fn current_second_hand(&self) -> u64 {
+        self.inner.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn next(&self) -> Result<u64, u64> {
+        self.inner
+            .fetch_update(Ordering::Release, Ordering::Relaxed, |x| {
+                Some((x + 1) % DEFAULT_TIMER_SLOT_COUNT)
+            })
+    }
+}
+
 /// Builds DelayTimer with custom configuration values.
 ///
 /// Methods can be chained in order to set the configuration values. The
@@ -176,7 +193,7 @@ impl Default for SharedHeader {
     fn default() -> Self {
         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 second_hand = SencondHand::default();
         let global_time = Arc::new(AtomicU64::new(timestamp()));
         let shared_motivation = Arc::new(AtomicBool::new(true));
         let runtime_instance = RuntimeInstance::default();
@@ -234,7 +251,7 @@ impl DelayTimerBuilder {
     }
 
     fn assign_task(&mut self, event_handle: EventHandle, shared_header: SharedHeader) {
-        let timer = Timer::new(self.get_timer_event_sender(), shared_header.clone());
+        let timer = Timer::new(self.get_timer_event_sender(), shared_header);
 
         self.run_async_schedule(timer);
 

src/timer/event_handle.rs

@@ -273,17 +273,20 @@ impl EventHandle {
 
     // Add task to wheel_queue  slot
     fn add_task(&mut self, mut task: Box<Task>) -> AnyResult<TaskMark> {
-        let second_hand = self.shared_header.second_hand.load(Acquire);
+        let second_hand = self.shared_header.second_hand.current_second_hand();
 
         let exec_time: u64 = task
             .get_next_exec_timestamp()
             .ok_or_else(|| anyhow!("can't get_next_exec_timestamp in {}", &task.task_id))?;
 
         let timestamp = self.shared_header.global_time.load(Acquire);
+
+        // Put task on next slot.
         let time_seed: u64 = exec_time
             .checked_sub(timestamp)
             .unwrap_or(task.task_id % DEFAULT_TIMER_SLOT_COUNT)
-            + second_hand;
+            + second_hand
+            + 1;
         let slot_seed: u64 = time_seed % DEFAULT_TIMER_SLOT_COUNT;
 
         let cylinder_line = time_seed / DEFAULT_TIMER_SLOT_COUNT;
@@ -357,7 +360,7 @@ impl EventHandle {
         };
         task.clear_cylinder_line();
 
-        let slot_seed = self.shared_header.second_hand.load(Acquire) + 1;
+        let slot_seed = self.shared_header.second_hand.current_second_hand() + 1;
 
         if let Some(mut slot) = self.shared_header.wheel_queue.get_mut(&slot_seed) {
             slot.value_mut().add_task(task);

src/timer/timer_core.rs

@@ -6,7 +6,7 @@ use crate::entity::timestamp;
 use crate::entity::RuntimeKind;
 
 use std::mem::replace;
-use std::sync::atomic::Ordering::{Acquire, Relaxed, Release};
+use std::sync::atomic::Ordering::{Acquire, Release};
 use std::time::Duration;
 use std::time::Instant;
 
@@ -180,12 +180,7 @@ impl Timer {
     /// Offset the current slot by one when reading it,
     /// so event_handle can be easily inserted into subsequent slots.
     pub(crate) fn next_position(&mut self) -> u64 {
-        self.shared_header
-            .second_hand
-            .fetch_update(Release, Relaxed, |x| {
-                Some((x + 1) % DEFAULT_TIMER_SLOT_COUNT)
-            })
-            .unwrap_or_else(|e| e)
+        self.shared_header.second_hand.next().unwrap_or_else(|e| e)
     }
 
     /// Time goes on, the clock ticks.
@@ -270,7 +265,7 @@ impl Timer {
 
     /// Access to the second-hand
     pub(crate) fn second_hand(&self) -> u64 {
-        self.shared_header.second_hand.load(Acquire)
+        self.shared_header.second_hand.current_second_hand()
     }
 
     /// Send timer-event to event-handle.
@@ -440,6 +435,7 @@ mod tests {
         timer
             .shared_header
             .second_hand
+            .inner
             .store(3599, Ordering::SeqCst);
         assert_eq!(timer.next_position(), 3599);
         assert_eq!(timer.next_position(), 0);

tests/simulation.rs

@@ -12,11 +12,12 @@ use std::time::Duration;
 use smol::Timer;
 
 // TODO: Please turn on `--features=full` before test.
-#[tokio::test]
-async fn test_instance_state() -> anyhow::Result<()> {
+#[test]
+fn test_instance_state() -> anyhow::Result<()> {
     let delay_timer = DelayTimer::new();
 
     let body = || async {
+        println!("create_async_fn_body:i'success");
         Timer::after(Duration::from_millis(100)).await;
     };
 
@@ -25,16 +26,19 @@ async fn test_instance_state() -> anyhow::Result<()> {
         .set_task_id(1)
         .set_maximum_parallel_runnable_num(3)
         .spawn_async_routine(body)?;
+
     let task_instance_chain = delay_timer.insert_task(task)?;
 
     // Get the first task instance.
     let instance = task_instance_chain.next_with_wait()?;
 
+
     // The task was still running when the instance was first obtained.
     assert_eq!(instance.get_state(), instance::RUNNING);
 
     // Unsolicited mission cancellation.
     instance.cancel_with_wait()?;
+
     assert_eq!(instance.get_state(), instance::CANCELLED);
 
     // Get the second task instance.
@@ -267,8 +271,8 @@ async fn test_maximum_parallel_runnable_num() -> AnyResult<()> {
     Ok(())
 }
 
-#[tokio::test]
-async fn tests_countdown() -> AnyResult<()> {
+#[test]
+fn tests_countdown() -> AnyResult<()> {
     let delay_timer = DelayTimer::new();
     let share_num = Arc::new(AtomicI32::new(3));
     let share_num_bunshin = share_num.clone();