use super::super::Pool;
use super::super::slot::ConnFlags;
use super::Server;
use super::state::{SlotEgress, StreamState};
use super::wake::WakeQueue;
use crate::SlotId;
use crate::WriteBufStorage;
use crate::protocol::server::ServerProtocol;
use crate::protocol::server::{PendingOutcome, PendingResponse};
use dope_core::driver::PoolDriver;
use dope_core::profile::Profile;
use dope_transport::Transport;
use std::task::{Context, Poll};
impl<H: ServerProtocol, T: Transport, F: Profile, B: PoolDriver> Pool<Server<H, T>, F, B> {
#[inline(always)]
pub(crate) fn drain_ready_queue(&mut self, driver: &mut B::Driver) {
if !WakeQueue::is_pending() {
return;
}
self.drain_ready_queue_slow(driver);
}
#[cold]
#[inline(never)]
fn drain_ready_queue_slow(&mut self, driver: &mut B::Driver) {
loop {
let id = match WakeQueue::next() {
Some(i) => i,
None => return,
};
self.poll_pending(driver, id);
self.poll_long(driver, id);
}
}
#[cold]
#[inline(never)]
pub(super) fn poll_pending(&mut self, driver: &mut B::Driver, id: SlotId) {
let slot_idx = id.slot() as usize;
if slot_idx >= self.core.slots.len() {
return;
}
let user = self.core.slots[slot_idx].user.get();
let fresh = match user.pending_future.as_ref() {
Some(pf) => pf.generation == id.generation(),
None => false,
};
if !fresh {
return;
}
{
let conn = &self.core.slots[slot_idx].conn;
if !conn.flags.contains(ConnFlags::LIVE) || conn.generation != id.generation() {
self.core.slots[slot_idx].user.get_mut().pending_future = None;
return;
}
}
let waker = WakeQueue::make_waker(id);
let mut cx = Context::from_waker(&waker);
let poll_result = {
let user = self.core.slots[slot_idx].user.get_mut();
let pf = user
.pending_future
.as_mut()
.expect("just validated by fresh check");
unsafe { pf.future.poll_pinned(&mut cx) }
};
match poll_result {
Poll::Pending => {}
Poll::Ready(reply) => self.finish_async(driver, slot_idx, reply),
}
}
#[cold]
#[inline(never)]
pub(super) fn poll_long(&mut self, driver: &mut B::Driver, id: SlotId) {
let slot_idx = id.slot() as usize;
if slot_idx >= self.core.slots.len() {
return;
}
let user = self.core.slots[slot_idx].user.get();
let fresh = match user.long_future.as_ref() {
Some(lf) => lf.generation == id.generation(),
None => false,
};
if !fresh {
return;
}
{
let conn = &self.core.slots[slot_idx].conn;
if !conn.flags.contains(ConnFlags::LIVE) || conn.generation != id.generation() {
self.core.slots[slot_idx].user.get_mut().long_future = None;
return;
}
}
let waker = WakeQueue::make_waker(id);
let mut cx = Context::from_waker(&waker);
let poll_result = {
let user = self.core.slots[slot_idx].user.get_mut();
let lf = user
.long_future
.as_mut()
.expect("just validated by fresh check");
lf.future.as_mut().poll(&mut cx)
};
let wh = self.core.slots[slot_idx].conn.write_head as usize;
let written = {
let entry = &mut self.core.slots[slot_idx];
let cap = entry.buf.write_buf.as_slice().len();
let room = cap.saturating_sub(wh);
if room > 0 {
let slice = &mut entry.buf.write_buf.as_mut_slice()[wh..wh + room];
let user = entry.user.get_mut();
self.direction
.protocol
.drain_outbound(&mut user.protocol_state, slice)
} else {
0
}
};
if written > 0 {
let new_head = (wh as u32 + written).min(u16::MAX as u32) as u16;
self.core.slots[slot_idx].conn.write_head = new_head;
}
let close = poll_result.is_ready() || {
let user = self.core.slots[slot_idx].user.get();
self.direction.protocol.close_pending(&user.protocol_state)
};
if close {
self.core.slots[slot_idx]
.conn
.flags
.set(ConnFlags::CLOSE_AFTER, true);
self.core.slots[slot_idx].user.get_mut().long_future = None;
}
self.submit_slot_write(driver, slot_idx);
}
#[cold]
#[inline(never)]
pub(super) fn finish_async(
&mut self,
driver: &mut B::Driver,
slot_idx: usize,
reply: PendingResponse,
) {
let entry = &mut self.core.slots[slot_idx];
entry.user.get_mut().pending_future = None;
let outcome = {
let wh = entry.conn.write_head as usize;
let write_slice = &mut entry.buf.write_buf.as_mut_slice()[wh..];
(reply.serializer)(write_slice)
};
match outcome {
PendingOutcome::Inline {
written,
close_after,
} => {
entry.conn.flags.set(ConnFlags::PARKED, false);
let new_head = (entry.conn.write_head as u32).saturating_add(written as u32);
entry.conn.write_head = new_head.min(u16::MAX as u32) as u16;
if close_after {
entry.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
}
}
PendingOutcome::Static {
hdr_written,
body,
close_after,
} => {
entry.conn.flags.set(ConnFlags::PARKED, false);
let new_head =
(entry.conn.write_head as u32 + hdr_written).min(u16::MAX as u32) as u16;
entry.conn.write_head = new_head;
entry.conn.pending_body_ptr = body.as_ptr();
entry.conn.pending_body_len = body.len() as u32;
if close_after {
entry.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
}
}
PendingOutcome::Stream {
hdr_written,
stream,
close_after,
} => {
entry.conn.flags.set(ConnFlags::PARKED, false);
let new_head =
(entry.conn.write_head as u32 + hdr_written).min(u16::MAX as u32) as u16;
entry.conn.write_head = new_head;
entry.user.get_mut().egress =
SlotEgress::Stream(Box::new(StreamState::new(stream)));
if close_after {
entry.conn.flags.set(ConnFlags::CLOSE_AFTER, true);
}
}
}
self.parse_and_dispatch(slot_idx);
self.submit_slot_write(driver, slot_idx);
}
}