dope_runtime/executor/
mod.rs

1use std::{io, marker::PhantomData};
2
3use dope_core::driver::CompletionBackend;
4
5pub use dope_core::profile::{Profile, Tail, Throughput};
6
7mod ctx;
8mod external;
9pub mod join_handle;
10mod runner;
11pub mod runtime_tuning;
12pub mod task_queue;
13mod timer_heap;
14
15pub use ctx::{RuntimeCtx, RuntimeHandles};
16pub use dope_core::driver::ExternalDispatch;
17pub use external::install_external;
18pub use join_handle::JoinHandle;
19pub use runner::{block_on, current, spawn};
20pub use runtime_tuning::RuntimeTuning;
21pub use task_queue::{DEFAULT_READY_BUDGET, TaskQueue};
22pub(crate) use timer_heap::TimerHeap;
23
24use crate::runtime::{Manifold, Runtime};
25use external::{ExternalRouter, current_external};
26use runtime_tuning::IdleParkPolicy;
27
28#[inline(always)]
29pub fn current_driver_mut<'a, B: CompletionBackend>() -> &'a mut B::Driver {
30    let p = B::current_driver_mut();
31    debug_assert!(
32        !p.is_null(),
33        "dope-runtime: no current driver (called outside runtime?)"
34    );
35
36    unsafe { &mut *p }
37}
38
39#[inline(always)]
40pub fn in_runtime<B: CompletionBackend>() -> bool {
41    !B::current_driver_mut().is_null()
42}
43
44pub struct LocalExecutor<B: CompletionBackend, P: Profile = Throughput> {
45    pub(super) exec: Box<TaskQueue>,
46    pub(super) timers: Box<TimerHeap>,
47    pub(super) idle_park_policy: IdleParkPolicy,
48    pub(super) marker: PhantomData<(B, P)>,
49}
50
51impl<B: CompletionBackend, P: Profile> LocalExecutor<B, P> {
52    pub(super) fn new(idle_park_policy: IdleParkPolicy) -> Self {
53        Self {
54            exec: Box::new(TaskQueue::new()),
55            timers: Box::new(TimerHeap::new()),
56            idle_park_policy,
57            marker: PhantomData,
58        }
59    }
60}
61
62#[inline(always)]
63pub(super) fn run_ready<P: Profile>(exec: &TaskQueue) -> usize {
64    match P::TASK_BUDGET {
65        None => exec.run_ready_all(),
66        Some(n) => exec.run_ready_with_budget(n),
67    }
68}
69
70impl<B: dope_core::driver::PoolDriver, P: Profile> Manifold for LocalExecutor<B, P> {
71    type Backend = B;
72
73    #[inline(always)]
74    fn has_pending(&self, _driver: &mut B::Driver) -> bool {
75        !self.exec.is_empty()
76    }
77
78    #[inline(always)]
79    fn drive(&mut self, driver: &mut B::Driver) {
80        let aux = current_external::<B>();
81        let _ = B::drive_with(driver, &mut ExternalRouter::<B> { aux });
82        if let Some(a) = aux {
83            unsafe { (a.pump)(a.state, driver) };
84        }
85        self.timers.wake_due();
86        run_ready::<P>(&self.exec);
87    }
88}
89
90impl<B, P> Runtime<LocalExecutor<B, P>>
91where
92    B: dope_core::driver::PoolDriver,
93    P: Profile,
94{
95    pub fn new_local() -> io::Result<Self> {
96        Self::new_local_with_tuning(RuntimeTuning::for_profile::<P>())
97    }
98
99    pub fn new_local_with_tuning(
100        tuning: runtime_tuning::RuntimeTuning<B::Config>,
101    ) -> io::Result<Self> {
102        let driver = B::new_driver(tuning.driver)?;
103        let reactor = LocalExecutor::<B, P>::new(tuning.idle_park_policy);
104        Ok(Self::new(reactor, driver))
105    }
106}
dope_runtime/executor/mod.rs

dope_runtime/executor/
mod.rs
