use super::super::Pool;
use super::super::slot::WRITE_ROOM_MIN;
use super::super::slot::{BufSlot, Conn, ConnFlags, CoreSlot};
use super::Server;
use super::state::StreamState;
use super::state::{ServerSlotState, SlotEgress, SlotIngress};
use super::wake::WakeQueue;
use crate::CodecLayer;
use crate::SlotId;
use crate::WriteBufStorage;
use crate::protocol::server::{
HandlerAction, ParkAction, PendingFuture, PendingResponse, RequestBodyStream,
RequestStreamAction, RouteRequestBodyKind, ServerProtocol,
};
use dambi::BytesMut;
use dope_core::driver::{DriverOps, PoolDriver, ProvidedRingOps};
use dope_core::profile::Profile;
use dope_transport::Transport;
type PendingFutureBody = dambi::InlineFuture<'static, PendingResponse, 64>;
struct ParseCtx<'a, H: ServerProtocol> {
protocol: &'a mut H,
conn: &'a mut Conn,
buf: &'a mut BufSlot<crate::WriteBufArr>,
user: &'a mut ServerSlotState<H>,
slot_idx: usize,
}
enum ParseStep {
Continue(usize),
Stop(usize),
}
enum OutcomeStep {
Step(ParseStep),
Drop,
}
impl<'a, H: ServerProtocol> ParseCtx<'a, H> {
#[cold]
fn install_pending(&mut self, future: PendingFutureBody) {
self.conn.flags.set(ConnFlags::PARKED, true);
let id = SlotId::from_parts(self.slot_idx as u32, self.conn.generation);
self.user.pending_future = Some(PendingFuture {
generation: self.conn.generation,
future,
});
WakeQueue::push(id);
}
#[inline(always)]
fn clear_parse_phase(&mut self) {
self.conn.parse_phase = 0;
self.conn.body_needed = 0;
}
#[inline(always)]
fn apply_send(
&mut self,
wh: usize,
written: u16,
close_after: bool,
total: usize,
) -> OutcomeStep {
self.conn.write_head = wh as u16 + written;
if close_after {
self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
}
OutcomeStep::Step(ParseStep::Continue(total))
}
#[inline(always)]
fn apply_close_payload(
&mut self,
wh: usize,
bytes: &'static [u8],
total: usize,
) -> OutcomeStep {
let write_slice = &mut self.buf.write_buf.as_mut_slice()[wh..];
if write_slice.len() < bytes.len() {
return OutcomeStep::Drop;
}
write_slice[..bytes.len()].copy_from_slice(bytes);
self.conn.write_head = wh as u16 + bytes.len() as u16;
self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
OutcomeStep::Step(ParseStep::Continue(total))
}
#[inline(always)]
fn apply_outcome(&mut self, outcome: HandlerAction, wh: usize, total: usize) -> OutcomeStep {
match outcome {
HandlerAction::Send {
written,
close_after,
} => self.apply_send(wh, written, close_after, total),
HandlerAction::SendStatic {
hdr_written,
body,
close_after,
} => {
self.conn.write_head = (wh as u32 + hdr_written).min(u16::MAX as u32) as u16;
self.conn.pending_body_ptr = body.as_ptr();
self.conn.pending_body_len = body.len() as u32;
if close_after {
self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
}
OutcomeStep::Step(ParseStep::Continue(total))
}
HandlerAction::SendStream {
hdr_written,
stream,
close_after,
} => {
self.conn.write_head = (wh as u32 + hdr_written).min(u16::MAX as u32) as u16;
self.user.egress = SlotEgress::Stream(Box::new(StreamState::new(stream)));
if close_after {
self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
}
OutcomeStep::Step(ParseStep::Continue(total))
}
HandlerAction::Pending(future) => {
self.install_pending(future);
OutcomeStep::Step(ParseStep::Stop(total))
}
HandlerAction::Close(bytes) => self.apply_close_payload(wh, bytes, total),
HandlerAction::SendVectored {
iovs,
iov_count,
cookie,
total_bytes,
close_after,
} => {
let conn = &mut *self.conn;
let buf = &mut *self.buf;
if buf.staged_gather_cookie != cookie {
let n = (iov_count as usize).min(buf.pending_iovs.len());
for (i, iov) in iovs.iter().enumerate().take(n) {
buf.pending_iovs[i] = iov.as_libc();
}
buf.staged_gather_cookie = cookie;
}
conn.large_inflight = total_bytes;
conn.large_tail = 0;
self.user.egress = SlotEgress::Gather { iov_count };
if close_after {
conn.flags.set(ConnFlags::CLOSE_AFTER, true);
}
OutcomeStep::Step(ParseStep::Continue(total))
}
}
}
#[inline(always)]
fn apply_park_outcome(&mut self, outcome: ParkAction, wh: usize, total: usize) -> OutcomeStep {
match outcome {
ParkAction::Send {
written,
close_after,
} => self.apply_send(wh, written, close_after, total),
ParkAction::Park => {
self.conn.flags.set(ConnFlags::PARKED, true);
OutcomeStep::Step(ParseStep::Stop(total))
}
ParkAction::Close(bytes) => self.apply_close_payload(wh, bytes, total),
}
}
#[inline(always)]
fn parse_one(&mut self, view: &[u8]) -> ParseStep {
let Some((head, head_len, body_len)) =
self.protocol.parse(&mut self.user.protocol_state, view)
else {
return ParseStep::Stop(0);
};
let head_len = head_len as usize;
let body_needed: usize = if self.conn.parse_phase == 1 {
self.conn.body_needed as usize
} else {
body_len as usize
};
let route_kind = self.protocol.route_request_body_kind(&head);
if body_needed > 0 && matches!(route_kind, RouteRequestBodyKind::Stream) {
return self.dispatch_stream(view, head, head_len, body_needed);
}
if head_len.saturating_add(body_needed) > self.buf.ingress_buf.len() {
return self.dispatch_oversized(view, head_len, body_needed, route_kind);
}
let total = head_len + body_needed;
if view.len() < total {
self.conn.parse_phase = 1;
self.conn.body_needed = body_needed as u32;
return ParseStep::Stop(0);
}
let wh = self.conn.write_head as usize;
if self.buf.write_buf.as_slice().len().saturating_sub(wh) < WRITE_ROOM_MIN {
return ParseStep::Stop(0);
}
if !self.conn.pending_body_ptr.is_null()
|| !self.user.egress.is_idle()
|| self.conn.flags.contains(ConnFlags::PARKED)
{
return ParseStep::Stop(0);
}
let req_bytes: &[u8] = &view[..total];
let write_slice: &mut [u8] = &mut self.buf.write_buf.as_mut_slice()[wh..];
if H::SUPPORTS_PARK {
let conn_id = SlotId::from_parts(self.slot_idx as u32, self.conn.generation);
let outcome = self.protocol.handle_park(
conn_id,
&mut self.user.protocol_state,
req_bytes,
head,
write_slice,
);
return match self.apply_park_outcome(outcome, wh, total) {
OutcomeStep::Step(step) => {
self.clear_parse_phase();
step
}
OutcomeStep::Drop => ParseStep::Stop(0),
};
}
let outcome = self.protocol.handle(req_bytes, head, write_slice);
match self.apply_outcome(outcome, wh, total) {
OutcomeStep::Step(step) => {
self.clear_parse_phase();
step
}
OutcomeStep::Drop => ParseStep::Stop(0),
}
}
#[cold]
fn dispatch_stream<'v>(
&mut self,
view: &'v [u8],
head: <H as ServerProtocol>::Head<'v>,
head_len: usize,
body_needed: usize,
) -> ParseStep {
if !self.conn.pending_body_ptr.is_null()
|| !self.user.ingress.is_idle()
|| !self.user.egress.is_idle()
{
return ParseStep::Stop(0);
}
let wh = self.conn.write_head as usize;
if self.buf.write_buf.as_slice().len().saturating_sub(wh) < WRITE_ROOM_MIN {
return ParseStep::Stop(0);
}
let (receiver, handle) = RequestBodyStream::new(body_needed as u32);
let pushed = handle.push_slice(&view[head_len..]);
let eof = handle.is_complete();
if !eof {
self.user.ingress = SlotIngress::RequestStream(Box::new(handle));
}
let outcome = {
let head_bytes: &[u8] = &view[..head_len];
let write_slice: &mut [u8] = &mut self.buf.write_buf.as_mut_slice()[wh..];
self.protocol
.handle_request_stream(head_bytes, head, receiver, write_slice)
};
let consumed = head_len + pushed;
self.clear_parse_phase();
match outcome {
RequestStreamAction::Pending(future) => self.install_pending(future),
RequestStreamAction::Close(bytes) => {
let write_slice = &mut self.buf.write_buf.as_mut_slice()[wh..];
let n = bytes.len().min(write_slice.len());
write_slice[..n].copy_from_slice(&bytes[..n]);
self.conn.write_head = wh as u16 + n as u16;
self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
}
}
ParseStep::Stop(consumed)
}
#[cold]
fn dispatch_oversized(
&mut self,
view: &[u8],
head_len: usize,
body_needed: usize,
route_kind: RouteRequestBodyKind,
) -> ParseStep {
if matches!(route_kind, RouteRequestBodyKind::Spilled) {
let total = head_len + body_needed;
let mut spill = BytesMut::with_capacity(total);
let available = view.len().min(total);
spill.extend_from_slice(&view[..available]);
self.user.ingress = SlotIngress::Spill {
buf: Box::new(spill),
filled: available as u32,
target: total as u32,
};
self.clear_parse_phase();
return ParseStep::Stop(available);
}
let wh = self.conn.write_head as usize;
let bytes = self.protocol.oversize_response();
if self.buf.write_buf.as_slice().len().saturating_sub(wh) < bytes.len() {
return ParseStep::Stop(0);
}
self.buf.write_buf.as_mut_slice()[wh..wh + bytes.len()].copy_from_slice(bytes);
self.conn.write_head = wh as u16 + bytes.len() as u16;
self.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
self.clear_parse_phase();
ParseStep::Stop(view.len())
}
}
impl<H: ServerProtocol, T: Transport, F: Profile, B: PoolDriver> super::super::PoolRecv<B>
for Pool<Server<H, T>, F, B>
{
#[inline(always)]
fn on_recv_event(
&mut self,
driver: &mut B::Driver,
token: B::Token,
result: i32,
more: bool,
bid: Option<u16>,
) {
match self.core.classify_recv(driver, token, result, more, bid) {
super::super::core::RecvDispatch::Skipped => {}
super::super::core::RecvDispatch::EofClose { slot_idx } => {
self.core.free_slot(driver, slot_idx);
}
super::super::core::RecvDispatch::Errno {
slot_idx,
errno,
more,
} => {
self.on_recv_errno(driver, slot_idx, errno, more);
}
super::super::core::RecvDispatch::Chunk {
slot_idx,
ptr,
len,
bid,
more,
} => {
let src: &[u8] = unsafe { std::slice::from_raw_parts(ptr, len) };
if <H::CodecLayer as CodecLayer>::IS_PASSTHROUGH {
self.ingest_plaintext(driver, slot_idx, src);
} else {
self.ingest_codec(driver, slot_idx, src);
}
driver.provided_ring().defer(bid);
if !self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::LIVE)
{
return;
}
self.submit_slot_write(driver, slot_idx);
if !more {
let _ = self.core.arm_recv(driver, slot_idx);
}
}
super::super::core::RecvDispatch::DrainOnly { slot_idx, more } => {
if !more {
let _ = self.core.arm_recv(driver, slot_idx);
}
}
}
}
}
impl<H: ServerProtocol, T: Transport, F: Profile, B: PoolDriver> Pool<Server<H, T>, F, B> {
#[cold]
#[inline(never)]
fn on_recv_errno(&mut self, driver: &mut B::Driver, slot_idx: usize, errno: i32, more: bool) {
if errno == libc::EINVAL {
panic!("dope invariant violated: recv-multi unsupported on required uring platform");
}
if errno == libc::ECANCELED {
if !more {
self.core.slots[slot_idx].io.recv.armed = false;
}
return;
}
self.core.slots[slot_idx].io.recv.armed = false;
self.core.free_slot(driver, slot_idx);
}
#[cold]
fn ingest_codec(&mut self, driver: &mut B::Driver, slot_idx: usize, src: &[u8]) {
let Some(plaintext) = self.core.process_codec_inbound(driver, slot_idx, src) else {
return;
};
self.ingest_plaintext(driver, slot_idx, &plaintext);
self.core.install_plaintext(slot_idx, plaintext);
}
fn ingest_plaintext(&mut self, driver: &mut B::Driver, slot_idx: usize, src: &[u8]) {
let ingress_kind = match &self.core.slots[slot_idx].user.get().ingress {
SlotIngress::Spill { .. } => 1u8,
SlotIngress::RequestStream(_) => 2u8,
SlotIngress::Idle => 0u8,
};
match ingress_kind {
1 => {
self.spill_push(slot_idx, src);
return;
}
2 => {
self.request_stream_push(driver, slot_idx, src);
return;
}
_ => {}
}
let head = self.core.slots[slot_idx].conn.ingress_head as usize;
let parked = self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::PARKED);
if head == 0 && !parked {
let consumed = self.parse_and_dispatch_slice(slot_idx, src);
let spill_in_progress = self.core.slots[slot_idx].user.get().ingress.is_spill();
if !spill_in_progress {
let remaining = &src[consumed..];
let rl = remaining.len();
if rl > 0 {
let entry = &mut self.core.slots[slot_idx];
let n = rl.min(entry.buf.ingress_buf.len());
entry.buf.ingress_buf[..n].copy_from_slice(&remaining[..n]);
entry.conn.ingress_head = n as u16;
entry.conn.ingress_tail = 0;
}
}
} else {
{
let entry = &mut self.core.slots[slot_idx];
let head = entry.conn.ingress_head as usize;
let room = entry.buf.ingress_buf.len().saturating_sub(head);
let n = src.len().min(room);
if n > 0 {
entry.buf.ingress_buf[head..head + n].copy_from_slice(&src[..n]);
entry.conn.ingress_head = (head + n) as u16;
}
}
self.parse_and_dispatch(slot_idx);
self.try_dispatch_spilled(slot_idx);
}
}
pub(super) fn parse_and_dispatch_slice(&mut self, slot_idx: usize, src: &[u8]) -> usize {
if self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::PARKED)
{
return 0;
}
let CoreSlot {
conn, buf, user, ..
} = &mut self.core.slots[slot_idx];
let user = user.get_mut();
let mut ctx = ParseCtx {
protocol: &mut self.direction.protocol,
conn,
buf,
user,
slot_idx,
};
let mut cursor = 0usize;
while cursor < src.len() {
match ctx.parse_one(&src[cursor..]) {
ParseStep::Continue(n) => cursor += n,
ParseStep::Stop(n) => {
cursor += n;
break;
}
}
}
cursor
}
pub(super) fn parse_and_dispatch(&mut self, slot_idx: usize) {
if self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::PARKED)
{
return;
}
let CoreSlot {
conn, buf, user, ..
} = &mut self.core.slots[slot_idx];
let user = user.get_mut();
let ingress_ptr = buf.ingress_buf.as_ptr();
let mut ctx = ParseCtx {
protocol: &mut self.direction.protocol,
conn,
buf,
user,
slot_idx,
};
loop {
let tail = ctx.conn.ingress_tail as usize;
let head = ctx.conn.ingress_head as usize;
if head <= tail {
break;
}
let view: &[u8] =
unsafe { std::slice::from_raw_parts(ingress_ptr.add(tail), head - tail) };
match ctx.parse_one(view) {
ParseStep::Continue(n) => ctx.conn.ingress_tail = (tail + n) as u16,
ParseStep::Stop(n) => {
ctx.conn.ingress_tail = (tail + n) as u16;
break;
}
}
}
if ctx.conn.ingress_tail == ctx.conn.ingress_head {
ctx.conn.ingress_head = 0;
ctx.conn.ingress_tail = 0;
}
}
#[cold]
#[inline(never)]
fn spill_push(&mut self, slot_idx: usize, src: &[u8]) {
let trigger = {
let user = self.core.slots[slot_idx].user.get_mut();
let SlotIngress::Spill {
buf,
filled,
target,
} = &mut user.ingress
else {
return;
};
let needed = (*target as usize).saturating_sub(*filled as usize);
let take = src.len().min(needed);
buf.extend_from_slice(&src[..take]);
*filled = filled.saturating_add(take as u32);
*filled >= *target
};
if trigger {
self.dispatch_spilled(slot_idx);
}
}
#[cold]
fn request_stream_push(&mut self, driver: &mut B::Driver, slot_idx: usize, src: &[u8]) {
let (consumed, eof) = {
let user = self.core.slots[slot_idx].user.get();
match &user.ingress {
SlotIngress::RequestStream(handle) => {
let consumed = handle.push_slice(src);
(consumed, handle.is_complete())
}
_ => return,
}
};
if consumed == 0 && !src.is_empty() && !eof {
self.core.free_slot(driver, slot_idx);
return;
}
if !eof {
return;
}
self.core.slots[slot_idx].user.get_mut().ingress = SlotIngress::Idle;
let remaining = &src[consumed..];
if remaining.is_empty() {
return;
}
let entry = &mut self.core.slots[slot_idx];
let head = entry.conn.ingress_head as usize;
let room = entry.buf.ingress_buf.len().saturating_sub(head);
let n = remaining.len().min(room);
if n > 0 {
entry.buf.ingress_buf[head..head + n].copy_from_slice(&remaining[..n]);
entry.conn.ingress_head = (head + n) as u16;
}
}
#[inline(always)]
fn try_dispatch_spilled(&mut self, slot_idx: usize) {
let user = self.core.slots[slot_idx].user.get();
let SlotIngress::Spill { filled, target, .. } = &user.ingress else {
return;
};
if *filled < *target {
return;
}
self.dispatch_spilled(slot_idx);
}
#[cold]
#[inline(never)]
fn dispatch_spilled(&mut self, slot_idx: usize) {
let spill_box = {
let user = self.core.slots[slot_idx].user.get_mut();
let prev = std::mem::replace(&mut user.ingress, SlotIngress::Idle);
match prev {
SlotIngress::Spill { buf, .. } => buf,
other => {
user.ingress = other;
return;
}
}
};
let spill: BytesMut = *spill_box;
let req_bytes: &[u8] = spill.as_ref();
let parsed_head = {
let state = &mut self.core.slots[slot_idx].user.get_mut().protocol_state;
match self.direction.protocol.parse(state, req_bytes) {
Some((head, _, _)) => head,
None => return,
}
};
let egress_busy = !self.core.slots[slot_idx].user.get().egress.is_idle();
if !self.core.slots[slot_idx].conn.pending_body_ptr.is_null() || egress_busy {
return;
}
let CoreSlot {
conn, buf, user, ..
} = &mut self.core.slots[slot_idx];
let user = user.get_mut();
let wh = conn.write_head as usize;
if buf.write_buf.as_slice().len().saturating_sub(wh) < WRITE_ROOM_MIN {
return;
}
let mut ctx = ParseCtx {
protocol: &mut self.direction.protocol,
conn,
buf,
user,
slot_idx,
};
let write_slice: &mut [u8] = &mut ctx.buf.write_buf.as_mut_slice()[wh..];
let outcome = ctx.protocol.handle(req_bytes, parsed_head, write_slice);
let _ = ctx.apply_outcome(outcome, wh, req_bytes.len());
}
}