dope-core 0.28.0

use std::mem;

use o3::base::cell::region;

use crate::{
    driver,
    driver::{
        life::finish,
        result, retained,
        retained::{flight, flight::cancel},
        route,
        route::{indices, kind},
        schedule,
    },
    io::{event::batch::Flow, fd::handles},
};

#[derive(Clone, Copy, Eq, PartialEq)]
enum Phase {
    Ready,
    Armed,
    Deferred,
    CancelDeferred,
    Cancelling,
    Idle,
    Waiting,
    Probing,
    ProbeDeferred,
}

pub struct State<'d, Tag: route::Tag> {
    epoch: route::Epoch,
    slot: indices::SlotIndex,
    driver: route::Brand<'d, Tag>,
    phase: Phase,
    flight: Option<flight::Flight<'d, Tag::Class>>,
}

impl<'d, Tag: route::Tag> State<'d, Tag> {
    pub fn new(target: route::Target<'d, Tag>) -> Self {
        Self {
            epoch: target.epoch(),
            slot: target.slot(),
            driver: target.driver,
            phase: Phase::Ready,
            flight: None,
        }
    }

    pub fn progress<'a>(
        &self,
        region: &region::Token<region::Identity<'a>>,
    ) -> schedule::Progress<'a> {
        match self.phase {
            Phase::Ready | Phase::ProbeDeferred | Phase::CancelDeferred => {
                schedule::Progress::Runnable
            }
            Phase::Deferred
            | Phase::Armed
            | Phase::Probing
            | Phase::Cancelling
            | Phase::Waiting => schedule::Progress::waiting(region),
            Phase::Idle => schedule::Progress::Quiescent,
        }
    }

    /// Each caller retains an exclusive arming loan, so its abandoned outcome
    /// can be installed before native submission without a transient phase.
    pub(super) fn start(&mut self) -> bool {
        if !matches!(self.phase, Phase::Ready | Phase::Deferred) {
            return false;
        }
        self.phase = Phase::Ready;
        true
    }

    pub(super) fn operation(&self) -> route::Operation<'d, Tag> {
        route::Space {
            driver: self.driver,
        }
        .bind(self.slot, self.epoch)
        .dispatch()
    }

    pub(super) fn apply(
        &mut self,
        flight: Result<flight::Flight<'d, Tag::Class>, result::SubmitError>,
    ) -> Submission {
        let probing = self.phase == Phase::ProbeDeferred;
        match flight {
            Ok(flight) => {
                self.flight = Some(flight);
                self.phase = if probing {
                    Phase::Probing
                } else {
                    Phase::Armed
                };
                Submission::Armed
            }
            Err(error) => match error.cause() {
                result::SubmitCause::Full => {
                    self.phase = if probing {
                        Phase::ProbeDeferred
                    } else {
                        Phase::Deferred
                    };
                    Submission::Deferred
                }
                result::SubmitCause::IdentityExhausted => {
                    self.phase = Phase::Idle;
                    Submission::Failed(error)
                }
            },
        }
    }

    pub fn stop(&mut self, driver: &mut driver::Context<'_, 'd>) {
        match self.phase {
            Phase::Ready | Phase::Deferred | Phase::Waiting | Phase::ProbeDeferred => {
                self.phase = Phase::Idle
            }
            Phase::Armed | Phase::Probing | Phase::CancelDeferred => self.cancel(driver),
            Phase::Cancelling | Phase::Idle => {}
        }
    }

    pub fn is_live(&self) -> bool {
        !matches!(
            self.phase,
            Phase::Ready | Phase::Deferred | Phase::ProbeDeferred | Phase::Idle
        )
    }

    pub fn finish(&mut self, proof: &mut finish::Finish<'_, 'd>) {
        if let Some(flight) = self.flight.take() {
            flight.finish(proof);
        }
        self.phase = Phase::Idle;
    }

    fn cancel(&mut self, driver: &mut driver::Context<'_, 'd>) {
        let Some(flight) = self.flight.as_mut() else {
            unreachable!("flight");
        };
        match flight.try_stop(driver) {
            Ok(
                cancel::Outcome::Submitted | cancel::Outcome::Queued | cancel::Outcome::Completed,
            ) => {
                self.phase = Phase::Cancelling;
            }
            Ok(cancel::Outcome::Deferred) => self.phase = Phase::CancelDeferred,
            Err(failure) => {
                self.phase = Phase::Cancelling;
                driver.rejected(failure);
            }
        }
    }
}

#[must_use]
#[derive(Clone, Copy, Eq, PartialEq)]
pub enum Disposition {
    Deliver,
    Discard,
    Probe,
    Retiring,
}

#[must_use]
#[derive(Clone, Copy, Eq, PartialEq)]
pub enum Submission {
    Armed,
    Deferred,
    /// Admission stopped permanently without submitting a native operation.
    Failed(result::SubmitError),
}

#[derive(Clone, Copy, Eq, PartialEq)]
pub enum Resume {
    Resumed,
    Unchanged,
}

impl<'d, Tag: route::Tag> State<'d, Tag> {
    pub fn begin(&mut self) -> Option<Arming<'_, 'd, Tag>> {
        self.start().then_some(Arming { state: self })
    }

    pub fn begin_retry(&mut self) -> Option<Arming<'_, 'd, Tag>> {
        if !matches!(
            self.phase,
            Phase::Ready | Phase::Deferred | Phase::ProbeDeferred
        ) {
            return None;
        }
        self.phase = Phase::ProbeDeferred;
        Some(Arming { state: self })
    }

    pub fn wait(&mut self) {
        if self.phase == Phase::Ready {
            self.phase = Phase::Waiting;
        }
    }

    pub fn resume(&mut self) {
        if self.phase == Phase::Waiting {
            self.phase = Phase::Ready;
        }
    }

    /// Restarts a stopped operation only after its native flight has retired.
    pub fn restart(&mut self) {
        if self.phase == Phase::Idle {
            self.phase = Phase::Ready;
        }
    }

    pub fn try_resume(&mut self) -> Resume {
        if self.phase != Phase::Waiting {
            return Resume::Unchanged;
        }
        self.phase = Phase::Ready;
        Resume::Resumed
    }

    pub fn settle_probe(&mut self) {
        if self.phase == Phase::Probing {
            self.phase = Phase::Armed;
        }
    }

    pub fn complete(&mut self, token: route::Erased<'d, Tag::Class>, flow: Flow) -> Disposition {
        if !self
            .flight
            .as_ref()
            .is_some_and(|flight| flight.matches(token))
        {
            return Disposition::Discard;
        }
        let (phase, result) = match self.phase {
            Phase::Armed => (Phase::Ready, Disposition::Deliver),
            Phase::Probing => (Phase::Ready, Disposition::Probe),
            Phase::CancelDeferred | Phase::Cancelling => (Phase::Idle, Disposition::Retiring),
            Phase::Ready
            | Phase::Deferred
            | Phase::Idle
            | Phase::Waiting
            | Phase::ProbeDeferred => return Disposition::Discard,
        };
        if flow == Flow::Last {
            let Some(flight) = self.flight.take_if(|flight| !flight.is_pending()) else {
                return Disposition::Discard;
            };
            drop(flight);
            self.phase = phase;
        }
        result
    }
}

#[must_use]
pub struct Arming<'a, 'd, Tag: route::Tag> {
    state: &'a mut State<'d, Tag>,
}

impl<'d, Tag: route::Tag> Arming<'_, 'd, Tag> {
    pub fn accept(
        self,
        driver: &mut driver::Context<'_, 'd>,
        slots: &flight::Slots<'d, Tag>,
        fd: &handles::Descriptor<'d>,
    ) -> Submission
    where
        Tag: route::Tag<Class = kind::Accept>,
    {
        if !matches!(
            retained::accept::Readiness::from(driver.driver_ref()),
            retained::accept::Readiness::Ready
        ) {
            return Submission::Deferred;
        }
        let flight = slots.accept(driver.reborrow(), fd, self.state.operation());
        self.state.apply(flight)
    }

    pub fn recv(
        self,
        driver: &mut driver::Context<'_, 'd>,
        slots: &flight::Slots<'d, Tag>,
        fd: &handles::Datagram<'d>,
    ) -> Submission
    where
        Tag: route::Tag<Class = kind::Recv>,
    {
        let flight = slots.recv_dgram(driver.reborrow(), fd, self.state.operation());
        self.state.apply(flight)
    }
}

const _: () = {
    type Route = route::KeyTag<1, kind::Recv>;
    assert!(mem::size_of::<Phase>() == 1);
    assert!(mem::size_of::<State<'static, Route>>() == 24);
    assert!(mem::size_of::<Option<State<'static, Route>>>() == 24);
    assert!(mem::align_of::<State<'static, Route>>() == mem::align_of::<route::Epoch>());
    assert!(
        mem::size_of::<State<'static, Route>>()
            == mem::size_of::<(
                route::Operation<'static, Route>,
                Option<flight::Flight<'static>>,
            )>()
    );
    assert!(mem::size_of::<Arming<'static, 'static, Route>>() == mem::size_of::<usize>());
    assert!(mem::size_of::<Submission>() == 1);
};