use std::{marker, mem, pin};
use dope_sys::memory::Context as _;
pub(in crate::driver) use epochs::Final;
use finish::Finish;
use o3::{
base::{
cell::{brand, region},
marker::ThreadBound,
},
buffer,
buffer::storage::{provided, provided::fixed},
};
use plans::Provided;
use schedule::state::Core;
use state::scope;
use crate::{
Engine, driver,
driver::{
life::recovery::{Observe, Recover},
retained::flight,
route::lease::Registry,
schedule,
schedule::state,
storage,
storage::{resident, resident::Alloc},
},
io,
};
pub mod creation;
pub mod domain;
pub mod finish;
pub(crate) mod native;
mod plans;
pub mod recovery;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct Busy;
pub mod raw;
pub trait Operation<S: storage::Plan, Q: ?Sized>: Sized {
type Output;
fn rejected(&self, _source: &Q) {}
fn run<'a, 'p, 'd>(
self,
scope: Scope<'a, 'd>,
storage: storage::Loan<'a, 'p, 'd, S>,
source: &'a Q,
) -> Self::Output
where
'd: 'p,
'p: 'a,
'd: 'a,
Self: 'd,
S: 'd,
Q: 'd;
}
pub(crate) struct Local<'d> {
pub(crate) driver: driver::Reference<'d>,
pub(crate) backend: fixed::Loan<'d, Engine>,
pub(in crate::driver) retry: bool,
pub(crate) flights: &'d mut flight::Arena,
pub(in crate::driver) resident: Alloc<'d>,
pub(in crate::driver) routes: Registry,
pub(in crate::driver) region: (
region::Token<region::Identity<'d>>,
region::Token<region::Identity<'d, 1>>,
),
pub(crate) lifecycle: Lifecycle<'d>,
}
pub(crate) struct Lifecycle<'d> {
pub(crate) source: Option<dope_sys::engine::register::Source<'d>>,
pub(crate) resuming: bool,
pub(crate) pending: bool,
pub(crate) report: &'d mut dyn FnMut(io::Reclaimed),
pub(in crate::driver) recover: &'d dyn recovery::Recover,
}
impl Lifecycle<'_> {
pub(crate) fn report(&mut self, event: io::Reclaimed) {
if matches!(event, io::Reclaimed::Shutdown) {
let owned = self.source.take().is_some();
if self.resuming && owned {
self.pending = true;
return;
}
}
(self.report)(event);
}
}
pub(in crate::driver) struct State<'d> {
local: Local<'d>,
work: Core<'d>,
token: brand::Token<brand::Identity<'d>>,
}
impl<'d> State<'d> {
fn new(local: Local<'d>, token: brand::Token<brand::Identity<'d>>) -> Self {
Self {
work: Core::new(local.driver),
local,
token,
}
}
}
mod epochs;
#[repr(transparent)]
pub struct Scope<'a, 'd> {
state: &'a mut State<'d>,
}
impl<'a, 'd> Scope<'a, 'd> {
fn new(state: &'a mut State<'d>) -> Self {
Self { state }
}
pub fn context(&mut self) -> driver::Context<'_, 'd> {
let state = &mut *self.state;
driver::Context::new(&mut state.local)
}
pub fn token(&mut self) -> &mut brand::Token<brand::Identity<'d>> {
&mut self.state.token
}
#[inline]
pub fn split(
&mut self,
) -> (
&mut brand::Token<brand::Identity<'d>>,
driver::Context<'_, 'd>,
scope::Scope<'_, 'd>,
) {
let state = &mut *self.state;
let ctx = driver::Context::new(&mut state.local);
(&mut state.token, ctx, scope::Scope::new(&mut state.work))
}
pub fn driver_ref(&self) -> driver::Reference<'d> {
self.state.local.driver
}
pub fn try_boot(&self) -> Result<(), Busy> {
if self.state.local.flights.is_pending() {
Err(Busy)
} else {
Ok(())
}
}
pub fn quiesce(&mut self) {
let state = &mut *self.state;
let ctx = driver::Context::new(&mut state.local);
Final::suspend(ctx).resume()
}
#[doc(hidden)]
pub fn finish<R>(
&mut self,
forced: bool,
run: impl for<'finish> FnOnce(
&mut brand::Token<brand::Identity<'d>>,
&mut Finish<'finish, 'd>,
) -> R,
) -> R {
let state = &mut *self.state;
let ctx = driver::Context::new(&mut state.local);
let last = Final::suspend(ctx);
let mut finish = Finish::new(last, forced);
let output = run(&mut state.token, &mut finish);
finish.resume();
output
}
}
impl driver::Driver {
fn run_scope<A, Q: ?Sized, O>(
&mut self,
arenas: A,
source: &domain::Source<Q>,
op: O,
report: &mut dyn Observe,
) -> O::Output
where
A: buffer::raw::Arenas,
O: Operation<storage::Scope<A>, Q>,
{
let runtime = &mut self.runtime;
let (report, recover) = report.split();
let root = Root {
flights: &mut runtime.flights,
resident: &mut runtime.resident,
source,
report,
recover,
};
runtime.shared.run(
pin::pin!((root, arenas)),
runtime.shared.context().buffer().layout().0.accounts(),
&mut runtime.backend,
Provided(op),
)
}
fn scope<'b, Q: ?Sized, O>(
&'b mut self,
source: &'b domain::Source<Q>,
op: O,
report: &'b mut dyn Observe,
) -> O::Output
where
O: fixed::Operation<Root<'b, Q>, driver::Buffer, driver::Shared, Terminal = crate::Engine>,
{
let runtime = &mut self.runtime;
let (report, recover) = report.split();
let root = Root {
flights: &mut runtime.flights,
resident: &mut runtime.resident,
source,
report,
recover,
};
runtime.shared.run(
pin::pin!(root),
runtime.shared.context().buffer().layout().0.accounts(),
&mut runtime.backend,
op,
)
}
}
const _: () = {
assert!(mem::size_of::<Scope<'static, 'static>>() == mem::size_of::<&mut ()>());
assert!(mem::align_of::<Scope<'static, 'static>>() == mem::align_of::<&mut ()>());
};
struct Root<'a, Q: ?Sized> {
flights: &'a mut flight::Arena,
resident: &'a mut resident::Arena,
source: &'a domain::Source<Q>,
report: &'a mut dyn FnMut(io::Reclaimed),
recover: &'a dyn Recover,
}
impl<Q: ?Sized> Root<'_, Q> {
fn enter<'d>(
&'d mut self,
shared: provided::Root<'d, driver::Buffer, driver::Shared, ()>,
token: brand::Token<brand::Identity<'d>>,
regions: (
region::Token<region::Identity<'d>>,
region::Token<region::Identity<'d, 1>>,
),
backend: fixed::Loan<'d, Engine>,
) -> (State<'d>, &'d Q) {
let alloc = self.resident.enter(®ions.0);
let local = Local {
driver: driver::Reference::new(shared),
backend,
retry: false,
flights: &mut *self.flights,
resident: alloc,
routes: Registry::new(),
region: regions,
lifecycle: Lifecycle {
source: self.source.descriptor(),
resuming: false,
pending: false,
report: &mut *self.report,
recover: self.recover,
},
};
let state = State::new(local, token);
(state, &self.source.value)
}
}
#[must_use]
pub struct Install<'a, 'd> {
owner: marker::PhantomData<(&'a mut raw::Anchor<'a>, ThreadBound)>,
_driver: driver::Invariant<'d>,
}
impl<'a, 'd> Install<'a, 'd> {
pub fn new<T: raw::Root<'a>>(root: T, driver: driver::Reference<'d>) -> Self {
Self {
owner: root.anchor().owner(),
_driver: driver.brand,
}
}
}
const _: () = {
assert!(mem::size_of::<Install<'static, 'static>>() == 0);
assert!(mem::align_of::<Install<'static, 'static>>() == 1);
};