use std::{mem, ops::ControlFlow, pin, task};
use dope_core::driver::{result::poll, route, route::indices, schedule, schedule::ready};
use drive::{clock, turn};
use schedule::clock::{Moment, Sample};
use scope::Roots as _;
use turn::{modes, routes, scope};
use crate::executor::{self, boot, drive, root};
#[must_use = "fixed-point evidence and its root output must be observed"]
pub struct Fixed<'d, T> {
output: T,
now: Moment<'d>,
}
impl<'d, T> Fixed<'d, T> {
pub(crate) const fn new(output: T, now: Moment<'d>) -> Self {
Self { output, now }
}
pub const fn output(&self) -> &T {
&self.output
}
pub fn into_output(self) -> T {
self.output
}
pub const fn now(&self) -> Moment<'d> {
self.now
}
}
const _: () = {
assert!(mem::size_of::<Fixed<'_, ()>>() == mem::size_of::<Moment<'_>>());
assert!(mem::align_of::<Fixed<'_, ()>>() == mem::align_of::<Moment<'_>>());
};
pub(crate) struct Block<'d, T, C> {
state: T,
source: C,
slot: ready::Slot<'d, routes::Wake>,
}
enum Attempt<R, O> {
Complete(Result<O, drive::Error>),
Cancel(R, drive::Error),
}
enum Outcome<O, P> {
Complete(O, P),
Shutdown(Option<O>),
}
impl<O, P> Outcome<O, P> {
fn complete(self) -> Result<(O, P), drive::Error> {
match self {
Self::Complete(output, proof) => Ok((output, proof)),
Self::Shutdown(_) => Err(drive::Error::Shutdown),
}
}
}
impl<R, O> Attempt<R, O> {
fn finish<'d>(self) -> Result<O, drive::Error>
where
R: boot::Root<'d, Output = O>,
{
match self {
Self::Complete(result) => result,
Self::Cancel(root, source) => {
R::cancel(root);
Err(source)
}
}
}
}
impl<'run, 'driver, 'd, C> Block<'d, scope::Boot<'run, 'driver, 'd>, C>
where
C: clock::Source,
{
pub(crate) fn run_boot<R>(
start: Result<Self, drive::Error>,
root: R,
) -> Result<R::Output, drive::Error>
where
R: boot::Root<'d>,
{
match start {
Ok(block) => block.poll_boot(root),
Err(source) => Attempt::Cancel(root, source),
}
.finish()
}
fn poll_boot<R>(mut self, mut root: R) -> Attempt<R, R::Output>
where
R: boot::Root<'d>,
{
let wake = self.slot.target();
let (mut polled, output) = {
let mut cause = boot::Cause::Start;
loop {
let tick = self.source.tick();
let ready = match scope::Roots::next_root(&mut self.state, &self.slot, tick) {
Ok(ready) => ready,
Err(source) => return Attempt::Cancel(root, source),
};
let polled = if matches!(cause, boot::Cause::Start) || ready.is_root() {
let (polled, output) = ready.poll_boot(root, wake, cause);
match output {
boot::Poll::Pending(next) => {
root = next;
polled
}
boot::Poll::Ready(output) => break (polled, output),
}
} else {
ready.skip()
};
cause = boot::Cause::Wake;
let parked = polled.park();
match parked.status() {
turn::Status::Active => {}
turn::Status::Shutdown => return Attempt::Cancel(root, drive::Error::Shutdown),
}
let resolved = parked.resolve(self.source, |idle| {
let state = idle.progress();
Ok(turn::Resolved::Wait(idle, state))
});
if let Err(source) = resolved {
return Attempt::Cancel(root, source);
}
}
};
Attempt::Complete(loop {
let parked = polled.park();
match parked.status() {
turn::Status::Active => {}
turn::Status::Shutdown => break Err(drive::Error::Shutdown),
}
let drained = match parked.resolve(self.source, |idle| {
let state = idle.progress();
if !matches!(state, schedule::Progress::Quiescent) {
return Ok(turn::Resolved::Wait(idle, state));
}
Ok(match idle.sync()? {
poll::Commit::Drained => turn::Resolved::Done,
poll::Commit::Pending => turn::Resolved::Again,
})
}) {
Ok(drained) => drained,
Err(error) => break Err(error),
};
if drained {
break Ok(output);
}
let tick = self.source.tick();
polled = match self.state.next_root(&self.slot, tick) {
Ok(ready) => ready.skip(),
Err(error) => break Err(error),
};
})
}
}
impl<'d, T, C> Block<'d, T, C>
where
T: scope::Roots<'d>,
{
pub(crate) fn new(state: T, source: C) -> Result<Self, drive::Error> {
let driver = state.driver_ref();
let target = driver
.targets::<routes::Wake>()
.bind(indices::SlotIndex::ZERO, route::Epoch::INITIAL);
let slot = driver.ready().make_slot(target.dispatch())?;
Ok(Self {
state,
source,
slot,
})
}
}
impl<'run, 'driver, 'app, 'd: 'app, D, S, C>
Block<'d, scope::Scope<'run, 'driver, 'app, 'd, D, S>, C>
where
D: executor::raw::App<'d>,
C: clock::Time,
{
pub(crate) fn run<R>(
self,
root: R,
tick: impl Fn(C) -> Result<Sample, drive::Error>,
) -> Result<R::Output, drive::Error>
where
R: root::Root<'d>,
{
let (output, ()) = self
.drive(root, tick, |idle| {
Ok(matches!(idle.sync()?, poll::Commit::Drained).then_some(()))
})?
.complete()?;
Ok(output)
}
pub(crate) fn run_until<R>(
self,
root: R,
tick: impl Fn(C) -> Result<Sample, drive::Error>,
) -> Result<ControlFlow<R::Output>, drive::Error>
where
R: root::Root<'d>,
{
Ok(
match self.drive(root, tick, |idle| {
Ok(matches!(idle.sync()?, poll::Commit::Drained).then_some(()))
})? {
Outcome::Complete(output, ()) | Outcome::Shutdown(Some(output)) => {
ControlFlow::Break(output)
}
Outcome::Shutdown(None) => ControlFlow::Continue(()),
},
)
}
pub(crate) fn run_fixed<R>(
self,
root: R,
tick: impl Fn(C) -> Result<Sample, drive::Error>,
) -> Result<Fixed<'d, R::Output>, drive::Error>
where
R: root::Root<'d>,
{
let (output, now) = self
.drive(root, tick, |idle| {
let fixed = idle.is_fixed();
let now = idle.now();
let commit = idle.sync()?;
Ok((fixed && matches!(commit, poll::Commit::Drained)).then_some(now))
})?
.complete()?;
Ok(Fixed::new(output, now))
}
fn drive<R, P>(
mut self,
root: R,
tick: impl Fn(C) -> Result<Sample, drive::Error>,
complete: impl for<'turn> Fn(
turn::Idle<'turn, 'd, modes::App<'turn, 'app, 'd, D, S>>,
) -> Result<Option<P>, drive::Error>,
) -> Result<Outcome<R::Output, P>, drive::Error>
where
R: root::Root<'d>,
{
let wake = self.slot.target();
let (mut polled, output) = {
let mut root = pin::pin!(root);
let mut first = true;
loop {
let tick = tick(self.source)?;
let ready = self.state.next_root(&self.slot, tick)?;
let poll = first || ready.is_root();
let (polled, output) = if poll {
ready.poll(root.as_mut(), wake)
} else {
(ready.skip(), task::Poll::Pending)
};
first = false;
match output {
task::Poll::Ready(output) => break (polled, output),
task::Poll::Pending => {
let parked = polled.park();
match parked.status() {
turn::Status::Active => {}
turn::Status::Shutdown => return Ok(Outcome::Shutdown(None)),
}
parked.resolve(self.source, |idle| {
let state = idle.progress();
Ok(turn::Resolved::Wait(idle, state))
})?;
}
}
}
};
loop {
let parked = polled.park();
match parked.status() {
turn::Status::Active => {}
turn::Status::Shutdown => return Ok(Outcome::Shutdown(Some(output))),
}
if let turn::Sync::Idle(idle) = parked.sync()?
&& let Some(proof) = complete(idle)?
{
return Ok(Outcome::Complete(output, proof));
}
let tick = tick(self.source)?;
polled = self.state.next_root(&self.slot, tick)?.skip();
}
}
}