qubit-thread-pool 0.5.0

Dynamic and fixed thread pool executor services for Qubit Rust libraries
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

/*******************************************************************************
 *
 *    Copyright (c) 2025 - 2026 Haixing Hu.
 *
 *    SPDX-License-Identifier: Apache-2.0
 *
 *    Licensed under the Apache License, Version 2.0.
 *
 ******************************************************************************/
use std::{
    sync::{
        Arc,
        atomic::Ordering,
    },
    time::Instant,
};

use qubit_executor::service::ExecutorServiceLifecycle;

use super::delayed_task_scheduler_inner::DelayedTaskSchedulerInner;
use super::delayed_task_scheduler_state::DelayedTaskSchedulerState;
use super::delayed_task_state::is_task_cancelled;

/// Worker loop entry point for delayed task schedulers.
pub struct DelayedTaskSchedulerWorker;

impl DelayedTaskSchedulerWorker {
    /// Runs the delayed task scheduler loop.
    ///
    /// # Parameters
    ///
    /// * `inner` - Shared scheduler state.
    pub fn run(inner: Arc<DelayedTaskSchedulerInner>) {
        run_delayed_scheduler(inner);
    }
}

/// Runs the delayed task scheduler loop.
///
/// # Parameters
///
/// * `inner` - Shared scheduler state.
fn run_delayed_scheduler(inner: Arc<DelayedTaskSchedulerInner>) {
    loop {
        let task = {
            let mut state = inner.state.lock();
            loop {
                prune_cancelled_front(&mut state);
                if state.lifecycle == ExecutorServiceLifecycle::Stopping {
                    inner.terminate(&mut state);
                    return;
                }
                if state.tasks.is_empty() && state.lifecycle != ExecutorServiceLifecycle::Running {
                    inner.terminate(&mut state);
                    return;
                }
                let Some(next_deadline) = state.tasks.peek().map(|task| task.deadline) else {
                    state = state.wait();
                    continue;
                };
                let now = Instant::now();
                if next_deadline > now {
                    let timeout = next_deadline.saturating_duration_since(now);
                    let (next_state, _) = state.wait_timeout(timeout);
                    state = next_state;
                    continue;
                }
                break state.tasks.pop();
            }
        };
        if let Some(mut task) = task {
            if !inner.start_task_state(&task.state) {
                continue;
            }
            let Some(action) = task.task.take() else {
                continue;
            };
            inner.running_task_count.fetch_add(1, Ordering::AcqRel);
            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(action));
            inner.running_task_count.fetch_sub(1, Ordering::AcqRel);
            inner.completed_task_count.fetch_add(1, Ordering::AcqRel);
            inner.state.notify_all();
        }
    }
}

/// Removes already-cancelled tasks from the front of the deadline heap.
///
/// # Parameters
///
/// * `state` - Locked scheduler state.
fn prune_cancelled_front(state: &mut DelayedTaskSchedulerState) {
    while state
        .tasks
        .peek()
        .is_some_and(|task| is_task_cancelled(&task.state))
    {
        state.tasks.pop();
    }
}