use std::mem::size_of;
use std::os::fd::{FromRawFd, OwnedFd, RawFd};
use super::super::pending::PendingCompletion;
use super::super::retry::Retry;
use super::super::udata::{Event, Udata};
use super::super::{Kqueue, MAX_DRAIN_PER_FD, PENDING_CAP};
use super::arm::Arm;
use super::{DrainOutcome, ReadKind, ReadSlot, Resume};
use crate::platform::raw::abi::PlatformAbi;
use crate::backend::kqueue::errno::Errno;
use crate::driver::Driver;
use crate::driver::token::Token;
use crate::io::ffi::Handle;
use crate::io::socket::msg::IoVec;
use crate::io::socket::msg::MsgHdr;
use libc::EMFILE;
use libc::ENOBUFS;
use libc::EVFILT_TIMER;
use libc::EVFILT_USER;
use libc::MSG_TRUNC;
use libc::accept;
use std::slice::from_raw_parts_mut;
use std::slice::from_ref;
use libc::kevent;
use libc::msghdr;
use libc::recv;
use libc::recvmsg;
use libc::sockaddr;
use libc::sockaddr_storage;
use libc::socklen_t;
pub(crate) trait Dispatch {
fn dispatch_event(&mut self, ev: &kevent);
fn dispatch_accept(&mut self, slot_idx: usize, epoch: u32, ev: &kevent);
fn drain_accept(
&mut self,
fd: RawFd,
ud: Token,
addr_ptr: *mut sockaddr,
addrlen_ptr: *mut socklen_t,
oneshot: bool,
) -> DrainOutcome;
fn dispatch_recv(&mut self, slot_idx: usize, epoch: u32, ev: &kevent);
fn drain_recv(&mut self, fd: RawFd, ud: Token) -> DrainOutcome;
fn dispatch_recv_msg(&mut self, slot_idx: usize, epoch: u32, ev: &kevent);
fn drain_recv_msg(
&mut self,
fd: RawFd,
ud: Token,
msg_tpl: *const msghdr,
) -> DrainOutcome;
fn queue_resume(&mut self, resume: Resume);
fn take_resume(&mut self, resume: Resume) -> bool;
fn resume_pending(&mut self);
}
impl Dispatch for Kqueue {
fn dispatch_event(&mut self, ev: &kevent) {
if ev.filter == EVFILT_USER {
return;
}
if ev.filter == EVFILT_TIMER {
if let Some(ud) = Token::try_from_raw(ev.udata as usize as u64) {
self.push_pending(PendingCompletion::Timer { ud });
}
return;
}
match Udata::from_kevent(ev.udata).decode() {
Some(Event::Accept { key, epoch }) => self.dispatch_accept(key.index(), epoch, ev),
Some(Event::Recv { key, epoch }) => self.dispatch_recv(key.index(), epoch, ev),
Some(Event::RecvMsg { key, epoch }) => {
self.dispatch_recv_msg(key.index(), epoch, ev);
}
Some(Event::WriteRetry { index, epoch }) => {
self.dispatch_write_retry(index, epoch);
}
Some(Event::Shutdown) => self.push_pending(PendingCompletion::Shutdown),
None => {}
}
}
fn dispatch_accept(&mut self, slot_idx: usize, epoch: u32, ev: &kevent) {
let outcome = {
let Some(slot) = self
.read_slots
.get_mut(&slot_idx)
.and_then(ReadSlot::accept_mut)
else {
return;
};
match slot.hdr.validate(ev, epoch) {
Ok((fd, ud)) => Ok((fd, ud, slot.addr_ptr, slot.addrlen_ptr, slot.oneshot)),
Err(e) => Err(e),
}
};
let (fd, ud, addr_ptr, addrlen_ptr, oneshot) = match outcome {
Ok(t) => t,
Err(None) => return,
Err(Some((ud, data))) => {
self.push_pending(PendingCompletion::Accept {
ud,
result: -data,
more: false,
});
return;
}
};
match self.drain_accept(fd, ud, addr_ptr, addrlen_ptr, oneshot) {
DrainOutcome::Done => {
if !oneshot {
self.re_enable_read(fd, Udata::accept(ud));
} else if let Some(s) = self
.read_slots
.get_mut(&slot_idx)
.and_then(ReadSlot::accept_mut)
{
s.hdr.armed = false;
}
}
DrainOutcome::Yield => self.queue_resume(Resume {
key: slot_idx,
epoch,
kind: ReadKind::Accept,
}),
DrainOutcome::Closed => {}
}
}
fn drain_accept(
&mut self,
fd: RawFd,
ud: Token,
addr_ptr: *mut sockaddr,
addrlen_ptr: *mut socklen_t,
oneshot: bool,
) -> DrainOutcome {
for _ in 0..MAX_DRAIN_PER_FD {
if self.pending.len() >= PENDING_CAP {
return DrainOutcome::Yield;
}
if !addrlen_ptr.is_null() {
unsafe {
*addrlen_ptr = size_of::<sockaddr_storage>() as socklen_t;
}
}
let raw = unsafe { accept(fd, addr_ptr, addrlen_ptr) };
let more_flag = !oneshot;
if raw >= 0 {
let Some(slot) = self.next_accept_slot() else {
drop(unsafe { OwnedFd::from_raw_fd(raw) });
self.push_pending(PendingCompletion::Accept {
ud,
result: -EMFILE,
more: more_flag,
});
return DrainOutcome::Done;
};
let accepted = Handle::from_owned(unsafe { OwnedFd::from_raw_fd(raw) });
if accepted.set_nonblocking().is_err()
|| accepted.set_cloexec().is_err()
|| Driver::set_no_sigpipe(&accepted).is_err()
{
continue;
}
self.register_fd(slot, accepted.into_owned());
self.push_pending(PendingCompletion::Accept {
ud,
result: slot as i32,
more: more_flag,
});
if oneshot {
return DrainOutcome::Done;
}
continue;
}
let errno = Errno::last();
if errno.is_block() {
return DrainOutcome::Done;
}
self.push_pending(PendingCompletion::Accept {
ud,
result: -errno.raw(),
more: more_flag,
});
return DrainOutcome::Done;
}
DrainOutcome::Yield
}
fn dispatch_recv(&mut self, slot_idx: usize, epoch: u32, ev: &kevent) {
let outcome = match self
.read_slots
.get_mut(&slot_idx)
.and_then(ReadSlot::recv_mut)
{
Some(h) => h.validate(ev, epoch),
None => return,
};
let (fd, ud) = match outcome {
Ok(p) => p,
Err(None) => return,
Err(Some((ud, data))) => {
self.push_pending(PendingCompletion::Recv {
ud,
result: -data,
more: false,
buffer: None,
});
return;
}
};
match self.drain_recv(fd, ud) {
DrainOutcome::Done => self.re_enable_read(fd, Udata::recv(ud)),
DrainOutcome::Yield => self.queue_resume(Resume {
key: slot_idx,
epoch,
kind: ReadKind::Recv,
}),
DrainOutcome::Closed => {}
}
}
fn drain_recv(&mut self, fd: RawFd, ud: Token) -> DrainOutcome {
for _ in 0..MAX_DRAIN_PER_FD {
if self.pending.len() >= PENDING_CAP {
return DrainOutcome::Yield;
}
let Some(mut buffer) = self.recv.take() else {
return DrainOutcome::Yield;
};
let cap = buffer.capacity();
let ptr = buffer.as_mut_ptr();
let n = unsafe { recv(fd, ptr.cast(), cap, 0) };
if n > 0 {
self.push_pending(PendingCompletion::Recv {
ud,
result: n as i32,
more: true,
buffer: Some(buffer.into_buffer()),
});
continue;
}
if n == 0 {
self.recv.defer(buffer.into_buffer());
self.push_pending(PendingCompletion::Recv {
ud,
result: 0,
more: false,
buffer: None,
});
return DrainOutcome::Closed;
}
let errno = Errno::last();
self.recv.defer(buffer.into_buffer());
if errno.is_block() {
return DrainOutcome::Done;
}
self.push_pending(PendingCompletion::Recv {
ud,
result: -errno.raw(),
more: true,
buffer: None,
});
return DrainOutcome::Done;
}
DrainOutcome::Yield
}
fn dispatch_recv_msg(&mut self, slot_idx: usize, epoch: u32, ev: &kevent) {
let outcome = {
let Some(slot) = self
.read_slots
.get_mut(&slot_idx)
.and_then(ReadSlot::recv_msg_mut)
else {
return;
};
match slot.hdr.validate(ev, epoch) {
Ok((fd, ud)) => Ok((fd, ud, slot.msg_template)),
Err(e) => Err(e),
}
};
let (fd, ud, msg_tpl) = match outcome {
Ok(t) => t,
Err(None) => return,
Err(Some((ud, data))) => {
self.push_pending(PendingCompletion::Recv {
ud,
result: -data,
more: false,
buffer: None,
});
return;
}
};
match self.drain_recv_msg(fd, ud, msg_tpl) {
DrainOutcome::Done => self.re_enable_read(fd, Udata::recv_msg(ud)),
DrainOutcome::Yield => self.queue_resume(Resume {
key: slot_idx,
epoch,
kind: ReadKind::RecvMsg,
}),
DrainOutcome::Closed => {}
}
}
fn drain_recv_msg(
&mut self,
fd: RawFd,
ud: Token,
msg_tpl: *const msghdr,
) -> DrainOutcome {
let template = unsafe { *msg_tpl };
let namelen = template.msg_namelen as usize;
for _ in 0..MAX_DRAIN_PER_FD {
if self.pending.len() >= PENDING_CAP {
return DrainOutcome::Yield;
}
let Some(mut buffer) = self.recv.take() else {
return DrainOutcome::Yield;
};
let cap = buffer.capacity();
let ptr = buffer.as_mut_ptr();
if cap <= namelen {
self.recv.defer(buffer.into_buffer());
self.push_pending(PendingCompletion::Recv {
ud,
result: -ENOBUFS,
more: true,
buffer: None,
});
return DrainOutcome::Done;
}
let iov = IoVec::from_mut_slice(unsafe {
from_raw_parts_mut(ptr.add(namelen), cap - namelen)
});
let mut local_msg = MsgHdr::empty();
local_msg.set_name_ptr(ptr.cast(), template.msg_namelen);
local_msg.set_iov(from_ref(&iov));
let n = unsafe { recvmsg(fd, local_msg.as_mut_ptr(), 0) };
if n > 0 {
if local_msg.flags() & MSG_TRUNC != 0 {
self.recv.defer(buffer.into_buffer());
continue;
}
let total = namelen + n as usize;
self.push_pending(PendingCompletion::Recv {
ud,
result: total as i32,
more: true,
buffer: Some(buffer.into_buffer()),
});
continue;
}
if n == 0 {
self.push_pending(PendingCompletion::Recv {
ud,
result: namelen as i32,
more: true,
buffer: Some(buffer.into_buffer()),
});
return DrainOutcome::Done;
}
let errno = Errno::last();
self.recv.defer(buffer.into_buffer());
if errno.is_block() {
return DrainOutcome::Done;
}
self.push_pending(PendingCompletion::Recv {
ud,
result: -errno.raw(),
more: true,
buffer: None,
});
return DrainOutcome::Done;
}
DrainOutcome::Yield
}
fn queue_resume(&mut self, resume: Resume) {
let Some(slot) = self.read_slots.get_mut(&resume.key) else {
return;
};
if slot.kind() != resume.kind {
return;
}
let header = slot.header_mut();
if header.epoch != resume.epoch || !header.armed || header.resume_queued {
return;
}
let Some(entry) = self.resume.vacant_entry() else {
unreachable!()
};
header.resume_queued = true;
entry.push_back(resume);
}
fn take_resume(&mut self, resume: Resume) -> bool {
let Some(slot) = self.read_slots.get_mut(&resume.key) else {
return false;
};
if slot.kind() != resume.kind {
return false;
}
let header = slot.header_mut();
if header.epoch != resume.epoch || !header.resume_queued {
return false;
}
header.resume_queued = false;
header.armed
}
fn resume_pending(&mut self) {
let n = self.resume.len();
for _ in 0..n {
let Some(item) = self.resume.pop_front() else {
break;
};
if !self.take_resume(item) {
continue;
}
let slot_idx = item.key;
match item.kind {
ReadKind::Accept => {
let Some((fd, ud, addr_ptr, addrlen_ptr, oneshot)) = self
.read_slots
.get(&slot_idx)
.and_then(ReadSlot::accept)
.filter(|s| s.hdr.armed)
.map(|s| (s.hdr.fd, s.hdr.ud, s.addr_ptr, s.addrlen_ptr, s.oneshot))
else {
continue;
};
match self.drain_accept(fd, ud, addr_ptr, addrlen_ptr, oneshot) {
DrainOutcome::Done => {
if !oneshot {
self.re_enable_read(fd, Udata::accept(ud))
} else if let Some(s) = self
.read_slots
.get_mut(&slot_idx)
.and_then(ReadSlot::accept_mut)
{
s.hdr.armed = false;
}
}
DrainOutcome::Yield => self.queue_resume(item),
DrainOutcome::Closed => {}
}
}
ReadKind::Recv => {
let Some((fd, ud)) = self
.read_slots
.get(&slot_idx)
.and_then(ReadSlot::recv)
.filter(|h| h.armed)
.map(|h| (h.fd, h.ud))
else {
continue;
};
match self.drain_recv(fd, ud) {
DrainOutcome::Done => {
self.re_enable_read(fd, Udata::recv(ud))
}
DrainOutcome::Yield => self.queue_resume(item),
DrainOutcome::Closed => {}
}
}
ReadKind::RecvMsg => {
let Some((fd, ud, tpl)) = self
.read_slots
.get(&slot_idx)
.and_then(ReadSlot::recv_msg)
.filter(|s| s.hdr.armed)
.map(|s| (s.hdr.fd, s.hdr.ud, s.msg_template))
else {
continue;
};
match self.drain_recv_msg(fd, ud, tpl) {
DrainOutcome::Done => {
self.re_enable_read(fd, Udata::recv_msg(ud))
}
DrainOutcome::Yield => self.queue_resume(item),
DrainOutcome::Closed => {}
}
}
}
}
}
}