use crate::CodecLayer;
use crate::WriteBufStorage;
use crate::protocol::client::ClientProtocol;
use dope_core::driver::{DriverOps, PoolDriver};
use dope_core::profile::Profile;
use dope_transport::Transport;
use super::super::Pool;
use super::super::core::PushError;
use super::super::slot::{ConnFlags, CoreSlot};
use super::Client;
use super::connect_source::ConnectSource;
use super::state::CLIENT_BYTES_CAP;
use crate::SlotId;
impl<P, T, S, F, B> Pool<Client<P, T, S>, F, B>
where
P: ClientProtocol,
T: Transport,
T::Addr: Clone,
S: ConnectSource<T>,
F: Profile,
B: PoolDriver,
{
#[inline]
pub(super) fn try_submit_outbound(&mut self, driver: &mut B::Driver, slot_idx: usize) {
if !self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::LIVE)
{
return;
}
if self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::CLOSE_AFTER)
{
return;
}
if <P::CodecLayer as CodecLayer>::IS_PASSTHROUGH {
self.try_submit_outbound_plain(driver, slot_idx);
} else {
self.try_submit_outbound_codec(driver, slot_idx);
}
}
#[inline]
fn try_submit_outbound_plain(&mut self, driver: &mut B::Driver, slot_idx: usize) {
if self.core.slots[slot_idx].io.write.has_in_flight {
return;
}
let conn_fd;
let token_epoch: Option<(B::Token, u32)>;
let msg_ptr: *const libc::msghdr;
let close;
{
let external = self.core.slot_id_base + slot_idx;
let entry = &mut self.core.slots[slot_idx];
conn_fd = entry.conn.fd;
let user = entry.user.get_mut();
match user.session.fill_msghdr(CLIENT_BYTES_CAP) {
Some(m) => {
msg_ptr = m as *const _;
token_epoch = Some(entry.io.write.next_token::<B::Token>(external));
close = false;
}
None => {
msg_ptr = std::ptr::null();
token_epoch = None;
close = self.direction.protocol.wants_close(user.session.state());
}
}
}
if let Some((token, epoch)) = token_epoch {
if unsafe { driver.submit_send_msg_tagged(token, conn_fd, msg_ptr) } {
self.core.slots[slot_idx].io.write.mark_inflight(epoch);
}
} else if close {
self.disconnect_and_free(driver, slot_idx);
}
}
#[cold]
fn try_submit_outbound_codec(&mut self, driver: &mut B::Driver, slot_idx: usize) {
let cap = self.core.slots[slot_idx].buf.write_buf.as_slice().len();
let head = self.core.slots[slot_idx].conn.write_head as usize;
let room = cap.saturating_sub(head);
if room > 0 {
let CoreSlot {
conn, buf, user, ..
} = &mut self.core.slots[slot_idx];
let user = user.get_mut();
let req_cap = room.min(CLIENT_BYTES_CAP);
if let Some(msg) = user.session.fill_msghdr(req_cap) {
let iov_count = msg.msg_iovlen as usize;
let iov_slice: &[libc::iovec] =
unsafe { std::slice::from_raw_parts(msg.msg_iov, iov_count) };
let mut copied = 0u32;
for iov in iov_slice {
let len = iov.iov_len;
if (head + copied as usize) + len > cap {
break;
}
let src = unsafe { std::slice::from_raw_parts(iov.iov_base.cast::<u8>(), len) };
let off = head + copied as usize;
buf.write_buf.as_mut_slice()[off..off + len].copy_from_slice(src);
copied += len as u32;
}
conn.write_head = (head + copied as usize).min(u16::MAX as usize) as u16;
user.session.ack_send(copied);
}
}
let close = {
let user = self.core.slots[slot_idx].user.get();
self.direction.protocol.wants_close(user.session.state())
};
if close {
self.core.slots[slot_idx]
.conn
.flags
.set(ConnFlags::CLOSE_AFTER, true);
}
self.core.encrypt_write_buf(slot_idx);
self.core.try_submit_codec(driver, slot_idx);
}
pub fn poke_outbound(
&mut self,
driver: &mut B::Driver,
conn_id: SlotId,
) -> Result<(), PushError> {
let slot_idx = self.core.resolve_live_slot(conn_id)?;
self.try_submit_outbound(driver, slot_idx);
Ok(())
}
pub fn pump_outbound(&mut self, driver: &mut B::Driver) -> usize {
let mut count = 0usize;
for slot_idx in 0..self.core.slots.len() {
if !self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::LIVE)
{
continue;
}
self.try_submit_outbound(driver, slot_idx);
count += 1;
}
count
}
#[cold]
fn handle_write_event_codec(&mut self, driver: &mut B::Driver, slot_idx: usize, result: i32) {
match self.core.handle_codec_inflight(slot_idx, result) {
super::super::core::CodecAck::NoOp => {}
super::super::core::CodecAck::Err => self.disconnect_and_free(driver, slot_idx),
super::super::core::CodecAck::Progressed => {
self.core.try_submit_codec(driver, slot_idx);
self.try_submit_outbound(driver, slot_idx);
let close = self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::CLOSE_AFTER)
|| self.core.codec_close_pending(slot_idx);
if self.core.codec_idle(slot_idx) && close {
self.disconnect_and_free(driver, slot_idx);
}
}
}
}
}
impl<P, T, S, F, B> super::super::PoolWrite<B> for Pool<Client<P, T, S>, F, B>
where
P: ClientProtocol,
T: Transport,
T::Addr: Clone,
S: ConnectSource<T>,
F: Profile,
B: PoolDriver,
{
fn on_write_event(
&mut self,
driver: &mut B::Driver,
token: B::Token,
result: i32,
notif: bool,
) {
let (slot_idx, result) = match self.core.classify_write(token, result, notif) {
super::super::core::WriteDispatch::Skipped => return,
super::super::core::WriteDispatch::Ready { slot_idx, result } => (slot_idx, result),
};
if <P::CodecLayer as CodecLayer>::IS_PASSTHROUGH {
if !self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::LIVE)
{
return;
}
if result < 0 {
self.disconnect_and_free(driver, slot_idx);
return;
}
let n = result as u32;
if n > 0 {
let user = self.core.slots[slot_idx].user.get_mut();
user.session.ack_send(n);
}
if self.core.slots[slot_idx]
.conn
.flags
.contains(ConnFlags::CLOSE_AFTER)
{
self.disconnect_and_free(driver, slot_idx);
return;
}
self.try_submit_outbound(driver, slot_idx);
} else {
self.handle_write_event_codec(driver, slot_idx, result);
}
}
}