use std::{convert, mem, pin};
use dope_core::driver::{receive::Stale, retained, route::Identity, schedule};
use dope_net::{
link::{
event,
pool::{self, recv::arrival},
},
wire::receive::{
Chunk,
mode::{Disabled, access},
},
};
use crate::app::receive::{End, Failure, Outcome};
pub(crate) mod batch;
mod batched;
mod dispatch;
mod policy;
mod retain;
mod scope;
mod single;
mod strategy;
pub(crate) use batched::Batched;
pub(crate) use dispatch::Dispatch;
pub(crate) use policy::Policy;
pub(crate) use retain::Retain;
pub(crate) use scope::Scope;
pub(in crate::app::receive) use single::Single;
pub(crate) use strategy::{Admission, Replay, Strategy};
pub(crate) const IOV_CAP: usize = 32;
#[derive(Clone, Copy)]
pub(crate) enum Empty {
NoChunk,
Discarded { finished: bool },
}
#[derive(Clone, Copy)]
pub(crate) enum Finish {
Empty(Empty),
Chunk,
CloseAfter,
}
pub(crate) enum CloseCause {
Local,
Transport,
Io(dope_core::Errno),
Application(Failure),
Receive(event::ParkError),
}
enum Control {
Eof,
Cancelled,
Empty(Empty),
}
pub(crate) enum Progress<P> {
Done(Outcome),
Pending(P),
}
pub(crate) enum Terminal {
Idle,
Pending,
Ended(End),
Closed,
}
const _: () = assert!(mem::size_of::<Progress<convert::Infallible>>() == mem::size_of::<Outcome>());
fn prepare_terminal<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P: Policy<'p, 'w, 'd, ID>>(
mut policy: pin::Pin<&mut P>,
key: pool::Key<'d, ID>,
) {
let send = P::storage(policy.as_mut())
.get_mut(key)
.is_some_and(|slot| slot.terminal.has_send());
if send {
P::latch(policy, key);
}
}
fn finish_eof<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P: Policy<'p, 'w, 'd, ID>>(
mut policy: pin::Pin<&mut P>,
key: pool::Key<'d, ID>,
end: End,
turn: schedule::Turn<'_, 'd>,
driver: &mut retained::Context<'_, '_, 'd>,
) {
let dead = P::storage(policy.as_mut())
.get_mut(key)
.is_some_and(|slot| slot.is_discarded());
if dead {
P::close(policy, key, CloseCause::Transport, turn, driver);
} else {
P::eof(policy.as_mut(), key, end, turn, driver);
prepare_terminal(policy, key);
}
}
fn discard_transport<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P: Policy<'p, 'w, 'd, ID>>(
mut policy: pin::Pin<&mut P>,
key: pool::Key<'d, ID>,
) {
if let Some(slot) = P::storage(policy.as_mut()).get_mut(key) {
slot.discard();
}
}
fn discard_with_cause<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P: Policy<'p, 'w, 'd, ID>>(
mut policy: pin::Pin<&mut P>,
key: pool::Key<'d, ID>,
err: CloseCause,
turn: schedule::Turn<'_, 'd>,
driver: &mut retained::Context<'_, '_, 'd>,
) {
prepare_terminal(policy.as_mut(), key);
match P::terminal(policy.as_mut(), key, turn.reborrow(), driver) {
Terminal::Closed | Terminal::Pending => {}
Terminal::Ended(_) | Terminal::Idle => {
discard_transport(policy.as_mut(), key);
P::close(policy, key, err, turn, driver);
}
}
}
fn finish_chunk<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P: Policy<'p, 'w, 'd, ID>>(
mut policy: pin::Pin<&mut P>,
key: pool::Key<'d, ID>,
done: Outcome,
turn: schedule::Turn<'_, 'd>,
driver: &mut retained::Context<'_, '_, 'd>,
) {
match done {
done @ (Outcome::Continue | Outcome::CloseAfter) => {
match P::terminal(policy.as_mut(), key, turn.reborrow(), driver) {
Terminal::Ended(end) => {
finish_eof(policy, key, end, turn, driver);
}
Terminal::Closed => {}
Terminal::Idle if matches!(done, Outcome::CloseAfter) => {
P::storage(policy.as_mut()).recv().close_after(key);
P::finish(policy, key, Finish::CloseAfter, turn, driver);
}
Terminal::Pending | Terminal::Idle => {
P::finish(policy, key, Finish::Chunk, turn, driver);
}
}
}
Outcome::Failed(err) => {
discard_with_cause(policy, key, CloseCause::Application(err), turn, driver);
}
}
}
fn drive_control<'p, 'a, 'c, 'owner, 'w, 'd, ID: Identity + 'd, P, A>(
mut policy: pin::Pin<&mut P>,
ctrl: A,
turn: schedule::Turn<'a, 'd>,
driver: &'a mut retained::Context<'c, 'owner, 'd>,
) where
'd: 'owner + 'w + 'p,
P: Policy<'p, 'w, 'd, ID>,
A: arrival::Arrival<'d, Observed = event::Control<'d>>,
{
let next = P::storage(policy.as_mut()).recv().dispatch(ctrl);
let next = match next {
event::Dispatch::Settled => return,
event::Dispatch::Ready(next) => next,
event::Dispatch::Observed {
completion: done,
next,
} => {
driver.driver().reclaim(done);
next
}
event::Dispatch::Retired(done)
| event::Dispatch::Stale {
completion: done,
cause: Stale,
} => {
driver.driver().reclaim(done);
return;
}
event::Dispatch::Rejected {
completion: done,
key,
error,
} => {
driver.driver().reclaim(done);
discard_with_cause(policy, key, CloseCause::Receive(error), turn, driver);
return;
}
event::Dispatch::Parked(parked) => {
apply_parked(policy, parked, turn, driver);
return;
}
};
apply_control(policy, next, turn, driver);
}
fn discard_pending<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P>(
policy: pin::Pin<&mut P>,
key: pool::Key<'d, ID>,
mut pending: Chunk<'w, 'd, P::Wire, Disabled>,
cause: event::ParkError,
turn: schedule::Turn<'_, 'd>,
driver: &mut retained::Context<'_, '_, 'd>,
) where
P: Policy<'p, 'w, 'd, ID>,
{
pending.drive((), &mut access::Discard);
drop(pending);
discard_with_cause(policy, key, CloseCause::Receive(cause), turn, driver);
}
fn apply_control<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P>(
mut policy: pin::Pin<&mut P>,
next: event::ControlRecv<'d, ID>,
turn: schedule::Turn<'_, 'd>,
driver: &mut retained::Context<'_, '_, 'd>,
) where
P: Policy<'p, 'w, 'd, ID>,
{
let (key, end) = match next {
event::ControlRecv::Eof(key) => (key, Control::Eof),
event::ControlRecv::Cancelled(key) => (key, Control::Cancelled),
event::ControlRecv::Failed { key, error } => {
return P::close(policy, key, CloseCause::Io(error), turn, driver);
}
event::ControlRecv::Credit { key, cause } => {
return P::close(
policy,
key,
CloseCause::Receive(event::ParkError::Credit(cause)),
turn,
driver,
);
}
event::ControlRecv::NoChunk(key) => (key, Control::Empty(Empty::NoChunk)),
event::ControlRecv::Discarded { key, finished } => {
(key, Control::Empty(Empty::Discarded { finished }))
}
};
prepare_terminal(policy.as_mut(), key);
match P::terminal(policy.as_mut(), key, turn.reborrow(), driver) {
Terminal::Ended(end) => {
finish_eof(policy, key, end, turn.reborrow(), driver);
}
Terminal::Closed | Terminal::Pending => {}
Terminal::Idle => match end {
Control::Eof => {
let notify = P::storage(policy.as_mut())
.get_mut(key)
.is_some_and(|slot| !slot.is_discarded() && !slot.terminal.is_done());
if notify {
P::eof(policy, key, End::Read, turn.reborrow(), driver);
} else {
P::close(policy, key, CloseCause::Transport, turn.reborrow(), driver);
}
}
Control::Cancelled => {
P::close(policy, key, CloseCause::Local, turn.reborrow(), driver);
}
Control::Empty(empty) => {
P::finish(policy, key, Finish::Empty(empty), turn.reborrow(), driver);
}
},
}
}
fn apply_parked<'p, 'w, 'd: 'w + 'p, ID: Identity + 'd, P>(
policy: pin::Pin<&mut P>,
parked: event::ParkRecv<'d, ID>,
turn: schedule::Turn<'_, 'd>,
driver: &mut retained::Context<'_, '_, 'd>,
) where
P: Policy<'p, 'w, 'd, ID>,
{
match parked {
event::ParkRecv::Closing(key) => {
P::close(policy, key, CloseCause::Transport, turn, driver);
}
event::ParkRecv::Failed { key, error } => {
discard_with_cause(policy, key, CloseCause::Receive(error), turn, driver);
}
event::ParkRecv::Deferred => {}
}
}