use std::{
array,
mem::{self, MaybeUninit},
ops::ControlFlow,
pin::Pin,
task::Poll,
};
use cartel_core::{deadline, reply};
use dope::{
core::driver::{route::KeyTag, schedule::clock::timer},
manifold::client::conn,
};
use dope_fiber::{abi, context, wait};
use o3::{base::cell::region, depot::array::init};
use reply::{
Slot,
recv::{
self,
attach::{self, Attach},
outcome::Terminal,
},
};
use crate::{
api::{
self, Response,
results::{
GroupFailure,
group::{Outcome, Sync},
},
},
port::replies::{self, groups},
reply::{Error, group::Ledger},
schema,
};
use crate::api::task::{State, lease};
type Output<'r, 's, 'b, 'c, 'd, const ID: u8, const N: usize> =
GroupFailure<Option<replies::Payload<'b, 'c, 'd>>, Error<'d, ID>, Ledger<'r, 's>, N>;
struct First<const ID: u8>;
impl<'app, 'b, 'c: 'b, 'd: 'c, const ID: u8>
recv::Extract<'app, 'b, 'c, 'd, replies::Row<'d>, replies::ReplyFailure<'d>> for First<ID>
{
type Output = Outcome<Option<replies::Payload<'b, 'c, 'd>>, Error<'d, ID>>;
type Context = ();
const ACCEPTANCE: recv::Acceptance = recv::Acceptance::First;
fn extract(
_: &(),
slot: &mut Slot<'_, 'app, 'b, 'c, 'd, replies::Row<'d>, replies::ReplyFailure<'d>>,
) -> Option<Self::Output> {
Some(match slot.terminal()? {
Terminal::Complete(mut done) => match done.try_node() {
Ok(payload) => Outcome::Completed(payload.map(replies::Payload::from_node)),
Err(error) => Outcome::Failed(crate::Error::ResponseRetain(error).into()),
},
Terminal::Failed(failure) => match failure.into_error() {
replies::ReplyFailure::Server(error) => Outcome::Failed(Error::Server(error)),
replies::ReplyFailure::GroupServer => Outcome::Server,
replies::ReplyFailure::Failed(crate::Error::Skipped) => Outcome::Skipped,
replies::ReplyFailure::Failed(error) => Outcome::Interrupted(error.into()),
},
})
}
}
type Admit<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, const ID: u8, const N: usize> = State<
'app,
'g,
'r,
's,
'b,
'c,
'd,
B,
I,
ID,
Ready<'g, 'r, 's, 'b, 'c, 'd, N>,
conn::Id<'d, KeyTag<ID>>,
lease::Detached,
groups::Owner<'g, 'r, 's, 'b, 'c, 'd>,
>;
enum Stop<'d, const ID: u8> {
Deadline(crate::Error),
Admission(Error<'d, ID>),
Sync,
}
struct Completed<'r, 's, 'd, const ID: u8> {
diagnostics: Ledger<'r, 's>,
sync: Sync,
sync_error: Option<Error<'d, ID>>,
cause: Option<Error<'d, ID>>,
}
fn collect<'app, 'g, 'r, 's, 'b, 'c: 'b, 'd: 'c, const ID: u8, const N: usize>(
commands: &mut [Response<'app, 'g, 'r, 's, 'b, 'c, 'd, First<ID>>; N],
token: &mut region::Token<region::Identity<'d>>,
) -> [Outcome<Option<replies::Payload<'b, 'c, 'd>>, Error<'d, ID>>; N] {
commands
.each_mut()
.map(|command| match command.try_take(token) {
Some(outcome) => outcome,
None => {
command.detach();
Outcome::Unobserved
}
})
}
fn complete<
'app,
'g,
'r,
's,
'b,
'c: 'b,
'd: 'c + 'app,
B: ?Sized,
I: schema::QuerySet + 'd,
const ID: u8,
const N: usize,
>(
admit: &mut Admit<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID, N>,
reply: &mut Response<'app, 'g, 'r, 's, 'b, 'c, 'd, api::ExtractUnit<ID>>,
guard: Pin<&mut Result<deadline::Guard<'d>, timer::Exhausted>>,
waiter: Pin<&wait::Waiter<'app, 'd>>,
stop: Stop<'d, ID>,
sync: Option<Result<(), Error<'d, ID>>>,
) -> Completed<'r, 's, 'd, ID> {
waiter.unregister();
let (mut cause, abort) = match stop {
Stop::Deadline(error) => (None, Some(error)),
Stop::Admission(error) => (Some(error), None),
Stop::Sync => (None, None),
};
let (sync, failure) = match sync {
Some(Ok(())) => (Sync::Complete, None),
Some(Err(error)) => (Sync::Unobserved, Some(error)),
None => (Sync::Unobserved, None),
};
let ledger = match &*admit {
State::Admitted(_, owner) => owner.snapshot(),
_ => Ledger::empty(),
};
reply.detach();
match abort {
Some(error) => cause = Some(admit.cancel(error)),
None => {
admit.take();
if let Ok(guard) = guard.get_mut() {
guard.complete();
}
}
}
Completed {
diagnostics: ledger,
sync,
sync_error: failure,
cause,
}
}
struct Replies<'app, 'g, 'r, 's, 'b, 'c: 'b, 'd: 'c, const ID: u8, const N: usize> {
commands: [Response<'app, 'g, 'r, 's, 'b, 'c, 'd, First<ID>>; N],
sync: Response<'app, 'g, 'r, 's, 'b, 'c, 'd, api::ExtractUnit<ID>>,
}
impl<'app, 'g, 'r, 's, 'b, 'c: 'b, 'd: 'c, const ID: u8, const N: usize>
Replies<'app, 'g, 'r, 's, 'b, 'c, 'd, ID, N>
{
fn new() -> Self {
const { assert!(N > 0 && N < u32::MAX as usize) };
Self {
commands: array::from_fn(|_| Response::new()),
sync: Response::new(),
}
}
}
pub(in crate::api) struct Ready<'g, 'r, 's, 'b, 'c, 'd, const N: usize> {
collector: groups::Draft<'g, 'r, 's, 'b, 'c, 'd>,
commands: [groups::Owner<'g, 'r, 's, 'b, 'c, 'd>; N],
sync: groups::Owner<'g, 'r, 's, 'b, 'c, 'd>,
}
impl<'g, 'r, 's, 'b, 'c, 'd, const N: usize> Ready<'g, 'r, 's, 'b, 'c, 'd, N> {
pub(in crate::api) fn new(
access: &groups::Access<'g, 'r, 's, 'b, 'c, 'd>,
) -> Result<Self, crate::Error> {
let owner = access.prepare()?;
let mut storage = MaybeUninit::uninit();
let mut guard = init::Guard::<_, N>::new(&mut storage);
let commands = loop {
match guard.next() {
ControlFlow::Break(full) => break full.into_array(),
ControlFlow::Continue(vacant) => {
let item = owner.try_clone().map_err(crate::Error::ResponseRetain)?;
guard = vacant.insert(item);
}
}
};
let sync = owner.try_clone().map_err(crate::Error::ResponseRetain)?;
Ok(Self {
collector: owner,
commands,
sync,
})
}
}
#[pin_project::pin_project]
pub(in crate::api) struct Group<
'app,
'g,
'r,
's,
'b,
'c: 'b,
'd: 'c + 'app,
B: ?Sized,
I: schema::QuerySet + 'd,
const ID: u8,
const N: usize,
> {
client: api::Client<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID>,
admission: Admit<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID, N>,
reply: Replies<'app, 'g, 'r, 's, 'b, 'c, 'd, ID, N>,
#[pin]
deadline: Result<deadline::Guard<'d>, timer::Exhausted>,
#[pin]
waiter: wait::Waiter<'app, 'd>,
}
impl<
'app,
'g,
'r,
's,
'b,
'c: 'b,
'd: 'c + 'app,
B: ?Sized,
I: schema::QuerySet + 'd,
const ID: u8,
const N: usize,
> Group<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID, N>
{
pub(in crate::api) fn new(
client: api::Client<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID>,
target: Option<api::Target<'d, ID>>,
next: Result<api::Pending<'d, Ready<'g, 'r, 's, 'b, 'c, 'd, N>>, crate::Error>,
) -> Self {
let admit = State::new(target, next);
let deadline = client.port.lifecycle.deadline.guard();
Self {
client,
admission: admit,
reply: Replies::new(),
deadline,
waiter: wait::Waiter::new(),
}
}
}
impl<
'app,
'g,
'r,
's,
'b,
'c: 'b,
'd: 'c + 'app,
B: ?Sized,
I: schema::QuerySet + 'd,
const ID: u8,
const N: usize,
> abi::Fiber<'d> for Group<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID, N>
{
type Output = Output<'r, 's, 'b, 'c, 'd, ID, N>;
fn poll(call: context::PollCall<'_, '_, 'd, Self>) -> Poll<Self::Output> {
let (this, mut cx) = call.into_parts();
let mut this = this.project();
let mut sync = None;
let expired = match this.deadline.as_mut().get_mut() {
Ok(guard) => match Pin::new(guard).poll_expired(cx.as_mut(), this.admission.phase()) {
Poll::Ready(time) => Some(Stop::Deadline(crate::Error::RequestTimeout(time))),
Poll::Pending => None,
},
Err(error) => Some(Stop::Deadline(crate::Error::Deadline(*error))),
};
let stop = match expired {
Some(stop) => stop,
None => {
match this.admission.poll_with(
this.client,
cx.as_mut(),
this.waiter.as_ref(),
|admit, client, weight, context| {
client
.port
.request()
.try_with(
context,
weight,
admit,
|token, lane, tag, admit| {
let State::Connecting { request, .. } = &*admit else {
unreachable!("connecting");
};
let reserved =
match lane.try_reserve(request.state.collector.usage()) {
Ok(reserved) => reserved,
Err(error) => {
return Err((attach::Error::Payload(error), admit));
}
};
Response::try_group_by(
token,
lane,
&mut this.reply.commands,
&mut this.reply.sync,
(admit, reserved),
|(admit, reserved)| match mem::replace(admit, State::Done) {
State::Connecting { request, .. } => {
let api::Pending {
frame,
state:
Ready {
collector,
commands,
sync,
},
..
} = request;
let owner = collector.admit(reserved);
let mut index = 0u32;
let commands = commands.map(|owner| {
let position = index;
index += 1;
replies::Tag::command(
tag.batch(),
position,
index as usize == N,
owner,
)
});
(
commands,
replies::Tag::sync(tag.batch(), sync),
(admit, frame, owner),
)
}
_ => unreachable!("connecting"),
},
)
.map_err(|(error, (admit, _))| (error, admit))
},
|(admit, frame, owner), conn| {
*admit = State::Admitted(
lease::Lease::admit((), client, conn),
owner,
);
(frame, ())
},
)
.map_err(|(error, conn, _)| (error, conn))
},
) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Err(error)) => Stop::Admission(error),
Poll::Ready(Ok(())) => {
let Poll::Ready(observed) =
Pin::new(&mut this.reply.sync).poll_reply(cx.as_mut())
else {
return Poll::Pending;
};
sync = Some(observed);
Stop::Sync
}
}
}
};
let token = cx.region_token();
let sync = sync.or_else(|| this.reply.sync.try_take(token));
let outcomes = collect(&mut this.reply.commands, token);
let done = complete(
this.admission,
&mut this.reply.sync,
this.deadline,
this.waiter.as_ref(),
stop,
sync,
);
Poll::Ready(GroupFailure {
outcomes,
diagnostics: done.diagnostics,
sync: done.sync,
sync_error: done.sync_error,
cause: done.cause,
})
}
}