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