kestrel_timer/wheel.rs
1use crate::CallbackWrapper;
2use crate::config::{BatchConfig, WheelConfig};
3use crate::task::{
4 TaskCompletion, TaskHandle, TaskId, TaskLocation, TaskTypeWithCompletionNotifier,
5 TimerTaskForWheel, TimerTaskWithCompletionNotifier,
6};
7use deferred_map::DeferredMap;
8use std::time::Duration;
9
10pub struct WheelAdvanceResult {
11 pub id: TaskId,
12 pub callback: Option<CallbackWrapper>,
13}
14
15/// Timing wheel single layer data structure
16///
17/// 时间轮单层数据结构
18struct WheelLayer {
19 /// Slot array, each slot stores a group of timer tasks
20 ///
21 /// 槽数组，每个槽存储一组定时器任务
22 slots: Vec<Vec<TimerTaskForWheel>>,
23
24 /// Current time pointer (tick index)
25 ///
26 /// 当前时间指针（tick 索引）
27 current_tick: u64,
28
29 /// Slot count
30 ///
31 /// 槽数量
32 slot_count: usize,
33
34 /// Duration of each tick
35 ///
36 /// 每个 tick 的持续时间
37 tick_duration: Duration,
38
39 /// Cache: tick duration in milliseconds (u64) - avoid repeated conversion
40 ///
41 /// 缓存：tick 持续时间（毫秒，u64）- 避免重复转换
42 tick_duration_ms: u64,
43
44 /// Cache: slot mask (slot_count - 1) - for fast modulo operation
45 ///
46 /// 缓存：槽掩码（slot_count - 1）- 用于快速取模运算
47 slot_mask: usize,
48}
49
50impl WheelLayer {
51 /// Create a new wheel layer
52 ///
53 /// 创建一个新的时间轮层
54 fn new(slot_count: usize, tick_duration: Duration) -> Self {
55 let mut slots = Vec::with_capacity(slot_count);
56 // Pre-allocate capacity for each slot to reduce subsequent reallocation during push
57 // Most slots typically contain 0-4 tasks, pre-allocate capacity of 4
58 //
59 // 为每个槽预分配容量以减少后续 push 时的重新分配
60 // 大多数槽通常包含 0-4 个任务，预分配容量为 4
61 for _ in 0..slot_count {
62 slots.push(Vec::with_capacity(4));
63 }
64
65 let tick_duration_ms = tick_duration.as_millis() as u64;
66 let slot_mask = slot_count - 1;
67
68 Self {
69 slots,
70 current_tick: 0,
71 slot_count,
72 tick_duration,
73 tick_duration_ms,
74 slot_mask,
75 }
76 }
77
78 /// Calculate the number of ticks corresponding to the delay
79 ///
80 /// 计算延迟对应的 tick 数量
81 fn delay_to_ticks(&self, delay: Duration) -> u64 {
82 let ticks = delay.as_millis() as u64 / self.tick_duration.as_millis() as u64;
83 ticks.max(1) // at least 1 tick (至少 1 个 tick)
84 }
85}
86
87/// Timing wheel data structure (hierarchical mode)
88///
89/// Now uses DeferredMap for O(1) task lookup and generational safety.
90///
91/// 时间轮数据结构（分层模式）
92///
93/// 现在使用 DeferredMap 实现 O(1) 任务查找和代数安全
94pub struct Wheel {
95 /// L0 layer (bottom layer)
96 ///
97 /// L0 层（底层）
98 l0: WheelLayer,
99
100 /// L1 layer (top layer)
101 ///
102 /// L1 层（顶层）
103 l1: WheelLayer,
104
105 /// L1 tick ratio relative to L0 tick
106 ///
107 /// L1 tick 相对于 L0 tick 的比率
108 pub(crate) l1_tick_ratio: u64,
109
110 /// Task index for fast lookup and cancellation using DeferredMap
111 ///
112 /// Keys are TaskIds (u64 from DeferredMap), values are TaskLocations
113 ///
114 /// 任务索引，使用 DeferredMap 实现快速查找和取消
115 ///
116 /// 键是 TaskId（来自 DeferredMap 的 u64），值是 TaskLocation
117 pub(crate) task_index: DeferredMap<TaskLocation>,
118
119 /// Batch processing configuration
120 ///
121 /// 批处理配置
122 batch_config: BatchConfig,
123
124 /// Cache: L0 layer capacity in milliseconds - avoid repeated calculation
125 ///
126 /// 缓存：L0 层容量（毫秒）- 避免重复计算
127 l0_capacity_ms: u64,
128
129 /// Cache: L1 layer capacity in ticks - avoid repeated calculation
130 ///
131 /// 缓存：L1 层容量（tick 数）- 避免重复计算
132 l1_capacity_ticks: u64,
133}
134
135impl Wheel {
136 /// Create new timing wheel
137 ///
138 /// # Parameters
139 /// - `config`: Timing wheel configuration (already validated)
140 /// - `batch_config`: Batch processing configuration
141 ///
142 /// # Notes
143 /// Configuration parameters have been validated in WheelConfig::builder().build(), so this method will not fail.
144 /// Now uses DeferredMap for task indexing with generational safety.
145 ///
146 /// 创建新的时间轮
147 ///
148 /// # 参数
149 /// - `config`: 时间轮配置（已验证）
150 /// - `batch_config`: 批处理配置
151 ///
152 /// # 注意
153 /// 配置参数已在 WheelConfig::builder().build() 中验证，因此此方法不会失败。
154 /// 现在使用 DeferredMap 进行任务索引，具有代数安全特性。
155 pub fn new(config: WheelConfig, batch_config: BatchConfig) -> Self {
156 let l0 = WheelLayer::new(config.l0_slot_count, config.l0_tick_duration);
157 let l1 = WheelLayer::new(config.l1_slot_count, config.l1_tick_duration);
158
159 // Calculate L1 tick ratio relative to L0 tick
160 // 计算 L1 tick 相对于 L0 tick 的比率
161 let l1_tick_ratio = l1.tick_duration_ms / l0.tick_duration_ms;
162
163 // Pre-calculate capacity to avoid repeated calculation in insert
164 // 预计算容量，避免在 insert 中重复计算
165 let l0_capacity_ms = (l0.slot_count as u64) * l0.tick_duration_ms;
166 let l1_capacity_ticks = l1.slot_count as u64;
167
168 // Estimate initial capacity for DeferredMap based on L0 slot count
169 // 根据 L0 槽数量估算 DeferredMap 的初始容量
170 let estimated_capacity = (l0.slot_count / 4).max(64);
171
172 Self {
173 l0,
174 l1,
175 l1_tick_ratio,
176 task_index: DeferredMap::with_capacity(estimated_capacity),
177 batch_config,
178 l0_capacity_ms,
179 l1_capacity_ticks,
180 }
181 }
182
183 /// Get current tick (L0 layer tick)
184 ///
185 /// 获取当前 tick（L0 层 tick）
186 #[allow(dead_code)]
187 pub fn current_tick(&self) -> u64 {
188 self.l0.current_tick
189 }
190
191 /// Get tick duration (L0 layer tick duration)
192 ///
193 /// 获取 tick 持续时间（L0 层 tick 持续时间）
194 #[allow(dead_code)]
195 pub fn tick_duration(&self) -> Duration {
196 self.l0.tick_duration
197 }
198
199 /// Get slot count (L0 layer slot count)
200 ///
201 /// 获取槽数量（L0 层槽数量）
202 #[allow(dead_code)]
203 pub fn slot_count(&self) -> usize {
204 self.l0.slot_count
205 }
206
207 /// Calculate the number of ticks corresponding to the delay (based on L0 layer)
208 ///
209 /// 计算延迟对应的 tick 数量（基于 L0 层）
210 #[allow(dead_code)]
211 pub fn delay_to_ticks(&self, delay: Duration) -> u64 {
212 self.l0.delay_to_ticks(delay)
213 }
214
215 /// Determine which layer the delay should be inserted into
216 ///
217 /// # Returns
218 /// Returns: (layer, ticks, rounds)
219 /// - Layer: 0 = L0, 1 = L1
220 /// - Ticks: number of ticks calculated from current tick
221 /// - Rounds: number of rounds (only used for very long delays in L1 layer)
222 ///
223 /// 确定延迟应该插入到哪一层
224 ///
225 /// # 返回值
226 /// 返回：(层级, ticks, 轮数)
227 /// - 层级：0 = L0, 1 = L1
228 /// - Ticks：从当前 tick 计算的 tick 数量
229 /// - 轮数：轮数（仅用于 L1 层的超长延迟）
230 #[inline(always)]
231 fn determine_layer(&self, delay: Duration) -> (u8, u64, u32) {
232 let delay_ms = delay.as_millis() as u64;
233
234 // Fast path: most tasks are within L0 range (using cached capacity)
235 // 快速路径：大多数任务在 L0 范围内（使用缓存的容量）
236 if delay_ms < self.l0_capacity_ms {
237 let l0_ticks = (delay_ms / self.l0.tick_duration_ms).max(1);
238 return (0, l0_ticks, 0);
239 }
240
241 // Slow path: L1 layer tasks (using cached values)
242 // 慢速路径：L1 层任务（使用缓存的值）
243 let l1_ticks = (delay_ms / self.l1.tick_duration_ms).max(1);
244 let rounds = ((l1_ticks - 1) / self.l1_capacity_ticks) as u32;
245 (1, l1_ticks, rounds)
246 }
247
248 /// Allocate a handle from DeferredMap
249 ///
250 /// 从 DeferredMap 分配一个 handle
251 ///
252 /// # Returns
253 /// A unique handle for later insertion
254 ///
255 /// # 返回值
256 /// 用于后续插入的唯一 handle
257 pub fn allocate_handle(&mut self) -> TaskHandle {
258 TaskHandle::new(self.task_index.allocate_handle())
259 }
260
261 /// Batch allocate handles from DeferredMap
262 ///
263 /// 从 DeferredMap 批量分配 handles
264 ///
265 /// # Parameters
266 /// - `count`: Number of handles to allocate
267 ///
268 /// # Returns
269 /// Vector of unique handles for later batch insertion
270 ///
271 /// # 参数
272 /// - `count`: 要分配的 handle 数量
273 ///
274 /// # 返回值
275 /// 用于后续批量插入的唯一 handles 向量
276 pub fn allocate_handles(&mut self, count: usize) -> Vec<TaskHandle> {
277 let mut handles = Vec::with_capacity(count);
278 for _ in 0..count {
279 handles.push(TaskHandle::new(self.task_index.allocate_handle()));
280 }
281 handles
282 }
283
284 /// Insert timer task
285 ///
286 /// # Parameters
287 /// - `handle`: Handle for the task
288 /// - `task`: Timer task with completion notifier
289 ///
290 /// # Returns
291 /// Unique identifier of the task (TaskId) - now generated from DeferredMap
292 ///
293 /// # Implementation Details
294 /// - Allocates Handle from DeferredMap to generate TaskId with generational safety
295 /// - Automatically calculate the layer and slot where the task should be inserted
296 /// - Hierarchical mode: short delay tasks are inserted into L0, long delay tasks are inserted into L1
297 /// - Use bit operations to optimize slot index calculation
298 /// - Use DeferredMap for O(1) lookup and cancellation with generation checking
299 ///
300 /// 插入定时器任务
301 ///
302 /// # 参数
303 /// - `handle`: 任务的 handle
304 /// - `task`: 带完成通知器的定时器任务
305 ///
306 /// # 返回值
307 /// 任务的唯一标识符（TaskId）- 现在从 DeferredMap 生成
308 ///
309 /// # 实现细节
310 /// - 自动计算任务应该插入的层级和槽位
311 /// - 分层模式：短延迟任务插入 L0，长延迟任务插入 L1
312 /// - 使用位运算优化槽索引计算
313 /// - 使用 DeferredMap 实现 O(1) 查找和取消，带代数检查
314 #[inline]
315 pub fn insert(&mut self, handle: TaskHandle, task: TimerTaskWithCompletionNotifier) {
316 let task_id = handle.task_id();
317
318 let (level, ticks, rounds) = self.determine_layer(task.delay);
319
320 // Use match to reduce branches, and use cached slot mask
321 // 使用 match 减少分支，并使用缓存的槽掩码
322 let (current_tick, slot_mask, slots) = match level {
323 0 => (self.l0.current_tick, self.l0.slot_mask, &mut self.l0.slots),
324 _ => (self.l1.current_tick, self.l1.slot_mask, &mut self.l1.slots),
325 };
326
327 let total_ticks = current_tick + ticks;
328 let slot_index = (total_ticks as usize) & slot_mask;
329
330 // Create task with the assigned TaskId
331 // 使用分配的 TaskId 创建任务
332 let task = TimerTaskForWheel::new_with_id(task_id, task, total_ticks, rounds);
333
334 // Get the index position of the task in Vec (the length before insertion is the index of the new task)
335 // 获取任务在 Vec 中的索引位置（插入前的长度就是新任务的索引）
336 let vec_index = slots[slot_index].len();
337 let location = TaskLocation::new(level, slot_index, vec_index);
338
339 // Insert task into slot
340 // 将任务插入槽中
341 slots[slot_index].push(task);
342
343 // Insert task location into DeferredMap using handle
344 // 使用 handle 将任务位置插入 DeferredMap
345 self.task_index.insert(handle.into_handle(), location);
346 }
347
348 /// Batch insert timer tasks
349 ///
350 /// # Parameters
351 /// - `handles`: List of pre-allocated handles for tasks
352 /// - `tasks`: List of timer tasks with completion notifiers
353 ///
354 /// # Returns
355 /// - `Ok(())` if all tasks are successfully inserted
356 /// - `Err(TimerError::BatchLengthMismatch)` if handles and tasks lengths don't match
357 ///
358 /// # Performance Advantages
359 /// - Reduce repeated boundary checks and capacity adjustments
360 /// - For tasks with the same delay, calculation results can be reused
361 /// - Uses DeferredMap for efficient task indexing with generational safety
362 ///
363 /// 批量插入定时器任务
364 ///
365 /// # 参数
366 /// - `handles`: 任务的预分配 handles 列表
367 /// - `tasks`: 带完成通知器的定时器任务列表
368 ///
369 /// # 返回值
370 /// - `Ok(())` 如果所有任务成功插入
371 /// - `Err(TimerError::BatchLengthMismatch)` 如果 handles 和 tasks 长度不匹配
372 ///
373 /// # 性能优势
374 /// - 减少重复的边界检查和容量调整
375 /// - 对于相同延迟的任务，可以重用计算结果
376 /// - 使用 DeferredMap 实现高效的任务索引，具有代数安全特性
377 #[inline]
378 pub fn insert_batch(
379 &mut self,
380 handles: Vec<TaskHandle>,
381 tasks: Vec<TimerTaskWithCompletionNotifier>,
382 ) -> Result<(), crate::error::TimerError> {
383 // Validate that handles and tasks have the same length
384 // 验证 handles 和 tasks 长度相同
385 if handles.len() != tasks.len() {
386 return Err(crate::error::TimerError::BatchLengthMismatch {
387 handles_len: handles.len(),
388 tasks_len: tasks.len(),
389 });
390 }
391
392 for (handle, task) in handles.into_iter().zip(tasks) {
393 let task_id = handle.task_id();
394
395 let (level, ticks, rounds) = self.determine_layer(task.delay);
396
397 // Use match to reduce branches, and use cached slot mask
398 // 使用 match 减少分支，并使用缓存的槽掩码
399 let (current_tick, slot_mask, slots) = match level {
400 0 => (self.l0.current_tick, self.l0.slot_mask, &mut self.l0.slots),
401 _ => (self.l1.current_tick, self.l1.slot_mask, &mut self.l1.slots),
402 };
403
404 let total_ticks = current_tick + ticks;
405 let slot_index = (total_ticks as usize) & slot_mask;
406
407 // Create task with the assigned TaskId
408 // 使用分配的 TaskId 创建任务
409 let task = TimerTaskForWheel::new_with_id(task_id, task, total_ticks, rounds);
410
411 // Get the index position of the task in Vec
412 // 获取任务在 Vec 中的索引位置
413 let vec_index = slots[slot_index].len();
414 let location = TaskLocation::new(level, slot_index, vec_index);
415
416 // Insert task into slot
417 // 将任务插入槽中
418 slots[slot_index].push(task);
419
420 // Insert task location into DeferredMap using handle
421 // 使用 handle 将任务位置插入 DeferredMap
422 self.task_index.insert(handle.into_handle(), location);
423 }
424
425 Ok(())
426 }
427
428 /// Cancel timer task
429 ///
430 /// # Parameters
431 /// - `task_id`: Task ID
432 ///
433 /// # Returns
434 /// Returns true if the task exists and is successfully cancelled, otherwise returns false
435 ///
436 /// Now uses DeferredMap for safe task removal with generational checking
437 ///
438 /// 取消定时器任务
439 ///
440 /// # 参数
441 /// - `task_id`: 任务 ID
442 ///
443 /// # 返回值
444 /// 如果任务存在且成功取消则返回 true，否则返回 false
445 ///
446 /// 现在使用 DeferredMap 实现安全的任务移除，带代数检查
447 #[inline]
448 pub fn cancel(&mut self, task_id: TaskId) -> bool {
449 // Remove task location from DeferredMap using TaskId as key
450 // 使用 TaskId 作为 key 从 DeferredMap 中移除任务位置
451 let location = match self.task_index.remove(task_id.key()) {
452 Some(loc) => loc,
453 None => return false, // Task not found or already removed (generation mismatch)
454 };
455
456 // Use match to get slot reference, reduce branches
457 // 使用 match 获取槽引用，减少分支
458 let slot = match location.level {
459 0 => &mut self.l0.slots[location.slot_index],
460 _ => &mut self.l1.slots[location.slot_index],
461 };
462
463 // Boundary check and ID verification
464 // 边界检查和 ID 验证
465 if location.vec_index >= slot.len() || slot[location.vec_index].get_id() != task_id {
466 // Index inconsistent - this shouldn't happen with DeferredMap, but handle it anyway
467 // 索引不一致 - DeferredMap 不应该出现这种情况，但还是处理一下
468 return false;
469 }
470
471 // Use swap_remove to remove task, record swapped task ID
472 // 使用 swap_remove 移除任务，记录被交换的任务 ID
473 let removed_task = slot.swap_remove(location.vec_index);
474
475 // Ensure the correct task was removed
476 // 确保移除了正确的任务
477 debug_assert_eq!(removed_task.get_id(), task_id);
478
479 match removed_task.into_task_type() {
480 TaskTypeWithCompletionNotifier::OneShot {
481 completion_notifier,
482 } => {
483 // Use Notify + AtomicU8 for zero-allocation notification
484 // 使用 Notify + AtomicU8 实现零分配通知
485 completion_notifier.notify(crate::task::TaskCompletion::Cancelled);
486 }
487 TaskTypeWithCompletionNotifier::Periodic {
488 completion_notifier,
489 ..
490 } => {
491 // Use flume for high-performance periodic notification
492 // 使用 flume 实现高性能周期通知
493 let _ = completion_notifier.0.try_send(TaskCompletion::Cancelled);
494 }
495 }
496
497 // If a swap occurred (vec_index is not the last element)
498 // 如果发生了交换（vec_index 不是最后一个元素）
499 if location.vec_index < slot.len() {
500 let swapped_task_id = slot[location.vec_index].get_id();
501 // Update swapped element's index in one go, using DeferredMap's get_mut
502 // 一次性更新被交换元素的索引，使用 DeferredMap 的 get_mut
503 if let Some(swapped_location) = self.task_index.get_mut(swapped_task_id.key()) {
504 swapped_location.vec_index = location.vec_index;
505 }
506 }
507
508 true
509 }
510
511 /// Batch cancel timer tasks
512 ///
513 /// # Parameters
514 /// - `task_ids`: List of task IDs to cancel
515 ///
516 /// # Returns
517 /// Number of successfully cancelled tasks
518 ///
519 /// # Performance Advantages
520 /// - Reduce repeated HashMap lookup overhead
521 /// - Multiple cancellation operations on the same slot can be batch processed
522 /// - Use unstable sort to improve performance
523 /// - Small batch optimization: skip sorting based on configuration threshold, process directly
524 ///
525 /// 批量取消定时器任务
526 ///
527 /// # 参数
528 /// - `task_ids`: 要取消的任务 ID 列表
529 ///
530 /// # 返回值
531 /// 成功取消的任务数量
532 ///
533 /// # 性能优势
534 /// - 减少重复的 HashMap 查找开销
535 /// - 同一槽位的多个取消操作可以批量处理
536 /// - 使用不稳定排序提高性能
537 /// - 小批量优化：根据配置阈值跳过排序，直接处理
538 #[inline]
539 pub fn cancel_batch(&mut self, task_ids: &[TaskId]) -> usize {
540 let mut cancelled_count = 0;
541
542 // Small batch optimization: cancel one by one to avoid grouping and sorting overhead
543 // 小批量优化：逐个取消以避免分组和排序开销
544 if task_ids.len() <= self.batch_config.small_batch_threshold {
545 for &task_id in task_ids {
546 if self.cancel(task_id) {
547 cancelled_count += 1;
548 }
549 }
550 return cancelled_count;
551 }
552
553 // Group by layer and slot to optimize batch cancellation
554 // Use SmallVec to avoid heap allocation in most cases
555 // 按层级和槽位分组以优化批量取消
556 // 使用 SmallVec 避免在大多数情况下进行堆分配
557 let l0_slot_count = self.l0.slot_count;
558 let l1_slot_count = self.l1.slot_count;
559
560 let mut l0_tasks_by_slot: Vec<Vec<(TaskId, usize)>> = vec![Vec::new(); l0_slot_count];
561 let mut l1_tasks_by_slot: Vec<Vec<(TaskId, usize)>> = vec![Vec::new(); l1_slot_count];
562
563 // Collect information of tasks to be cancelled
564 // 收集要取消的任务信息
565 for &task_id in task_ids {
566 // Use DeferredMap's get with TaskId key
567 // 使用 DeferredMap 的 get，传入 TaskId key
568 if let Some(location) = self.task_index.get(task_id.key()) {
569 if location.level == 0 {
570 l0_tasks_by_slot[location.slot_index].push((task_id, location.vec_index));
571 } else {
572 l1_tasks_by_slot[location.slot_index].push((task_id, location.vec_index));
573 }
574 }
575 }
576
577 // Process L0 layer cancellation
578 // 处理 L0 层取消
579 for (slot_index, tasks) in l0_tasks_by_slot.iter_mut().enumerate() {
580 if tasks.is_empty() {
581 continue;
582 }
583
584 // Sort by vec_index in descending order, delete from back to front to avoid index invalidation
585 // 按 vec_index 降序排序，从后向前删除以避免索引失效
586 tasks.sort_unstable_by_key(|b| std::cmp::Reverse(b.1));
587
588 let slot = &mut self.l0.slots[slot_index];
589
590 for &(task_id, vec_index) in tasks.iter() {
591 if vec_index < slot.len() && slot[vec_index].get_id() == task_id {
592 let removed_task = slot.swap_remove(vec_index);
593 match removed_task.into_task_type() {
594 TaskTypeWithCompletionNotifier::OneShot {
595 completion_notifier,
596 } => {
597 // Use Notify + AtomicU8 for zero-allocation notification
598 // 使用 Notify + AtomicU8 实现零分配通知
599 completion_notifier.notify(crate::task::TaskCompletion::Cancelled);
600 }
601 TaskTypeWithCompletionNotifier::Periodic {
602 completion_notifier,
603 ..
604 } => {
605 // Use flume for high-performance periodic notification
606 // 使用 flume 实现高性能周期通知
607 let _ = completion_notifier.0.try_send(TaskCompletion::Cancelled);
608 }
609 }
610
611 if vec_index < slot.len() {
612 let swapped_task_id = slot[vec_index].get_id();
613 // Use DeferredMap's get_mut with TaskId key
614 // 使用 DeferredMap 的 get_mut，传入 TaskId key
615 if let Some(swapped_location) =
616 self.task_index.get_mut(swapped_task_id.key())
617 {
618 swapped_location.vec_index = vec_index;
619 }
620 }
621
622 // Use DeferredMap's remove with TaskId key
623 // 使用 DeferredMap 的 remove，传入 TaskId key
624 self.task_index.remove(task_id.key());
625 cancelled_count += 1;
626 }
627 }
628 }
629
630 // Process L1 layer cancellation
631 // 处理 L1 层取消
632 for (slot_index, tasks) in l1_tasks_by_slot.iter_mut().enumerate() {
633 if tasks.is_empty() {
634 continue;
635 }
636
637 tasks.sort_unstable_by_key(|b| std::cmp::Reverse(b.1));
638
639 let slot = &mut self.l1.slots[slot_index];
640
641 for &(task_id, vec_index) in tasks.iter() {
642 if vec_index < slot.len() && slot[vec_index].get_id() == task_id {
643 let removed_task = slot.swap_remove(vec_index);
644
645 match removed_task.into_task_type() {
646 TaskTypeWithCompletionNotifier::OneShot {
647 completion_notifier,
648 } => {
649 // Use Notify + AtomicU8 for zero-allocation notification
650 // 使用 Notify + AtomicU8 实现零分配通知
651 completion_notifier.notify(crate::task::TaskCompletion::Cancelled);
652 }
653 TaskTypeWithCompletionNotifier::Periodic {
654 completion_notifier,
655 ..
656 } => {
657 // Use flume for high-performance periodic notification
658 // 使用 flume 实现高性能周期通知
659 let _ = completion_notifier.0.try_send(TaskCompletion::Cancelled);
660 }
661 }
662
663 if vec_index < slot.len() {
664 let swapped_task_id = slot[vec_index].get_id();
665 // Use DeferredMap's get_mut with TaskId key
666 // 使用 DeferredMap 的 get_mut，传入 TaskId key
667 if let Some(swapped_location) =
668 self.task_index.get_mut(swapped_task_id.key())
669 {
670 swapped_location.vec_index = vec_index;
671 }
672 }
673
674 // Use DeferredMap's remove with TaskId key
675 // 使用 DeferredMap 的 remove，传入 TaskId key
676 self.task_index.remove(task_id.key());
677 cancelled_count += 1;
678 }
679 }
680 }
681
682 cancelled_count
683 }
684
685 /// Reinsert periodic task with the same TaskId
686 ///
687 /// Used to automatically reschedule periodic tasks after they expire
688 ///
689 /// TaskId remains unchanged, only location is updated in DeferredMap
690 ///
691 /// 重新插入周期性任务（保持相同的 TaskId）
692 ///
693 /// 用于在周期性任务过期后自动重新调度
694 ///
695 /// TaskId 保持不变，只更新 DeferredMap 中的位置
696 fn reinsert_periodic_task(&mut self, task_id: TaskId, task: TimerTaskWithCompletionNotifier) {
697 // Determine which layer the interval should be inserted into
698 // This method is only called by periodic tasks, so the interval is guaranteed to be Some.
699 // 确定间隔应该插入到哪一层
700 // 该方法只能由周期性任务调用，所以间隔是 guaranteed 应当保证为 Some.
701 let (level, ticks, rounds) = self.determine_layer(task.get_interval().unwrap());
702
703 // Use match to reduce branches, and use cached slot mask
704 // 使用 match 减少分支，并使用缓存的槽掩码
705 let (current_tick, slot_mask, slots) = match level {
706 0 => (self.l0.current_tick, self.l0.slot_mask, &mut self.l0.slots),
707 _ => (self.l1.current_tick, self.l1.slot_mask, &mut self.l1.slots),
708 };
709
710 let total_ticks = current_tick + ticks;
711 let slot_index = (total_ticks as usize) & slot_mask;
712
713 // Get the index position of the task in Vec
714 // 获取任务在 Vec 中的索引位置
715 let vec_index = slots[slot_index].len();
716 let new_location = TaskLocation::new(level, slot_index, vec_index);
717
718 // Insert task into slot
719 // 将任务插入槽中
720 slots[slot_index].push(TimerTaskForWheel::new_with_id(
721 task_id,
722 task,
723 total_ticks,
724 rounds,
725 ));
726
727 // Update task location in DeferredMap (doesn't remove, just updates)
728 // 更新 DeferredMap 中的任务位置（不删除，仅更新）
729 if let Some(location) = self.task_index.get_mut(task_id.key()) {
730 *location = new_location;
731 }
732 }
733
734 /// Advance the timing wheel by one tick, return all expired tasks
735 ///
736 /// # Returns
737 /// List of expired tasks
738 ///
739 /// # Implementation Details
740 /// - L0 layer advances 1 tick each time (no rounds check)
741 /// - L1 layer advances once every (L1_tick / L0_tick) times
742 /// - L1 expired tasks are batch demoted to L0
743 ///
744 /// 推进时间轮一个 tick，返回所有过期的任务
745 ///
746 /// # 返回值
747 /// 过期任务列表
748 ///
749 /// # 实现细节
750 /// - L0 层每次推进 1 个 tick（无轮数检查）
751 /// - L1 层每 (L1_tick / L0_tick) 次推进一次
752 /// - L1 层过期任务批量降级到 L0
753 pub fn advance(&mut self) -> Vec<WheelAdvanceResult> {
754 // Advance L0 layer
755 // 推进 L0 层
756 self.l0.current_tick += 1;
757
758 let mut expired_tasks = Vec::new();
759
760 // Process L0 layer expired tasks
761 // Distinguish between one-shot and periodic tasks
762 // 处理 L0 层过期任务
763 // 区分一次性任务和周期性任务
764 let l0_slot_index = (self.l0.current_tick as usize) & self.l0.slot_mask;
765
766 // Collect periodic tasks to reinsert
767 // 收集需要重新插入的周期性任务
768 let mut periodic_tasks_to_reinsert = Vec::new();
769
770 {
771 let l0_slot = &mut self.l0.slots[l0_slot_index];
772
773 // Process all tasks in the current slot by always removing from index 0
774 // This avoids index tracking issues with swap_remove
775 // 通过始终从索引 0 移除来处理当前槽中的所有任务
776 // 这避免了 swap_remove 的索引跟踪问题
777 while !l0_slot.is_empty() {
778 // Always remove from index 0
779 // 始终从索引 0 移除
780 let task_with_notifier = l0_slot.swap_remove(0);
781 let task_id = task_with_notifier.get_id();
782
783 // Determine if this is a periodic task (check before consuming)
784 // 判断是否为周期性任务（在消耗前检查）
785 let is_periodic = matches!(
786 task_with_notifier.task.task_type,
787 TaskTypeWithCompletionNotifier::Periodic { .. }
788 );
789
790 // For one-shot tasks, remove from DeferredMap index
791 // For periodic tasks, keep in index (will update location in reinsert)
792 // 对于一次性任务，从 DeferredMap 索引中移除
793 // 对于周期性任务，保留在索引中（重新插入时会更新位置）
794 if !is_periodic {
795 self.task_index.remove(task_id.key());
796 }
797
798 // Update swapped element's index (the element that was moved to index 0)
799 // 更新被交换元素的索引（被移动到索引 0 的元素）
800 if !l0_slot.is_empty() {
801 let swapped_task_id = l0_slot[0].get_id();
802 // Use DeferredMap's get_mut with TaskId key
803 // 使用 DeferredMap 的 get_mut，传入 TaskId key
804 if let Some(swapped_location) = self.task_index.get_mut(swapped_task_id.key()) {
805 swapped_location.vec_index = 0;
806 }
807 }
808
809 let TimerTaskForWheel { task_id, task, .. } = task_with_notifier;
810
811 match &task.task_type {
812 TaskTypeWithCompletionNotifier::Periodic {
813 completion_notifier,
814 ..
815 } => {
816 let _ = completion_notifier.0.try_send(TaskCompletion::Called);
817
818 periodic_tasks_to_reinsert.push((
819 task_id,
820 TimerTaskWithCompletionNotifier {
821 task_type: task.task_type,
822 delay: task.delay,
823 callback: task.callback.clone(),
824 },
825 ));
826 }
827 TaskTypeWithCompletionNotifier::OneShot {
828 completion_notifier,
829 } => {
830 completion_notifier.notify(crate::task::TaskCompletion::Called);
831 }
832 }
833
834 expired_tasks.push(WheelAdvanceResult {
835 id: task_id,
836 callback: task.callback,
837 });
838 }
839 }
840
841 // Reinsert periodic tasks for next interval
842 // 重新插入周期性任务到下一个周期
843 for (task_id, task) in periodic_tasks_to_reinsert {
844 self.reinsert_periodic_task(task_id, task);
845 }
846
847 // Process L1 layer
848 // Check if L1 layer needs to be advanced
849 // 处理 L1 层
850 // 检查是否需要推进 L1 层
851 if self.l0.current_tick.is_multiple_of(self.l1_tick_ratio) {
852 self.l1.current_tick += 1;
853 let l1_slot_index = (self.l1.current_tick as usize) & self.l1.slot_mask;
854 let l1_slot = &mut self.l1.slots[l1_slot_index];
855
856 // Collect L1 layer expired tasks
857 // 收集 L1 层过期任务
858 let mut tasks_to_demote = Vec::new();
859 let mut i = 0;
860 while i < l1_slot.len() {
861 let task = &mut l1_slot[i];
862
863 if task.rounds > 0 {
864 // Still has rounds, decrease rounds and keep
865 // 还有轮数，减少轮数并保留
866 task.rounds -= 1;
867 // Use DeferredMap's get_mut with TaskId key
868 // 使用 DeferredMap 的 get_mut，传入 TaskId key
869 if let Some(location) = self.task_index.get_mut(task.get_id().key()) {
870 location.vec_index = i;
871 }
872 i += 1;
873 } else {
874 // rounds = 0, need to demote to L0
875 // Don't remove from task_index - will update location in demote_tasks
876 // rounds = 0，需要降级到 L0
877 // 不从 task_index 中移除 - 在 demote_tasks 中会更新位置
878 let task_to_demote = l1_slot.swap_remove(i);
879
880 if i < l1_slot.len() {
881 let swapped_task_id = l1_slot[i].get_id();
882 // Use DeferredMap's get_mut with TaskId key
883 // 使用 DeferredMap 的 get_mut，传入 TaskId key
884 if let Some(swapped_location) =
885 self.task_index.get_mut(swapped_task_id.key())
886 {
887 swapped_location.vec_index = i;
888 }
889 }
890
891 tasks_to_demote.push(task_to_demote);
892 }
893 }
894
895 // Demote tasks to L0
896 // 将任务降级到 L0
897 self.demote_tasks(tasks_to_demote);
898 }
899
900 expired_tasks
901 }
902
903 /// Demote tasks from L1 to L0
904 ///
905 /// Recalculate and insert L1 expired tasks into L0 layer
906 ///
907 /// Updates task location in DeferredMap without removing/reinserting
908 ///
909 /// 将任务从 L1 降级到 L0
910 ///
911 /// 重新计算并将 L1 过期任务插入到 L0 层
912 ///
913 /// 更新 DeferredMap 中的任务位置而不删除/重新插入
914 fn demote_tasks(&mut self, tasks: Vec<TimerTaskForWheel>) {
915 for task in tasks {
916 // Calculate the remaining delay of the task in L0 layer
917 // The task's deadline_tick is based on L1 tick, needs to be converted to L0 tick
918 // 计算任务在 L0 层的剩余延迟
919 // 任务的 deadline_tick 基于 L1 tick，需要转换为 L0 tick
920 let l1_tick_ratio = self.l1_tick_ratio;
921
922 // Calculate the original expiration time of the task (L1 tick)
923 // 计算任务的原始过期时间（L1 tick）
924 let l1_deadline = task.deadline_tick;
925
926 // Convert to L0 tick expiration time
927 // 转换为 L0 tick 过期时间
928 let l0_deadline_tick = l1_deadline * l1_tick_ratio;
929 let l0_current_tick = self.l0.current_tick;
930
931 // Calculate remaining L0 ticks
932 // 计算剩余 L0 ticks
933 let remaining_l0_ticks = if l0_deadline_tick > l0_current_tick {
934 l0_deadline_tick - l0_current_tick
935 } else {
936 1 // At least trigger in next tick (至少在下一个 tick 触发)
937 };
938
939 // Calculate L0 slot index
940 // 计算 L0 槽索引
941 let target_l0_tick = l0_current_tick + remaining_l0_ticks;
942 let l0_slot_index = (target_l0_tick as usize) & self.l0.slot_mask;
943
944 let task_id = task.get_id();
945 let vec_index = self.l0.slots[l0_slot_index].len();
946 let new_location = TaskLocation::new(0, l0_slot_index, vec_index);
947
948 // Insert into L0 layer
949 // 插入到 L0 层
950 self.l0.slots[l0_slot_index].push(task);
951
952 // Update task location in DeferredMap (task already exists in index)
953 // 更新 DeferredMap 中的任务位置（任务已在索引中）
954 if let Some(location) = self.task_index.get_mut(task_id.key()) {
955 *location = new_location;
956 }
957 }
958 }
959
960 /// Check if the timing wheel is empty
961 ///
962 /// 检查时间轮是否为空
963 #[allow(dead_code)]
964 pub fn is_empty(&self) -> bool {
965 self.task_index.is_empty()
966 }
967
968 /// Postpone timer task (keep original TaskId)
969 ///
970 /// # Parameters
971 /// - `task_id`: Task ID to postpone
972 /// - `new_delay`: New delay time (recalculated from current tick, not continuing from original delay time)
973 /// - `new_callback`: New callback function (if None, keep original callback)
974 ///
975 /// # Returns
976 /// Returns true if the task exists and is successfully postponed, otherwise returns false
977 ///
978 /// # Implementation Details
979 /// - Remove task from original layer/slot, keep its completion_notifier (will not trigger cancellation notification)
980 /// - Update delay time and callback function (if provided)
981 /// - Recalculate target layer, slot, and rounds based on new_delay
982 /// - Cross-layer migration may occur (L0 <-> L1)
983 /// - Re-insert to new position using original TaskId
984 /// - Keep consistent with external held TaskId reference
985 ///
986 /// 延期定时器任务（保留原始 TaskId）
987 ///
988 /// # 参数
989 /// - `task_id`: 要延期的任务 ID
990 /// - `new_delay`: 新延迟时间（从当前 tick 重新计算，而非从原延迟时间继续）
991 /// - `new_callback`: 新回调函数（如果为 None，则保留原回调）
992 ///
993 /// # 返回值
994 /// 如果任务存在且成功延期则返回 true，否则返回 false
995 ///
996 /// # 实现细节
997 /// - 从原层级/槽位移除任务，保留其 completion_notifier（不会触发取消通知）
998 /// - 更新延迟时间和回调函数（如果提供）
999 /// - 根据 new_delay 重新计算目标层级、槽位和轮数
1000 /// - 可能发生跨层迁移（L0 <-> L1）
1001 /// - 使用原始 TaskId 重新插入到新位置
1002 /// - 与外部持有的 TaskId 引用保持一致
1003 #[inline]
1004 pub fn postpone(
1005 &mut self,
1006 task_id: TaskId,
1007 new_delay: Duration,
1008 new_callback: Option<crate::task::CallbackWrapper>,
1009 ) -> bool {
1010 // Step 1: Get task location from DeferredMap (don't remove)
1011 // 步骤 1: 从 DeferredMap 获取任务位置（不删除）
1012 let old_location = match self.task_index.get(task_id.key()) {
1013 Some(loc) => *loc,
1014 None => return false,
1015 };
1016
1017 // Use match to get slot reference
1018 // 使用 match 获取槽引用
1019 let slot = match old_location.level {
1020 0 => &mut self.l0.slots[old_location.slot_index],
1021 _ => &mut self.l1.slots[old_location.slot_index],
1022 };
1023
1024 // Verify task is still at expected position
1025 // 验证任务仍在预期位置
1026 if old_location.vec_index >= slot.len() || slot[old_location.vec_index].get_id() != task_id
1027 {
1028 // Index inconsistent, return failure
1029 // 索引不一致，返回失败
1030 return false;
1031 }
1032
1033 // Use swap_remove to remove task from slot
1034 // 使用 swap_remove 从槽中移除任务
1035 let mut task = slot.swap_remove(old_location.vec_index);
1036
1037 // Update swapped element's index (if a swap occurred)
1038 // 更新被交换元素的索引（如果发生了交换）
1039 if old_location.vec_index < slot.len() {
1040 let swapped_task_id = slot[old_location.vec_index].get_id();
1041 // Use DeferredMap's get_mut with TaskId key
1042 // 使用 DeferredMap 的 get_mut，传入 TaskId key
1043 if let Some(swapped_location) = self.task_index.get_mut(swapped_task_id.key()) {
1044 swapped_location.vec_index = old_location.vec_index;
1045 }
1046 }
1047
1048 // Step 2: Update task's delay and callback
1049 // 步骤 2: 更新任务的延迟和回调
1050 task.update_delay(new_delay);
1051 if let Some(callback) = new_callback {
1052 task.update_callback(callback);
1053 }
1054
1055 // Step 3: Recalculate layer, slot, and rounds based on new delay
1056 // 步骤 3: 根据新延迟重新计算层级、槽位和轮数
1057 let (new_level, ticks, new_rounds) = self.determine_layer(new_delay);
1058
1059 // Use match to reduce branches, and use cached slot mask
1060 // 使用 match 减少分支，并使用缓存的槽掩码
1061 let (current_tick, slot_mask, slots) = match new_level {
1062 0 => (self.l0.current_tick, self.l0.slot_mask, &mut self.l0.slots),
1063 _ => (self.l1.current_tick, self.l1.slot_mask, &mut self.l1.slots),
1064 };
1065
1066 let total_ticks = current_tick + ticks;
1067 let new_slot_index = (total_ticks as usize) & slot_mask;
1068
1069 // Update task's timing wheel parameters
1070 // 更新任务的时间轮参数
1071 task.deadline_tick = total_ticks;
1072 task.rounds = new_rounds;
1073
1074 // Step 4: Re-insert task to new layer/slot
1075 // 步骤 4: 将任务重新插入到新层级/槽位
1076 let new_vec_index = slots[new_slot_index].len();
1077 let new_location = TaskLocation::new(new_level, new_slot_index, new_vec_index);
1078
1079 slots[new_slot_index].push(task);
1080
1081 // Update task location in DeferredMap (task already exists in index)
1082 // 更新 DeferredMap 中的任务位置（任务已在索引中）
1083 if let Some(location) = self.task_index.get_mut(task_id.key()) {
1084 *location = new_location;
1085 }
1086
1087 true
1088 }
1089
1090 /// Batch postpone timer tasks
1091 ///
1092 /// # Parameters
1093 /// - `updates`: List of tuples of (task ID, new delay)
1094 ///
1095 /// # Returns
1096 /// Number of successfully postponed tasks
1097 ///
1098 /// # Performance Advantages
1099 /// - Batch processing reduces function call overhead
1100 /// - All delays are recalculated from current_tick at call time
1101 ///
1102 /// # Notes
1103 /// - If a task ID does not exist, that task will be skipped without affecting other tasks' postponement
1104 ///
1105 /// 批量延期定时器任务
1106 ///
1107 /// # 参数
1108 /// - `updates`: (任务 ID, 新延迟) 元组列表
1109 ///
1110 /// # 返回值
1111 /// 成功延期的任务数量
1112 ///
1113 /// # 性能优势
1114 /// - 批处理减少函数调用开销
1115 /// - 所有延迟在调用时从 current_tick 重新计算
1116 ///
1117 /// # 注意
1118 /// - 如果任务 ID 不存在，该任务将被跳过，不影响其他任务的延期
1119 #[inline]
1120 pub fn postpone_batch(&mut self, updates: Vec<(TaskId, Duration)>) -> usize {
1121 let mut postponed_count = 0;
1122
1123 for (task_id, new_delay) in updates {
1124 if self.postpone(task_id, new_delay, None) {
1125 postponed_count += 1;
1126 }
1127 }
1128
1129 postponed_count
1130 }
1131
1132 /// Batch postpone timer tasks (replace callbacks)
1133 ///
1134 /// # Parameters
1135 /// - `updates`: List of tuples of (task ID, new delay, new callback)
1136 ///
1137 /// # Returns
1138 /// Number of successfully postponed tasks
1139 ///
1140 /// 批量延期定时器任务（替换回调）
1141 ///
1142 /// # 参数
1143 /// - `updates`: (任务 ID, 新延迟, 新回调) 元组列表
1144 ///
1145 /// # 返回值
1146 /// 成功延期的任务数量
1147 pub fn postpone_batch_with_callbacks(
1148 &mut self,
1149 updates: Vec<(TaskId, Duration, Option<crate::task::CallbackWrapper>)>,
1150 ) -> usize {
1151 let mut postponed_count = 0;
1152
1153 for (task_id, new_delay, new_callback) in updates {
1154 if self.postpone(task_id, new_delay, new_callback) {
1155 postponed_count += 1;
1156 }
1157 }
1158
1159 postponed_count
1160 }
1161}
1162
1163#[cfg(test)]
1164mod tests {
1165 use super::*;
1166 use crate::task::{CallbackWrapper, TimerTask, TimerTaskWithCompletionNotifier};
1167
1168 #[test]
1169 fn test_wheel_creation() {
1170 let wheel = Wheel::new(WheelConfig::default(), BatchConfig::default());
1171 assert_eq!(wheel.slot_count(), 512);
1172 assert_eq!(wheel.current_tick(), 0);
1173 assert!(wheel.is_empty());
1174 }
1175
1176 #[test]
1177 fn test_hierarchical_wheel_creation() {
1178 let config = WheelConfig::default();
1179
1180 let wheel = Wheel::new(config, BatchConfig::default());
1181 assert_eq!(wheel.slot_count(), 512); // L0 slot count (L0 层槽数量)
1182 assert_eq!(wheel.current_tick(), 0);
1183 assert!(wheel.is_empty());
1184 // L1 layer always exists in hierarchical mode (L1 层在分层模式下始终存在)
1185 assert_eq!(wheel.l1.slot_count, 64);
1186 assert_eq!(wheel.l1_tick_ratio, 100); // 1000ms / 10ms = 100 ticks (1000毫秒 / 10毫秒 = 100个tick)
1187 }
1188
1189 #[test]
1190 fn test_hierarchical_config_validation() {
1191 // L1 tick must be an integer multiple of L0 tick (L1 tick 必须是 L0 tick 的整数倍)
1192 let result = WheelConfig::builder()
1193 .l0_tick_duration(Duration::from_millis(10))
1194 .l0_slot_count(512)
1195 .l1_tick_duration(Duration::from_millis(15)) // Not an integer multiple (不是整数倍)
1196 .l1_slot_count(64)
1197 .build();
1198
1199 assert!(result.is_err());
1200
1201 // Correct configuration (正确的配置)
1202 let result = WheelConfig::builder()
1203 .l0_tick_duration(Duration::from_millis(10))
1204 .l0_slot_count(512)
1205 .l1_tick_duration(Duration::from_secs(1)) // 1000ms / 10ms = 100 ticks (1000毫秒 / 10毫秒 = 100个tick)
1206 .l1_slot_count(64)
1207 .build();
1208
1209 assert!(result.is_ok());
1210 }
1211
1212 #[test]
1213 fn test_layer_determination() {
1214 let config = WheelConfig::default();
1215
1216 let wheel = Wheel::new(config, BatchConfig::default());
1217
1218 // Short delay should enter L0 layer (短延迟应该进入 L0 层)
1219 // L0: 512 slots * 10ms = 5120ms
1220 let (level, _, rounds) = wheel.determine_layer(Duration::from_millis(100));
1221 assert_eq!(level, 0);
1222 assert_eq!(rounds, 0);
1223
1224 // Long delay should enter L1 layer
1225 // 超过 L0 范围（>5120ms） (超过 L0 范围 (>5120毫秒))
1226 let (level, _, rounds) = wheel.determine_layer(Duration::from_secs(10));
1227 assert_eq!(level, 1);
1228 assert_eq!(rounds, 0);
1229
1230 // Long delay should enter L1 layer and have rounds
1231 // L1: 64 slots * 1000ms = 64000ms (L1: 64个槽 * 1000毫秒 = 64000毫秒)
1232 let (level, _, rounds) = wheel.determine_layer(Duration::from_secs(120));
1233 assert_eq!(level, 1);
1234 assert!(rounds > 0);
1235 }
1236
1237 #[test]
1238 fn test_hierarchical_insert_and_advance() {
1239 let config = WheelConfig::default();
1240
1241 let mut wheel = Wheel::new(config, BatchConfig::default());
1242
1243 // Insert short delay task into L0 (插入短延迟任务到 L0)
1244 let callback = CallbackWrapper::new(|| async {});
1245 let task = TimerTask::new_oneshot(Duration::from_millis(100), Some(callback));
1246 let (task_with_notifier, _completion_rx) =
1247 TimerTaskWithCompletionNotifier::from_timer_task(task);
1248 let handle = wheel.allocate_handle();
1249 let task_id = handle.task_id();
1250 wheel.insert(handle, task_with_notifier);
1251
1252 // Verify task is in L0 layer (验证任务在 L0 层)
1253 let location = wheel.task_index.get(task_id.key()).unwrap();
1254 assert_eq!(location.level, 0);
1255
1256 // Advance 10 ticks (100ms) (前进 10 个 tick (100毫秒))
1257 for _ in 0..10 {
1258 let expired = wheel.advance();
1259 if !expired.is_empty() {
1260 assert_eq!(expired.len(), 1);
1261 assert_eq!(expired[0].id, task_id);
1262 return;
1263 }
1264 }
1265 panic!("Task should have expired");
1266 }
1267
1268 #[test]
1269 fn test_cross_layer_cancel() {
1270 let config = WheelConfig::default();
1271
1272 let mut wheel = Wheel::new(config, BatchConfig::default());
1273
1274 // Insert L0 task (插入 L0 任务)
1275 let callback1 = CallbackWrapper::new(|| async {});
1276 let task1 = TimerTask::new_oneshot(Duration::from_millis(100), Some(callback1));
1277 let (task_with_notifier1, _rx1) = TimerTaskWithCompletionNotifier::from_timer_task(task1);
1278 let handle1 = wheel.allocate_handle();
1279 let task_id1 = handle1.task_id();
1280 wheel.insert(handle1, task_with_notifier1);
1281
1282 // Insert L1 task (插入 L1 任务)
1283 let callback2 = CallbackWrapper::new(|| async {});
1284 let task2 = TimerTask::new_oneshot(Duration::from_secs(10), Some(callback2));
1285 let (task_with_notifier2, _rx2) = TimerTaskWithCompletionNotifier::from_timer_task(task2);
1286 let handle2 = wheel.allocate_handle();
1287 let task_id2 = handle2.task_id();
1288 wheel.insert(handle2, task_with_notifier2);
1289
1290 // Verify levels (验证层级)
1291 assert_eq!(wheel.task_index.get(task_id1.key()).unwrap().level, 0);
1292 assert_eq!(wheel.task_index.get(task_id2.key()).unwrap().level, 1);
1293
1294 // Cancel L0 task (取消 L0 任务)
1295 assert!(wheel.cancel(task_id1));
1296 assert!(wheel.task_index.get(task_id1.key()).is_none());
1297
1298 // Cancel L1 task (取消 L1 任务)
1299 assert!(wheel.cancel(task_id2));
1300 assert!(wheel.task_index.get(task_id2.key()).is_none());
1301
1302 assert!(wheel.is_empty()); // 时间轮应该为空
1303 }
1304
1305 #[test]
1306 fn test_delay_to_ticks() {
1307 let wheel = Wheel::new(WheelConfig::default(), BatchConfig::default());
1308 assert_eq!(wheel.delay_to_ticks(Duration::from_millis(100)), 10);
1309 assert_eq!(wheel.delay_to_ticks(Duration::from_millis(50)), 5);
1310 assert_eq!(wheel.delay_to_ticks(Duration::from_millis(1)), 1); // Minimum 1 tick (最小 1 个 tick)
1311 }
1312
1313 #[test]
1314 fn test_wheel_invalid_slot_count() {
1315 let result = WheelConfig::builder().l0_slot_count(100).build();
1316 assert!(result.is_err());
1317 if let Err(crate::error::TimerError::InvalidSlotCount { slot_count, reason }) = result {
1318 assert_eq!(slot_count, 100);
1319 assert_eq!(reason, "L0 layer slot count must be power of 2"); // L0 层槽数量必须是 2 的幂
1320 } else {
1321 panic!("Expected InvalidSlotCount error"); // 期望 InvalidSlotCount 错误
1322 }
1323 }
1324
1325 #[test]
1326 fn test_minimum_delay() {
1327 let mut wheel = Wheel::new(WheelConfig::default(), BatchConfig::default());
1328
1329 // Test minimum delay (delays less than 1 tick should be rounded up to 1 tick) (测试最小延迟 (延迟小于 1 个 tick 应该向上舍入到 1 个 tick))
1330 let callback = CallbackWrapper::new(|| async {});
1331 let task = TimerTask::new_oneshot(Duration::from_millis(1), Some(callback));
1332 let (task_with_notifier, _completion_rx) =
1333 TimerTaskWithCompletionNotifier::from_timer_task(task);
1334 let handle = wheel.allocate_handle();
1335 let task_id: TaskId = handle.task_id();
1336 wheel.insert(handle, task_with_notifier);
1337
1338 // Advance 1 tick, task should trigger (前进 1 个 tick，任务应该触发)
1339 let expired = wheel.advance();
1340 assert_eq!(
1341 expired.len(),
1342 1,
1343 "Minimum delay task should be triggered after 1 tick"
1344 ); // 最小延迟任务应该在 1 个 tick 后触发
1345 assert_eq!(expired[0].id, task_id);
1346 }
1347
1348 #[test]
1349 fn test_advance_empty_slots() {
1350 let mut wheel = Wheel::new(WheelConfig::default(), BatchConfig::default());
1351
1352 // Do not insert any tasks, advance multiple ticks (不插入任何任务，前进多个tick)
1353 for _ in 0..100 {
1354 let expired = wheel.advance();
1355 assert!(
1356 expired.is_empty(),
1357 "Empty slots should not return any tasks"
1358 );
1359 }
1360
1361 assert_eq!(
1362 wheel.current_tick(),
1363 100,
1364 "current_tick should correctly increment"
1365 ); // current_tick 应该正确递增
1366 }
1367
1368 #[test]
1369 fn test_slot_boundary() {
1370 let mut wheel = Wheel::new(WheelConfig::default(), BatchConfig::default());
1371
1372 // Test slot boundary and wraparound (测试槽边界和环绕)
1373 // 第一个任务：延迟 10ms（1 tick），应该在 slot 1 触发
1374 let callback1 = CallbackWrapper::new(|| async {});
1375 let task1 = TimerTask::new_oneshot(Duration::from_millis(10), Some(callback1));
1376 let (task_with_notifier1, _rx1) = TimerTaskWithCompletionNotifier::from_timer_task(task1);
1377 let handle1 = wheel.allocate_handle();
1378 let task_id_1 = handle1.task_id();
1379 wheel.insert(handle1, task_with_notifier1);
1380
1381 // Second task: delay 5110ms (511 ticks), should trigger on slot 511 (第二个任务：延迟 5110毫秒 (511个tick)，应该在槽 511 触发)
1382 let callback2 = CallbackWrapper::new(|| async {});
1383 let task2 = TimerTask::new_oneshot(Duration::from_millis(5110), Some(callback2));
1384 let (task_with_notifier2, _rx2) = TimerTaskWithCompletionNotifier::from_timer_task(task2);
1385 let handle2 = wheel.allocate_handle();
1386 let task_id_2 = handle2.task_id();
1387 wheel.insert(handle2, task_with_notifier2);
1388
1389 // Advance 1 tick, first task should trigger (前进 1 个 tick，第一个任务应该触发)
1390 let expired = wheel.advance();
1391 assert_eq!(expired.len(), 1, "First task should trigger on tick 1"); // 第一个任务应该在第 1 个 tick 触发
1392 assert_eq!(expired[0].id, task_id_1);
1393
1394 // Continue advancing to 511 ticks (from tick 1 to tick 511), second task should trigger (继续前进 511个tick (从tick 1到tick 511)，第二个任务应该触发)
1395 let mut triggered = false;
1396 for i in 0..510 {
1397 let expired = wheel.advance();
1398 if !expired.is_empty() {
1399 assert_eq!(
1400 expired.len(),
1401 1,
1402 "The {}th advance should trigger the second task",
1403 i + 2
1404 ); // 第 {i + 2} 次前进应该触发第二个任务
1405 assert_eq!(expired[0].id, task_id_2);
1406 triggered = true;
1407 break;
1408 }
1409 }
1410 assert!(triggered, "Second task should trigger on tick 511"); // 第二个任务应该在第 511 个 tick 触发
1411
1412 assert!(wheel.is_empty(), "All tasks should have been triggered"); // 所有任务都应该被触发
1413 }
1414
1415 #[test]
1416 fn test_task_id_uniqueness() {
1417 let mut wheel = Wheel::new(WheelConfig::default(), BatchConfig::default());
1418
1419 // Insert multiple tasks, verify TaskId uniqueness (插入多个任务，验证 TaskId 唯一性)
1420 let mut task_ids = std::collections::HashSet::new();
1421 for _ in 0..100 {
1422 let callback = CallbackWrapper::new(|| async {});
1423 let task = TimerTask::new_oneshot(Duration::from_millis(100), Some(callback));
1424 let (task_with_notifier, _rx) = TimerTaskWithCompletionNotifier::from_timer_task(task);
1425 let handle = wheel.allocate_handle();
1426 let task_id = handle.task_id();
1427 wheel.insert(handle, task_with_notifier);
1428
1429 assert!(task_ids.insert(task_id), "TaskId should be unique"); // TaskId 应该唯一
1430 }
1431
1432 assert_eq!(task_ids.len(), 100);
1433 }
1434}