ax-task 0.8.0

OS-independent IRQ-safe SMP task scheduling core
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

//! Scheduler-backed local future execution.

use core::{
    future::{Future, IntoFuture, poll_fn},
    pin::{Pin, pin},
    task::Poll,
    time::Duration,
};

use super::LocalExecutor;
use crate::{sync::WaitQueue, thread::current::current_thread_handle, time::MonotonicDeadline};

/// Error returned when a runtime-local future misses its monotonic deadline.
#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)]
pub enum BlockOnError {
    /// The requested relative timeout elapsed before the future completed.
    #[error("future polling deadline elapsed")]
    TimedOut,
}

/// Polls a future to completion on the calling runtime scheduler thread.
///
/// The executor is local to this call. Its waker may be invoked from another
/// CPU or hard IRQ, while all future polling and destruction remain on the
/// owner thread.
#[track_caller]
pub fn block_on<F: IntoFuture>(future: F) -> F::Output {
    let thread = current_thread_handle()
        .unwrap_or_else(|error| panic!("future polling requires a scheduler thread: {error}"));
    let wait = WaitQueue::new();
    let executor = LocalExecutor::new(thread.wake_handle())
        .unwrap_or_else(|error| panic!("future executor requires its owner thread: {error}"));
    let output = executor.run(future.into_future(), |condition| {
        wait.wait_until(|| condition.should_abort());
    });
    drop(executor);
    output
}

/// Polls a future until completion or a relative monotonic timeout.
#[track_caller]
pub fn block_on_timeout<F: IntoFuture>(
    timeout: Duration,
    future: F,
) -> Result<F::Output, BlockOnError> {
    let thread = current_thread_handle()
        .unwrap_or_else(|error| panic!("future polling requires a scheduler thread: {error}"));
    let wait = WaitQueue::new();
    let executor = LocalExecutor::new(thread.wake_handle())
        .unwrap_or_else(|error| panic!("future executor requires its owner thread: {error}"));
    let deadline = crate::runtime::task_runtime::monotonic_now().deadline_after(timeout);
    let mut future = pin!(future.into_future());
    let timed = poll_fn(|context| poll_until_deadline(future.as_mut(), context, deadline));
    let output = executor.run(timed, |condition| {
        let _timed_out = wait.wait_until_deadline(deadline, || condition.should_abort());
    });
    drop(executor);
    output
}

fn poll_until_deadline<F: Future>(
    future: Pin<&mut F>,
    context: &mut core::task::Context<'_>,
    deadline: MonotonicDeadline,
) -> Poll<Result<F::Output, BlockOnError>> {
    if let Poll::Ready(output) = Future::poll(future, context) {
        return Poll::Ready(Ok(output));
    }
    if crate::runtime::task_runtime::monotonic_now().reached(deadline) {
        return Poll::Ready(Err(BlockOnError::TimedOut));
    }
    Poll::Pending
}