mod current;
mod provided;
mod request;
mod sock;
mod token;
mod udata;
use std::collections::{HashMap, VecDeque};
use std::io;
use std::mem::MaybeUninit;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
use std::time::Duration;
use dope_core::driver as caps;
use dope_core::driver::{DriverOps, ProvidedRingOps};
use dope_core::{Fd, FixedFd};
use crate::DriverConfig;
use crate::Errno;
pub use self::current::{CurrentSlot, current, current_driver_mut, set_current};
pub use self::provided::{ProvidedBuf, ProvidedRing};
pub use self::request::Request;
pub(crate) use self::request::Requests;
pub(crate) use self::sock::Sock;
pub use self::token::Token;
use self::request::{OneShotOp, RequestId, TAG_REQUEST};
use self::sock::Sock as SockHelper;
use self::udata::Udata;
pub(crate) const TAG_ACCEPT: u8 = 1;
pub(crate) const TAG_RECV: u8 = 2;
pub(crate) const TAG_RECV_MSG: u8 = 3;
pub struct Driver {
kq: OwnedFd,
pub(crate) changes: Vec<libc::kevent>,
pub(crate) accept_slots: HashMap<usize, MultiSlot>,
pub(crate) recv_slots: HashMap<usize, MultiSlot>,
next_fixed_idx: u32,
pub(crate) pending: VecDeque<PendingCompletion>,
pub(crate) requests: Requests,
pub(crate) provided: ProvidedRing,
}
pub(crate) struct MultiSlot {
pub(crate) fd: RawFd,
pub(crate) epoch: u32,
pub(crate) armed: bool,
}
#[derive(Clone, Copy, Debug)]
pub(crate) enum PendingCompletion {
Accept {
token: Token,
result: i32,
more: bool,
},
Recv {
token: Token,
result: i32,
more: bool,
bid: Option<u16>,
},
Write {
token: Token,
result: i32,
notif: bool,
},
}
impl From<PendingCompletion> for caps::Completion<crate::KqueueBackend> {
fn from(p: PendingCompletion) -> Self {
match p {
PendingCompletion::Accept {
token,
result,
more,
} => Self::Accept {
token,
result,
more,
},
PendingCompletion::Recv {
token,
result,
more,
bid,
} => Self::Recv {
token,
result,
more,
bid,
},
PendingCompletion::Write {
token,
result,
notif,
} => Self::Write {
token,
result,
notif,
},
}
}
}
enum OneShotSubmit {
Ok(RequestId),
Err(io::Error),
}
impl Driver {
pub fn new(cfg: DriverConfig) -> io::Result<Self> {
let raw = unsafe { libc::kqueue() };
if raw < 0 {
return Err(io::Error::last_os_error());
}
Ok(Self {
kq: unsafe { OwnedFd::from_raw_fd(raw) },
changes: Vec::with_capacity(64),
accept_slots: HashMap::new(),
recv_slots: HashMap::new(),
next_fixed_idx: 1,
pending: VecDeque::new(),
requests: Requests::default(),
provided: ProvidedRing::new(cfg.provided_buf_entries, cfg.provided_buf_len as u32),
})
}
pub fn has_pending(&mut self) -> bool {
!self.pending.is_empty() || !self.changes.is_empty()
}
pub fn submit_only(&mut self) -> io::Result<()> {
if self.changes.is_empty() {
return Ok(());
}
self.kevent_call(&mut [], Some(Duration::ZERO)).map(|_| ())
}
pub fn park(&mut self) -> io::Result<()> {
self.poll(None).map(|_| ())
}
pub fn park_for(&mut self, duration: Duration) -> io::Result<()> {
self.poll(Some(duration)).map(|_| ())
}
pub fn drive(&mut self) -> io::Result<bool> {
let before = self.pending.len();
let n = self.poll(Some(Duration::ZERO))?;
Ok(n > 0 || self.pending.len() != before)
}
fn poll(&mut self, timeout: Option<Duration>) -> io::Result<usize> {
let mut events: [MaybeUninit<libc::kevent>; 64] = [const { MaybeUninit::uninit() }; 64];
let n = self.kevent_call(&mut events, timeout)?;
let ready =
unsafe { std::slice::from_raw_parts(events.as_ptr().cast::<libc::kevent>(), n) };
for ev in ready {
self.dispatch_event(ev);
}
Ok(n)
}
fn kevent_call(
&mut self,
events: &mut [MaybeUninit<libc::kevent>],
timeout: Option<Duration>,
) -> io::Result<usize> {
let ts_storage;
let ts_ptr: *const libc::timespec = match timeout {
None => std::ptr::null(),
Some(d) => {
ts_storage = libc::timespec {
tv_sec: d.as_secs() as libc::time_t,
tv_nsec: d.subsec_nanos() as libc::c_long,
};
&ts_storage
}
};
let n = unsafe {
libc::kevent(
self.kq.as_raw_fd(),
self.changes.as_ptr(),
self.changes.len() as libc::c_int,
events.as_mut_ptr().cast(),
events.len() as libc::c_int,
ts_ptr,
)
};
self.changes.clear();
if n < 0 {
let err = io::Error::last_os_error();
return if err.raw_os_error() == Some(libc::EINTR) {
Ok(0)
} else {
Err(err)
};
}
Ok(n as usize)
}
pub fn drive_with<R>(&mut self, router: &mut R) -> io::Result<bool>
where
R: caps::CompletionRouter<crate::KqueueBackend>,
{
let n = self.poll(Some(Duration::ZERO))?;
let progressed = n > 0 || !self.pending.is_empty();
self.drain_pending(router);
Ok(progressed)
}
pub fn dispatch_completions<R>(&mut self, router: &mut R)
where
R: caps::CompletionRouter<crate::KqueueBackend>,
{
self.drain_pending(router);
}
}
impl Driver {
fn dispatch_event(&mut self, ev: &libc::kevent) {
let raw = Udata::from_kevent(ev.udata);
let (tag, slot, epoch) = raw.unpack();
if tag == TAG_REQUEST {
self.dispatch_request(RequestId::new(slot, epoch));
return;
}
let token = raw.into_token();
let slot_idx = slot as usize;
match tag {
TAG_ACCEPT => self.dispatch_accept(slot_idx, epoch, token, ev),
TAG_RECV => self.dispatch_recv(slot_idx, epoch, token, ev),
TAG_RECV_MSG => {}
_ => {}
}
}
fn validate_arm(
slot: Option<&mut MultiSlot>,
ev: &libc::kevent,
epoch: u32,
) -> Result<RawFd, Option<i32>> {
let slot = slot.ok_or(None)?;
if slot.epoch != epoch || !slot.armed {
return Err(None);
}
if (ev.flags & libc::EV_ERROR) != 0 && ev.data != 0 {
slot.armed = false;
return Err(Some(ev.data as i32));
}
Ok(slot.fd)
}
fn dispatch_accept(&mut self, slot_idx: usize, epoch: u32, token: Token, ev: &libc::kevent) {
let fd = match Self::validate_arm(self.accept_slots.get_mut(&slot_idx), ev, epoch) {
Ok(fd) => fd,
Err(None) => return,
Err(Some(data)) => {
self.pending.push_back(PendingCompletion::Accept {
token,
result: -data,
more: false,
});
return;
}
};
loop {
let raw = unsafe { libc::accept(fd, std::ptr::null_mut(), std::ptr::null_mut()) };
if raw >= 0 {
SockHelper::set_nonblocking(raw);
self.pending.push_back(PendingCompletion::Accept {
token,
result: raw as i32,
more: true,
});
continue;
}
let errno = Errno::last_raw();
if Errno::is_block_raw(errno) {
return;
}
self.pending.push_back(PendingCompletion::Accept {
token,
result: -errno,
more: true,
});
return;
}
}
fn dispatch_recv(&mut self, slot_idx: usize, epoch: u32, token: Token, ev: &libc::kevent) {
let fd = match Self::validate_arm(self.recv_slots.get_mut(&slot_idx), ev, epoch) {
Ok(fd) => fd,
Err(None) => return,
Err(Some(data)) => {
self.pending.push_back(PendingCompletion::Recv {
token,
result: -data,
more: false,
bid: None,
});
return;
}
};
loop {
let Some(bid) = self.provided.pop_free() else {
return;
};
let (ptr, cap) = self.provided.ptr_len(bid);
let n = unsafe { libc::recv(fd, ptr.cast(), cap, 0) };
if n > 0 {
self.pending.push_back(PendingCompletion::Recv {
token,
result: n as i32,
more: true,
bid: Some(bid),
});
continue;
}
if n == 0 {
self.pending.push_back(PendingCompletion::Recv {
token,
result: 0,
more: true,
bid: Some(bid),
});
return;
}
let errno = Errno::last_raw();
self.provided.defer(bid);
if Errno::is_block_raw(errno) {
return;
}
self.pending.push_back(PendingCompletion::Recv {
token,
result: -errno,
more: true,
bid: None,
});
return;
}
}
fn dispatch_request(&mut self, id: RequestId) {
let Some((fd, op)) = self.requests.with_mut(id, |slot| (slot.fd, slot.op)) else {
return;
};
let result = op.run(fd);
if let Some(w) = self.requests.fire(id, result) {
w.wake();
}
}
pub(super) fn drain_pending<R>(&mut self, router: &mut R)
where
R: caps::CompletionRouter<crate::KqueueBackend>,
{
while let Some(p) = self.pending.pop_front() {
router.on_complete(self, p.into());
}
}
}
impl Driver {
fn arm_read_multi(&mut self, raw: RawFd, udata: Udata) {
SockHelper::set_nonblocking(raw);
self.changes.push(libc::kevent {
ident: raw as libc::uintptr_t,
filter: libc::EVFILT_READ,
flags: libc::EV_ADD | libc::EV_CLEAR,
fflags: 0,
data: 0,
udata: udata.into_kevent(),
});
}
fn arm_write_oneshot(&mut self, raw: RawFd, udata: Udata) {
self.changes.push(libc::kevent {
ident: raw as libc::uintptr_t,
filter: libc::EVFILT_WRITE,
flags: libc::EV_ADD | libc::EV_CLEAR | libc::EV_ONESHOT,
fflags: 0,
data: 0,
udata: udata.into_kevent(),
});
}
pub(crate) fn disarm_filter(&mut self, fd: RawFd, filter: i16) {
if filter == 0 {
return;
}
self.changes.push(libc::kevent {
ident: fd as libc::uintptr_t,
filter,
flags: libc::EV_DELETE,
fflags: 0,
data: 0,
udata: std::ptr::null_mut(),
});
}
pub fn try_submit_sendmsg_raw_boxed<P>(
&mut self,
fd: Fd,
mut packet: Box<P>,
) -> Result<Request<P>, (io::Error, Box<P>)>
where
P: dope_core::submit::SendMsgPacket + Unpin + 'static,
{
let raw = fd.as_raw_fd();
SockHelper::set_nonblocking(raw);
let msg = packet.msg_ptr();
let op = OneShotOp::SendMsg { msg };
let n = unsafe { libc::sendmsg(raw, msg, 0) };
match self.handle_oneshot_submit(raw, op, n) {
OneShotSubmit::Ok(id) => Ok(Request::new(self, packet, id)),
OneShotSubmit::Err(err) => Err((err, packet)),
}
}
fn handle_oneshot_submit(
&mut self,
raw: RawFd,
op: OneShotOp,
n: libc::ssize_t,
) -> OneShotSubmit {
if n >= 0 {
return OneShotSubmit::Ok(self.requests.alloc(raw, 0, op, Some(Ok(n as u32))));
}
let err = io::Error::last_os_error();
if matches!(err.raw_os_error(), Some(libc::EAGAIN)) {
OneShotSubmit::Ok(self.register_oneshot(raw, op))
} else {
OneShotSubmit::Err(err)
}
}
fn register_oneshot(&mut self, raw: RawFd, op: OneShotOp) -> RequestId {
let id = self.requests.alloc(raw, libc::EVFILT_WRITE, op, None);
let _ = self.requests.with_mut(id, |s| s.armed = true);
self.arm_write_oneshot(raw, id.into_udata());
id
}
pub fn register_read_oneshot(&mut self, raw: RawFd) -> Request<()> {
let id = self
.requests
.alloc(raw, libc::EVFILT_READ, OneShotOp::Wake, None);
let _ = self.requests.with_mut(id, |s| s.armed = true);
self.changes.push(libc::kevent {
ident: raw as libc::uintptr_t,
filter: libc::EVFILT_READ,
flags: libc::EV_ADD | libc::EV_CLEAR | libc::EV_ONESHOT,
fflags: 0,
data: 0,
udata: id.into_udata().into_kevent(),
});
Request::new(self, Box::new(()), id)
}
}
impl caps::DriverLifecycle for Driver {
fn drive(&mut self) -> io::Result<bool> {
Self::drive(self)
}
fn has_pending(&mut self) -> bool {
Self::has_pending(self)
}
fn park(&mut self) -> io::Result<()> {
Self::park(self)
}
fn park_for(&mut self, duration: Duration) -> io::Result<()> {
Self::park_for(self, duration)
}
fn submit_only(&mut self) -> io::Result<()> {
Self::submit_only(self)
}
}
pub struct FixedBufGuard;
impl caps::FixedBuffers for Driver {
type Guard = FixedBufGuard;
fn fixed_guard(&mut self) -> Option<Self::Guard> {
Some(FixedBufGuard)
}
fn register_fixed(&mut self, _fd: Fd) -> io::Result<Option<u32>> {
let idx = self.next_fixed_idx;
self.next_fixed_idx = self.next_fixed_idx.wrapping_add(1).max(1);
Ok(Some(idx))
}
fn unregister_fixed(&mut self, _idx: u32) {}
}
impl caps::ProvidedBufRing for Driver {
type Provided = ProvidedBuf;
fn provided_group(&mut self) -> Option<(u16, u32)> {
Some(self.provided.group())
}
fn provided(&mut self, bid: u16, len: usize) -> Option<Self::Provided> {
Some(self.provided.make_buf(bid, len))
}
}
impl DriverOps for Driver {
type Token = Token;
type ProvidedRing = ProvidedRing;
type IoRequest<O: Unpin + 'static> = Request<O>;
fn provided_ring(&mut self) -> &mut Self::ProvidedRing {
&mut self.provided
}
fn buf_group(&self) -> u16 {
self.provided.group().0
}
fn arm_accept_multi(&mut self, token: Self::Token, fd: Fd) -> bool {
let raw = fd.as_raw_fd();
let (slot_idx, epoch) = token.parts();
self.accept_slots.insert(
slot_idx,
MultiSlot {
fd: raw,
epoch,
armed: true,
},
);
self.arm_read_multi(raw, Udata::from_token(token, TAG_ACCEPT));
true
}
fn arm_recv_multi(&mut self, token: Self::Token, fd: FixedFd, _buf_group: u16) -> bool {
let raw = fd.as_fd().as_raw_fd();
let (slot_idx, epoch) = token.parts();
self.recv_slots.insert(
slot_idx,
MultiSlot {
fd: raw,
epoch,
armed: true,
},
);
self.arm_read_multi(raw, Udata::from_token(token, TAG_RECV));
true
}
unsafe fn arm_recv_msg_multi(
&mut self,
token: Self::Token,
fd: FixedFd,
_buf_group: u16,
_msg: *const libc::msghdr,
) -> bool {
let raw = fd.as_fd().as_raw_fd();
self.arm_read_multi(raw, Udata::from_token(token, TAG_RECV_MSG));
true
}
fn submit_send_tagged(
&mut self,
token: Self::Token,
fd: FixedFd,
ptr: *const u8,
len: u32,
) -> bool {
let raw = fd.as_fd().as_raw_fd();
let n = unsafe { libc::send(raw, ptr.cast(), len as usize, 0) };
let result = if n >= 0 { n as i32 } else { -Errno::last_raw() };
self.pending.push_back(PendingCompletion::Write {
token,
result,
notif: false,
});
true
}
unsafe fn submit_send_msg_tagged(
&mut self,
token: Self::Token,
fd: FixedFd,
msg: *const libc::msghdr,
) -> bool {
let raw = fd.as_fd().as_raw_fd();
let n = unsafe { libc::sendmsg(raw, msg, 0) };
let result = if n >= 0 { n as i32 } else { -Errno::last_raw() };
self.pending.push_back(PendingCompletion::Write {
token,
result,
notif: false,
});
true
}
fn cancel_recv(&mut self, token: Self::Token) {
let (slot_idx, _epoch) = token.parts();
if let Some(slot) = self.recv_slots.remove(&slot_idx) {
self.disarm_filter(slot.fd, libc::EVFILT_READ);
}
}
fn try_submit_connect_raw(
&mut self,
fd: Fd,
addr: dope_core::addr::SockAddr,
) -> Result<Self::IoRequest<dope_core::addr::SockAddr>, (io::Error, dope_core::addr::SockAddr)>
{
let raw = fd.as_raw_fd();
SockHelper::set_nonblocking(raw);
let boxed = Box::new(addr);
let rc = unsafe { libc::connect(raw, boxed.addr_ptr(), boxed.addr_len()) };
if rc == 0 {
let id = self.requests.alloc(raw, 0, OneShotOp::Connect, Some(Ok(0)));
return Ok(Request::new(self, boxed, id));
}
let err = io::Error::last_os_error();
match err.raw_os_error() {
Some(libc::EINPROGRESS) | Some(libc::EALREADY) => {
let id = self.register_oneshot(raw, OneShotOp::Connect);
Ok(Request::new(self, boxed, id))
}
_ => Err((err, *boxed)),
}
}
fn try_submit_send_raw<B: AsRef<[u8]> + Unpin + 'static>(
&mut self,
fd: Fd,
buf: B,
) -> Result<Self::IoRequest<B>, (io::Error, B)> {
let raw = fd.as_raw_fd();
SockHelper::set_nonblocking(raw);
let boxed = Box::new(buf);
let bytes = (*boxed).as_ref();
let op = OneShotOp::Send {
ptr: bytes.as_ptr(),
len: bytes.len() as u32,
};
let n = unsafe { libc::send(raw, bytes.as_ptr().cast(), bytes.len(), 0) };
match self.handle_oneshot_submit(raw, op, n) {
OneShotSubmit::Ok(id) => Ok(Request::new(self, boxed, id)),
OneShotSubmit::Err(err) => Err((err, *boxed)),
}
}
fn try_submit_sendmsg_raw<P: dope_core::submit::SendMsgPacket + Unpin + 'static>(
&mut self,
fd: Fd,
packet: P,
) -> Result<Self::IoRequest<P>, (io::Error, P)> {
match self.try_submit_sendmsg_raw_boxed(fd, Box::new(packet)) {
Ok(req) => Ok(req),
Err((e, boxed)) => Err((e, *boxed)),
}
}
}
pub struct RecvMsgOut;
impl caps::RecvMsgOutView for RecvMsgOut {
fn parse_with<R>(
_raw: &[u8],
_msghdr: &libc::msghdr,
_f: impl FnOnce(&[u8], &[u8]) -> R,
) -> Option<R> {
None
}
}
#[cfg(test)]
mod tests {
use super::*;
use dope_core::driver::{Backend, Completion, CompletionBackend, CompletionRouter};
use crate::{DriverConfig, KqueueBackend};
#[test]
fn token_udata_round_trip() {
let t = Token::from_index_raw(42, 0x123456);
let raw = Udata::from_token(t, TAG_RECV);
let (tag, slot, epoch) = raw.unpack();
assert_eq!(tag, TAG_RECV);
assert_eq!((slot as usize, epoch), (42, 0x123456));
}
struct CountingRouter {
accepts: u32,
recvs: u32,
writes: u32,
}
impl CompletionRouter<KqueueBackend> for CountingRouter {
fn on_complete(&mut self, _driver: &mut Driver, ev: Completion<KqueueBackend>) {
match ev {
Completion::Accept { .. } => self.accepts += 1,
Completion::Recv { .. } => self.recvs += 1,
Completion::Write { .. } => self.writes += 1,
}
}
}
#[test]
fn dispatch_completions_routes_each_kind() {
let mut driver =
<KqueueBackend as Backend>::new_driver(DriverConfig::default()).expect("driver");
let token = Token::from_index_raw(7, 1);
driver.pending.push_back(PendingCompletion::Accept {
token,
result: 99,
more: true,
});
driver.pending.push_back(PendingCompletion::Recv {
token,
result: 4,
more: true,
bid: Some(0),
});
driver.pending.push_back(PendingCompletion::Write {
token,
result: 12,
notif: false,
});
let mut router = CountingRouter {
accepts: 0,
recvs: 0,
writes: 0,
};
<KqueueBackend as CompletionBackend>::dispatch_completions(&mut driver, &mut router);
assert_eq!((router.accepts, router.recvs, router.writes), (1, 1, 1));
assert!(driver.pending.is_empty());
}
}