kestrel_timer/task.rs
1use crate::error::TimerError;
2use std::collections::VecDeque;
3use std::future::Future;
4use std::num::{NonZeroU64, NonZeroUsize};
5use std::pin::Pin;
6use std::sync::Arc;
7use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
8use tokio::time::Instant;
9
10use deferred_map::{DefaultKey, Key};
11use lite_sync::oneshot::lite::{Receiver, Sender, State, channel};
12use parking_lot::Mutex;
13use tokio::sync::watch;
14use tokio::sync::{Notify, mpsc};
15
16/// One-shot task completion state constants
17///
18/// 一次性任务完成状态常量
19const ONESHOT_PENDING: u8 = 0;
20const ONESHOT_CALLED: u8 = 1;
21const ONESHOT_CANCELLED: u8 = 2;
22const ONESHOT_CLOSED: u8 = 3;
23const ONESHOT_RECEIVER_CLOSED: u8 = 4;
24
25/// Task Completion Reason for Periodic Tasks
26///
27/// Indicates the reason for task completion, called or cancelled.
28///
29/// 任务完成原因,调用或取消。
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum TaskCompletion {
32 /// Task was called
33 ///
34 /// 任务被调用
35 Called,
36 /// Task was cancelled
37 ///
38 /// 任务被取消
39 Cancelled,
40}
41
42impl State for TaskCompletion {
43 #[inline]
44 fn to_u8(&self) -> u8 {
45 match self {
46 TaskCompletion::Called => ONESHOT_CALLED,
47 TaskCompletion::Cancelled => ONESHOT_CANCELLED,
48 }
49 }
50
51 #[inline]
52 fn from_u8(value: u8) -> Option<Self> {
53 match value {
54 ONESHOT_CALLED => Some(TaskCompletion::Called),
55 ONESHOT_CANCELLED => Some(TaskCompletion::Cancelled),
56 _ => None,
57 }
58 }
59
60 #[inline]
61 fn pending_value() -> u8 {
62 ONESHOT_PENDING
63 }
64
65 #[inline]
66 fn closed_value() -> u8 {
67 ONESHOT_CLOSED
68 }
69
70 #[inline]
71 fn receiver_closed_value() -> u8 {
72 ONESHOT_RECEIVER_CLOSED
73 }
74}
75
76/// Globally unique identity for a timing wheel instance.
77///
78/// The value is private so callers cannot manufacture an owner identity for a
79/// different wheel. It is kept separate from DeferredMap's debug-only map ID,
80/// which is not available in release builds.
81///
82/// 时间轮实例的全局唯一身份。
83///
84/// 该值保持私有,因此调用者无法伪造另一个时间轮的 owner。它独立于
85/// DeferredMap 仅在 debug 构建中提供的 map ID。
86#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
87pub(crate) struct WheelId(NonZeroU64);
88
89static NEXT_WHEEL_ID: AtomicU64 = AtomicU64::new(1);
90
91impl WheelId {
92 #[inline]
93 pub(crate) fn new() -> Self {
94 let value = NEXT_WHEEL_ID
95 .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
96 current.checked_add(1)
97 })
98 .expect("timing wheel owner ID space exhausted");
99
100 Self(NonZeroU64::new(value).expect("timing wheel owner ID must be non-zero"))
101 }
102}
103
104/// Shared owner state for a timing wheel.
105///
106/// Besides identifying the wheel, this state receives unconsumed deferred-map
107/// handles. The wheel drains that queue before allocating another handle, so a
108/// failed cross-wheel registration does not permanently reserve a slot.
109///
110/// 时间轮共享 owner 状态。
111///
112/// 除了标识时间轮,该状态还接收未消费的 deferred-map handle。时间轮会在
113/// 再次分配 handle 前清空队列,因此跨轮注册失败不会永久占用槽位。
114pub(crate) struct WheelOwner {
115 id: WheelId,
116 released_handles: Mutex<Vec<deferred_map::Handle>>,
117 closed: AtomicBool,
118 close_signal: watch::Sender<bool>,
119}
120
121impl WheelOwner {
122 #[inline]
123 pub(crate) fn new() -> Arc<Self> {
124 let (close_signal, _) = watch::channel(false);
125 Arc::new(Self {
126 id: WheelId::new(),
127 released_handles: Mutex::new(Vec::new()),
128 closed: AtomicBool::new(false),
129 close_signal,
130 })
131 }
132
133 #[inline]
134 pub(crate) fn id(&self) -> WheelId {
135 self.id
136 }
137
138 #[inline]
139 fn release_handle(&self, handle: deferred_map::Handle) {
140 self.released_handles.lock().push(handle);
141 }
142
143 #[inline]
144 pub(crate) fn take_released_handles(&self) -> Vec<deferred_map::Handle> {
145 std::mem::take(&mut *self.released_handles.lock())
146 }
147
148 #[inline]
149 pub(crate) fn close(&self) {
150 self.closed.store(true, Ordering::Release);
151 let _ = self.close_signal.send(true);
152 }
153
154 #[inline]
155 pub(crate) fn is_closed(&self) -> bool {
156 self.closed.load(Ordering::Acquire)
157 }
158
159 #[inline]
160 pub(crate) fn subscribe_closed(&self) -> watch::Receiver<bool> {
161 self.close_signal.subscribe()
162 }
163}
164
165/// Unique identifier for timer tasks.
166///
167/// A task ID includes the owning timing wheel identity in addition to the
168/// generational DeferredMap key. The owner field is private and is checked by
169/// every wheel map operation before the raw key reaches DeferredMap.
170///
171/// 定时器任务唯一标识符。
172///
173/// 任务 ID 除了包含 DeferredMap 的代数 key,还包含所属时间轮身份。owner
174/// 字段保持私有,所有时间轮 map 操作都会在使用 raw key 前先校验它。
175#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
176pub struct TaskId {
177 wheel_id: WheelId,
178 key: DefaultKey,
179}
180
181impl TaskId {
182 /// Create TaskId from DeferredMap key (internal use)
183 ///
184 /// 从 DeferredMap key 创建 TaskId (内部使用)
185 #[inline]
186 pub(crate) fn from_key(wheel_id: WheelId, key: DefaultKey) -> Self {
187 Self { wheel_id, key }
188 }
189
190 /// Get the DeferredMap key
191 ///
192 /// 获取 DeferredMap key
193 #[inline]
194 pub(crate) fn key(&self) -> DefaultKey {
195 self.key
196 }
197
198 /// Check whether this ID belongs to the specified timing wheel.
199 ///
200 /// 检查该 ID 是否属于指定时间轮。
201 #[inline]
202 pub(crate) fn belongs_to(&self, wheel_id: WheelId) -> bool {
203 self.wheel_id == wheel_id
204 }
205
206 #[cfg(test)]
207 #[inline]
208 pub(crate) fn wheel_id(&self) -> WheelId {
209 self.wheel_id
210 }
211
212 /// Get the numeric value of the task ID
213 ///
214 /// 获取任务 ID 的数值
215 #[inline]
216 pub fn raw(&self) -> u64 {
217 self.key.raw()
218 }
219}
220
221pub struct TaskHandle {
222 owner: Arc<WheelOwner>,
223 handle: Option<deferred_map::Handle>,
224}
225
226impl TaskHandle {
227 /// Create a new task handle
228 ///
229 /// 创建一个新的任务句柄
230 #[inline]
231 pub(crate) fn new(owner: Arc<WheelOwner>, handle: deferred_map::Handle) -> Self {
232 Self {
233 owner,
234 handle: Some(handle),
235 }
236 }
237
238 /// Get the task ID
239 ///
240 /// 获取任务 ID
241 #[inline]
242 pub fn task_id(&self) -> TaskId {
243 let handle = self
244 .handle
245 .as_ref()
246 .expect("task handle has already been consumed");
247 TaskId::from_key(self.owner.id(), handle.key())
248 }
249
250 /// Check whether this handle belongs to the specified timing wheel.
251 ///
252 /// 检查该 handle 是否属于指定时间轮。
253 #[inline]
254 pub(crate) fn belongs_to(&self, wheel_id: WheelId) -> bool {
255 self.owner.id() == wheel_id
256 }
257
258 /// Convert to deferred map handle
259 ///
260 /// 转换为 deferred map 句柄
261 #[inline]
262 pub(crate) fn into_handle(mut self) -> deferred_map::Handle {
263 self.handle
264 .take()
265 .expect("task handle has already been consumed")
266 }
267}
268
269impl Drop for TaskHandle {
270 fn drop(&mut self) {
271 if let Some(handle) = self.handle.take() {
272 self.owner.release_handle(handle);
273 }
274 }
275}
276
277/// Timer Callback Trait
278///
279/// Types implementing this trait can be used as timer callbacks.
280///
281/// 可实现此特性的类型可以作为定时器回调函数。
282///
283/// # Examples (示例)
284///
285/// ```
286/// use kestrel_timer::task::TimerCallback;
287/// use std::future::Future;
288/// use std::pin::Pin;
289///
290/// struct MyCallback;
291///
292/// impl TimerCallback for MyCallback {
293/// fn call(&self) -> Pin<Box<dyn Future<Output = ()> + Send>> {
294/// Box::pin(async {
295/// println!("Timer callback executed!");
296/// })
297/// }
298/// }
299/// ```
300pub trait TimerCallback: Send + Sync + 'static {
301 /// Execute callback, returns a Future
302 ///
303 /// 执行回调函数,返回一个 Future
304 fn call(&self) -> Pin<Box<dyn Future<Output = ()> + Send>>;
305}
306
307/// Implement TimerCallback trait for closures
308///
309/// Supports Fn() -> Future closures, can be called multiple times, suitable for periodic tasks
310///
311/// 实现 TimerCallback 特性的类型,支持 Fn() -> Future 闭包,可以多次调用,适合周期性任务
312impl<F, Fut> TimerCallback for F
313where
314 F: Fn() -> Fut + Send + Sync + 'static,
315 Fut: Future<Output = ()> + Send + 'static,
316{
317 fn call(&self) -> Pin<Box<dyn Future<Output = ()> + Send>> {
318 Box::pin(self())
319 }
320}
321
322/// Callback wrapper for standardized callback creation and management
323///
324/// Callback 包装器,用于标准化回调创建和管理
325///
326/// # Examples (示例)
327///
328/// ```
329/// use kestrel_timer::CallbackWrapper;
330///
331/// let callback = CallbackWrapper::new(|| async {
332/// println!("Timer callback executed!");
333/// });
334/// ```
335#[derive(Clone)]
336pub struct CallbackWrapper {
337 callback: Arc<dyn TimerCallback>,
338}
339
340impl CallbackWrapper {
341 /// Create a new callback wrapper
342 ///
343 /// # Parameters
344 /// - `callback`: Callback object implementing TimerCallback trait
345 ///
346 /// # 创建一个新的回调包装器
347 ///
348 /// # 参数
349 /// - `callback`: 实现 TimerCallback 特性的回调对象
350 ///
351 /// # Examples (示例)
352 ///
353 /// ```
354 /// use kestrel_timer::CallbackWrapper;
355 ///
356 /// let callback = CallbackWrapper::new(|| async {
357 /// println!("Timer fired!"); // 定时器触发
358 /// });
359 /// ```
360 #[inline]
361 pub fn new(callback: impl TimerCallback) -> Self {
362 Self {
363 callback: Arc::new(callback),
364 }
365 }
366
367 /// Call the callback function
368 ///
369 /// 调用回调函数
370 #[inline]
371 pub(crate) fn call(&self) -> Pin<Box<dyn Future<Output = ()> + Send>> {
372 self.callback.call()
373 }
374}
375
376/// Task type enum to distinguish between one-shot and periodic timers
377///
378/// 任务类型枚举,用于区分一次性和周期性定时器
379#[derive(Clone)]
380pub enum TaskType {
381 /// One-shot timer: executes once and completes
382 ///
383 /// 一次性定时器:执行一次后完成
384 OneShot,
385
386 /// Periodic timer: repeats at fixed intervals
387 ///
388 /// 周期性定时器:按固定间隔重复执行
389 Periodic {
390 /// Period interval for periodic tasks
391 ///
392 /// 周期任务的间隔时间
393 interval: std::time::Duration,
394 /// Maximum number of pending Called notifications; None is unbounded
395 ///
396 /// 待消费 Called 通知的最大数量;None 表示无界
397 buffer_size: Option<NonZeroUsize>,
398 },
399}
400
401/// Task type enum to distinguish between one-shot and periodic timers
402///
403/// 任务类型枚举,用于区分一次性和周期性定时器
404pub enum TaskTypeWithCompletionNotifier {
405 /// One-shot timer: executes once and completes
406 ///
407 /// 一次性定时器:执行一次后完成
408 OneShot {
409 completion_notifier: Sender<TaskCompletion>,
410 },
411
412 /// Periodic timer: repeats at fixed intervals
413 ///
414 /// 周期性定时器:按固定间隔重复执行
415 Periodic {
416 /// Period interval for periodic tasks
417 ///
418 /// 周期任务的间隔时间
419 interval: std::time::Duration,
420 /// Completion notifier for periodic tasks
421 ///
422 /// 周期性任务完成通知器
423 completion_notifier: PeriodicCompletionNotifier,
424 },
425}
426
427impl TaskTypeWithCompletionNotifier {
428 /// Get the interval for periodic tasks
429 ///
430 /// Returns `None` for one-shot tasks
431 ///
432 /// 获取周期任务的间隔时间
433 ///
434 /// 对于一次性任务返回 `None`
435 #[inline]
436 pub fn get_interval(&self) -> Option<std::time::Duration> {
437 match self {
438 TaskTypeWithCompletionNotifier::Periodic { interval, .. } => Some(*interval),
439 TaskTypeWithCompletionNotifier::OneShot { .. } => None,
440 }
441 }
442}
443
444pub(crate) struct PeriodicCompletionQueue {
445 state: Mutex<PeriodicCompletionQueueState>,
446 notify: Notify,
447}
448
449struct PeriodicCompletionQueueState {
450 events: VecDeque<TaskCompletion>,
451 capacity: Option<NonZeroUsize>,
452 terminal_pending: bool,
453 closed: bool,
454 receiver_alive: bool,
455 dropped_called: usize,
456}
457
458enum PeriodicReceiveState {
459 Empty,
460 Closed,
461}
462
463impl PeriodicCompletionQueue {
464 fn new(capacity: Option<NonZeroUsize>) -> Arc<Self> {
465 Arc::new(Self {
466 state: Mutex::new(PeriodicCompletionQueueState {
467 events: VecDeque::new(),
468 capacity,
469 terminal_pending: false,
470 closed: false,
471 receiver_alive: true,
472 dropped_called: 0,
473 }),
474 notify: Notify::new(),
475 })
476 }
477
478 fn send(&self, completion: TaskCompletion) -> bool {
479 let mut state = self.state.lock();
480 if state.closed || !state.receiver_alive {
481 return false;
482 }
483
484 if completion == TaskCompletion::Cancelled {
485 state.terminal_pending = true;
486 drop(state);
487 self.notify.notify_waiters();
488 return true;
489 }
490
491 if state.terminal_pending {
492 return false;
493 }
494
495 if let Some(capacity) = state.capacity {
496 if state.events.len() >= capacity.get() {
497 state.dropped_called = state.dropped_called.saturating_add(1);
498 return false;
499 }
500 }
501
502 state.events.push_back(completion);
503 drop(state);
504 self.notify.notify_one();
505 true
506 }
507
508 fn try_receive(&self) -> Result<TaskCompletion, PeriodicReceiveState> {
509 let mut state = self.state.lock();
510 if let Some(completion) = state.events.pop_front() {
511 return Ok(completion);
512 }
513
514 if state.terminal_pending {
515 state.terminal_pending = false;
516 return Ok(TaskCompletion::Cancelled);
517 }
518
519 if state.closed {
520 Err(PeriodicReceiveState::Closed)
521 } else {
522 Err(PeriodicReceiveState::Empty)
523 }
524 }
525
526 fn close_sender(&self) {
527 let mut state = self.state.lock();
528 state.closed = true;
529 drop(state);
530 self.notify.notify_waiters();
531 }
532
533 fn close_receiver(&self) {
534 let mut state = self.state.lock();
535 state.receiver_alive = false;
536 state.closed = true;
537 state.events.clear();
538 state.terminal_pending = false;
539 drop(state);
540 self.notify.notify_waiters();
541 }
542
543 fn dropped_called(&self) -> usize {
544 self.state.lock().dropped_called
545 }
546}
547
548/// Completion notifier for periodic tasks.
549///
550/// `Some(buffer_size)` keeps at most that many pending `Called` notifications.
551/// `None` keeps every notification. A final `Cancelled` notification is kept
552/// separately and is never evicted by the bounded queue.
553///
554/// 周期任务完成通知器。
555///
556/// `Some(buffer_size)` 最多保留指定数量的待消费 `Called` 通知;`None` 保留
557/// 所有通知。最终的 `Cancelled` 通知独立保存,不会被有界队列挤出。
558pub struct PeriodicCompletionNotifier(pub(crate) Arc<PeriodicCompletionQueue>);
559
560impl PeriodicCompletionNotifier {
561 #[inline]
562 pub(crate) fn send(&self, completion: TaskCompletion) -> bool {
563 self.0.send(completion)
564 }
565}
566
567impl Drop for PeriodicCompletionNotifier {
568 fn drop(&mut self) {
569 self.0.close_sender();
570 }
571}
572
573/// Completion receiver for periodic tasks
574///
575/// 周期任务完成通知接收器
576pub struct PeriodicCompletionReceiver(pub(crate) Arc<PeriodicCompletionQueue>);
577
578impl PeriodicCompletionReceiver {
579 /// Try to receive a completion notification
580 ///
581 /// 尝试接收完成通知
582 #[inline]
583 pub fn try_recv(&mut self) -> Result<TaskCompletion, mpsc::error::TryRecvError> {
584 match self.0.try_receive() {
585 Ok(completion) => Ok(completion),
586 Err(PeriodicReceiveState::Empty) => Err(mpsc::error::TryRecvError::Empty),
587 Err(PeriodicReceiveState::Closed) => Err(mpsc::error::TryRecvError::Disconnected),
588 }
589 }
590
591 /// Receive a completion notification
592 ///
593 /// 接收完成通知
594 #[inline]
595 pub async fn recv(&mut self) -> Option<TaskCompletion> {
596 loop {
597 let notified = self.0.notify.notified();
598 match self.0.try_receive() {
599 Ok(completion) => return Some(completion),
600 Err(PeriodicReceiveState::Closed) => return None,
601 Err(PeriodicReceiveState::Empty) => notified.await,
602 }
603 }
604 }
605
606 /// Get the number of `Called` notifications discarded by a bounded queue.
607 ///
608 /// 获取有界队列丢弃的 `Called` 通知数量。
609 #[inline]
610 pub fn dropped_notifications(&self) -> usize {
611 self.0.dropped_called()
612 }
613}
614
615impl Drop for PeriodicCompletionReceiver {
616 fn drop(&mut self) {
617 self.0.close_receiver();
618 }
619}
620
621/// Completion notifier for one-shot tasks
622///
623/// 一次性任务完成通知器
624pub enum CompletionNotifier {
625 OneShot(Sender<TaskCompletion>),
626 Periodic(PeriodicCompletionNotifier),
627}
628
629/// Completion receiver for one-shot and periodic tasks
630///
631/// 一次性和周期任务完成通知接收器
632pub enum CompletionReceiver {
633 OneShot(Receiver<TaskCompletion>),
634 Periodic(PeriodicCompletionReceiver),
635}
636
637/// Timer Task
638///
639/// Users interact via a two-step API
640/// 1. Create task using `TimerTask::new_oneshot()` or `TimerTask::new_periodic()`
641/// 2. Register task using `TimerWheel::register()` or `TimerService::register()`
642///
643/// TaskId is assigned when the task is inserted into the timing wheel.
644///
645/// 定时器任务
646///
647/// 用户通过两步 API 与定时器交互
648/// 1. 使用 `TimerTask::new_oneshot()` 或 `TimerTask::new_periodic()` 创建任务
649/// 2. 使用 `TimerWheel::register()` 或 `TimerService::register()` 注册任务
650///
651/// TaskId 在任务插入到时间轮时分配
652pub struct TimerTask {
653 /// Task type (one-shot or periodic)
654 ///
655 /// 任务类型(一次性或周期性)
656 pub(crate) task_type: TaskType,
657
658 /// User-specified delay duration (initial delay for periodic tasks)
659 ///
660 /// 用户指定的延迟时间(周期任务的初始延迟)
661 pub(crate) delay: std::time::Duration,
662
663 /// Async callback function, optional
664 ///
665 /// 异步回调函数,可选
666 pub(crate) callback: Option<CallbackWrapper>,
667}
668
669impl TimerTask {
670 /// Create a new one-shot timer task
671 ///
672 /// # Parameters
673 /// - `delay`: Delay duration before task execution
674 /// - `callback`: Callback function, optional
675 ///
676 /// # Note
677 /// TaskId will be assigned when the task is inserted into the timing wheel.
678 /// 创建一个新的一次性定时器任务
679 ///
680 /// # 参数
681 /// - `delay`: 任务执行前的延迟时间
682 /// - `callback`: 回调函数,可选
683 ///
684 /// # 注意
685 /// TaskId 将在任务插入到时间轮时分配
686 #[inline]
687 pub fn new_oneshot(delay: std::time::Duration, callback: Option<CallbackWrapper>) -> Self {
688 Self {
689 task_type: TaskType::OneShot,
690 delay,
691 callback,
692 }
693 }
694
695 /// Create a new periodic timer task
696 ///
697 /// # Parameters
698 /// - `initial_delay`: Initial delay before first execution
699 /// - `interval`: Interval between subsequent executions
700 /// - `callback`: Callback function, optional
701 /// - `buffer_size`: Maximum pending `Called` notifications; `None` is
702 /// unbounded
703 ///
704 /// # Note
705 /// TaskId will be assigned when the task is inserted into the timing wheel.
706 /// The interval must be greater than zero. The timing wheel also requires it
707 /// to be at least the configured L0 tick duration.
708 ///
709 /// Returns `Err(TimerError::InvalidConfiguration)` for a zero interval.
710 ///
711 /// 创建一个新的周期性定时器任务
712 ///
713 /// # 参数
714 /// - `initial_delay`: 首次执行前的初始延迟
715 /// - `interval`: 后续执行之间的间隔
716 /// - `callback`: 回调函数,可选
717 /// - `buffer_size`: 待消费 `Called` 通知的最大数量;`None` 表示无界
718 ///
719 /// # 注意
720 /// TaskId 将在任务插入到时间轮时分配。
721 /// 有界模式超出的 `Called` 通知可通过 `PeriodicCompletionReceiver::dropped_notifications`
722 /// 查询。
723 #[inline]
724 pub fn new_periodic(
725 initial_delay: std::time::Duration,
726 interval: std::time::Duration,
727 callback: Option<CallbackWrapper>,
728 buffer_size: Option<NonZeroUsize>,
729 ) -> Result<Self, TimerError> {
730 if interval.is_zero() {
731 return Err(TimerError::InvalidConfiguration {
732 field: "interval".to_string(),
733 reason: "periodic interval must be greater than zero".to_string(),
734 });
735 }
736
737 Ok(Self {
738 task_type: TaskType::Periodic {
739 interval,
740 buffer_size,
741 },
742 delay: initial_delay,
743 callback,
744 })
745 }
746
747 /// Get task type
748 ///
749 /// 获取任务类型
750 #[inline]
751 pub fn get_task_type(&self) -> &TaskType {
752 &self.task_type
753 }
754
755 /// Get the interval for periodic tasks
756 ///
757 /// Returns `None` for one-shot tasks
758 ///
759 /// 获取周期任务的间隔时间
760 ///
761 /// 对于一次性任务返回 `None`
762 #[inline]
763 pub fn get_interval(&self) -> Option<std::time::Duration> {
764 match self.task_type {
765 TaskType::Periodic { interval, .. } => Some(interval),
766 TaskType::OneShot => None,
767 }
768 }
769}
770
771/// Timer Task
772///
773/// Users interact via a two-step API
774/// 1. Create task using `TimerTask::new_oneshot()` or `TimerTask::new_periodic()`
775/// 2. Register task using `TimerWheel::register()` or `TimerService::register()`
776///
777/// 定时器任务
778///
779/// 用户通过两步 API 与定时器交互
780/// 1. 使用 `TimerTask::new_oneshot()` 或 `TimerTask::new_periodic()` 创建任务
781/// 2. 使用 `TimerWheel::register()` 或 `TimerService::register()` 注册任务
782pub struct TimerTaskWithCompletionNotifier {
783 /// Task type (one-shot or periodic)
784 ///
785 /// 任务类型(一次性或周期性)
786 pub(crate) task_type: TaskTypeWithCompletionNotifier,
787
788 /// User-specified delay duration (initial delay for periodic tasks)
789 ///
790 /// 用户指定的延迟时间(周期任务的初始延迟)
791 pub(crate) delay: std::time::Duration,
792
793 /// Async callback function, optional
794 ///
795 /// 异步回调函数,可选
796 pub(crate) callback: Option<CallbackWrapper>,
797}
798
799impl TimerTaskWithCompletionNotifier {
800 /// Create a new timer task with completion notifier from a timer task
801 ///
802 /// TaskId will be assigned later when inserted into the timing wheel
803 ///
804 /// 从定时器任务创建一个新的定时器任务完成通知器
805 ///
806 /// TaskId 将在插入到时间轮时分配
807 ///
808 /// # Parameters
809 /// - `task`: The timer task to create from
810 ///
811 /// # Returns
812 /// A tuple containing the new timer task with completion notifier and the completion receiver
813 ///
814 /// 返回一个包含新的定时器任务完成通知器和完成通知接收器的元组
815 ///
816 pub fn from_timer_task(task: TimerTask) -> (Self, CompletionReceiver) {
817 match task.task_type {
818 TaskType::OneShot => {
819 // Create oneshot notifier and receiver with optimized single Arc allocation
820 // 创建 oneshot 通知器和接收器,使用优化的单个 Arc 分配
821 let (notifier, receiver) = channel();
822
823 (
824 Self {
825 task_type: TaskTypeWithCompletionNotifier::OneShot {
826 completion_notifier: notifier,
827 },
828 delay: task.delay,
829 callback: task.callback,
830 },
831 CompletionReceiver::OneShot(receiver),
832 )
833 }
834 TaskType::Periodic {
835 interval,
836 buffer_size,
837 } => {
838 let queue = PeriodicCompletionQueue::new(buffer_size);
839
840 let notifier = PeriodicCompletionNotifier(Arc::clone(&queue));
841 let receiver = PeriodicCompletionReceiver(queue);
842
843 (
844 Self {
845 task_type: TaskTypeWithCompletionNotifier::Periodic {
846 interval,
847 completion_notifier: notifier,
848 },
849 delay: task.delay,
850 callback: task.callback,
851 },
852 CompletionReceiver::Periodic(receiver),
853 )
854 }
855 }
856 }
857
858 /// Into task type
859 ///
860 /// 将任务类型转换为完成通知器
861 #[inline]
862 pub fn into_task_type(self) -> TaskTypeWithCompletionNotifier {
863 self.task_type
864 }
865
866 /// Get the interval for periodic tasks
867 ///
868 /// Returns `None` for one-shot tasks
869 ///
870 /// 获取周期任务的间隔时间
871 ///
872 /// 对于一次性任务返回 `None`
873 #[inline]
874 pub fn get_interval(&self) -> Option<std::time::Duration> {
875 self.task_type.get_interval()
876 }
877}
878
879pub(crate) struct TimerTaskForWheel {
880 pub(crate) task_id: TaskId,
881 pub(crate) task: TimerTaskWithCompletionNotifier,
882 pub(crate) deadline_tick: u64,
883 pub(crate) rounds: u32,
884 pub(crate) deadline_at: Instant,
885}
886
887impl TimerTaskForWheel {
888 /// Create a new timer task for wheel with assigned TaskId
889 ///
890 /// 使用分配的 TaskId 创建一个新的定时器任务用于时间轮
891 ///
892 /// # Parameters
893 /// - `task_id`: The TaskId assigned by DeferredMap
894 /// - `task`: The timer task to create from
895 /// - `deadline_tick`: The deadline tick for the task
896 /// - `rounds`: The rounds for the task
897 ///
898 /// # Returns
899 /// A new timer task for wheel
900 ///
901 /// 返回一个新的定时器任务用于时间轮
902 ///
903 #[inline]
904 pub(crate) fn new_with_id(
905 task_id: TaskId,
906 task: TimerTaskWithCompletionNotifier,
907 deadline_tick: u64,
908 rounds: u32,
909 deadline_at: Instant,
910 ) -> Self {
911 Self {
912 task_id,
913 task,
914 deadline_tick,
915 rounds,
916 deadline_at,
917 }
918 }
919
920 /// Get task ID
921 ///
922 /// 获取任务 ID
923 #[inline]
924 pub fn get_id(&self) -> TaskId {
925 self.task_id
926 }
927
928 /// Into task type
929 ///
930 /// 将任务类型转换为完成通知器
931 #[inline]
932 pub fn into_task_type(self) -> TaskTypeWithCompletionNotifier {
933 self.task.into_task_type()
934 }
935
936 /// Update the delay of the timer task
937 ///
938 /// 更新定时器任务的延迟
939 ///
940 /// # Parameters
941 /// - `delay`: The new delay for the task
942 ///
943 #[inline]
944 pub fn update_delay(&mut self, delay: std::time::Duration) {
945 self.task.delay = delay
946 }
947
948 /// Update the callback of the timer task
949 ///
950 /// 更新定时器任务的回调
951 ///
952 /// # Parameters
953 /// - `callback`: The new callback for the task
954 ///
955 #[inline]
956 pub fn update_callback(&mut self, callback: CallbackWrapper) {
957 self.task.callback = Some(callback)
958 }
959}
960
961/// Task location information (including level) for hierarchical timing wheel
962///
963/// Memory layout optimization: level field placed first to reduce padding via struct alignment
964///
965/// 任务位置信息(包括层级)用于分层时间轮
966///
967/// 内存布局优化:将 level 字段放在第一位,通过结构体对齐来减少填充
968#[derive(Debug, Clone, Copy)]
969pub(crate) struct TaskLocation {
970 /// Slot index
971 ///
972 /// 槽索引
973 pub slot_index: usize,
974 /// Index position of task in slot Vec for O(1) cancellation
975 ///
976 /// 槽向量中任务的索引位置,用于 O(1) 取消
977 pub vec_index: usize,
978 /// Level: 0 = L0 (bottom layer), 1 = L1 (upper layer)
979 /// Using u8 instead of bool to reserve space for potential multi-layer expansion
980 ///
981 /// 层级:0 = L0(底层),1 = L1(上层)
982 /// 使用 u8 而不是 bool 来保留空间,用于潜在的多层扩展
983 pub level: u8,
984}
985
986impl TaskLocation {
987 /// Create a new task location information
988 ///
989 /// 创建一个新的任务位置信息
990 #[inline(always)]
991 pub fn new(level: u8, slot_index: usize, vec_index: usize) -> Self {
992 Self {
993 slot_index,
994 vec_index,
995 level,
996 }
997 }
998}