use std::os::fd::{FromRawFd, OwnedFd, RawFd};
use super::Kqueue;
use super::pending::PendingCompletion;
use super::read::arm::Arm;
use super::retry::{Retry, WriteKind};
use crate::platform::raw::abi::PlatformAbi;
use crate::backend::kqueue::errno::Errno;
use crate::driver::Driver;
use crate::driver::token::{KIND_SHIFT, Token};
use crate::io::fd::FdSlot;
use crate::io::ffi::Handle;
use crate::driver::token::kind::ACCEPT;
use crate::driver::token::kind::CONNECT;
use libc::EBADF;
use libc::EINPROGRESS;
use libc::EVFILT_TIMER;
use libc::EV_DELETE;
use crate::driver::token::kind::RECV;
use crate::driver::token::kind::RECV_DISCARD;
use crate::driver::token::kind::SEND;
use crate::driver::token::kind::TIMER;
use libc::c_char;
use libc::c_uint;
use libc::connect;
use libc::kevent;
use libc::msghdr;
use std::ptr::null_mut;
use libc::off_t;
use libc::openat;
use libc::pread;
use libc::pwrite;
use libc::send;
use libc::sendmsg;
use libc::sockaddr;
use libc::socket;
use libc::socklen_t;
use libc::uintptr_t;
pub(crate) trait Submit {
fn cancel_inner(&mut self, target: Token) -> bool;
fn submit_send_tagged_inner(
&mut self,
ud: Token,
slot: FdSlot,
ptr: *const u8,
len: u32,
) -> bool;
fn complete_io(&mut self, ud: Token, rc: isize) -> bool;
fn complete_create(&mut self, ud: Token, result: i32, slot: FdSlot) -> bool;
fn submit_write_fd_inner(
&mut self,
ud: Token,
fd: RawFd,
ptr: *const u8,
len: u32,
offset: u64,
) -> bool;
fn submit_openat_inner(
&mut self,
ud: Token,
dir: RawFd,
path: *const c_char,
flags: i32,
mode: u32,
) -> bool;
fn submit_read_inner(
&mut self,
ud: Token,
fd: RawFd,
ptr: *mut u8,
len: u32,
offset: u64,
) -> bool;
fn submit_socket_at(
&mut self,
domain: i32,
socket_type: i32,
protocol: i32,
slot: FdSlot,
ud: Token,
) -> bool;
fn submit_connect(
&mut self,
slot: FdSlot,
addr_ptr: *const sockaddr,
addr_len: u32,
ud: Token,
) -> bool;
unsafe fn submit_send_msg_tagged_inner(
&mut self,
ud: Token,
slot: FdSlot,
msg: *const msghdr,
) -> bool;
}
impl Submit for Kqueue {
fn cancel_inner(&mut self, target: Token) -> bool {
match (target.raw() >> KIND_SHIFT) as u8 {
ACCEPT => self.cancel_accept_inner(target),
RECV | RECV_DISCARD => self.cancel_recv_inner(target),
SEND | CONNECT => self.cancel_write_inner(target),
TIMER => {
self.changes.push(kevent {
ident: target.raw() as uintptr_t,
filter: EVFILT_TIMER,
flags: EV_DELETE,
fflags: 0,
data: 0,
udata: null_mut(),
});
self.flush_changes_if_full();
true
}
_ => true,
}
}
fn submit_send_tagged_inner(
&mut self,
ud: Token,
slot: FdSlot,
ptr: *const u8,
len: u32,
) -> bool {
let Some(raw) = self.raw_fd(slot) else {
self.push_pending(PendingCompletion::Write {
ud,
result: -EBADF,
});
return true;
};
if self.write_retry_fd.contains_key(&(raw as usize)) {
return false;
}
let n = unsafe { send(raw, ptr.cast(), len as usize, 0) };
if n >= 0 {
self.push_pending(PendingCompletion::Write {
ud,
result: n as i32,
});
return true;
}
let errno = Errno::last();
if errno.is_block() {
return self.arm_write_retry(raw, ud, WriteKind::Send { ptr, len });
}
self.push_pending(PendingCompletion::Write {
ud,
result: -errno.raw(),
});
true
}
fn complete_io(&mut self, ud: Token, rc: isize) -> bool {
let result = if rc < 0 {
-Errno::last().raw()
} else {
rc as i32
};
self.push_pending(PendingCompletion::Write { ud, result });
true
}
fn complete_create(&mut self, ud: Token, result: i32, slot: FdSlot) -> bool {
self.push_pending(PendingCompletion::Create {
ud,
result,
slot: Some(slot),
});
true
}
fn submit_write_fd_inner(
&mut self,
ud: Token,
fd: RawFd,
ptr: *const u8,
len: u32,
offset: u64,
) -> bool {
let n = unsafe { pwrite(fd, ptr.cast(), len as usize, offset as off_t) };
self.complete_io(ud, n)
}
fn submit_openat_inner(
&mut self,
ud: Token,
dir: RawFd,
path: *const c_char,
flags: i32,
mode: u32,
) -> bool {
let fd = unsafe { openat(dir, path, flags, mode as c_uint) };
self.complete_io(ud, fd as isize)
}
fn submit_read_inner(
&mut self,
ud: Token,
fd: RawFd,
ptr: *mut u8,
len: u32,
offset: u64,
) -> bool {
let n = unsafe { pread(fd, ptr.cast(), len as usize, offset as off_t) };
self.complete_io(ud, n)
}
fn submit_socket_at(
&mut self,
domain: i32,
socket_type: i32,
protocol: i32,
slot: FdSlot,
ud: Token,
) -> bool {
let raw = unsafe { socket(domain, socket_type, protocol) };
if raw < 0 {
return self.complete_create(ud, -Errno::last().raw(), slot);
}
let sock = Handle::from_owned(unsafe { OwnedFd::from_raw_fd(raw) });
let result = if sock.set_cloexec().is_err()
|| sock.set_nonblocking().is_err()
|| Driver::set_no_sigpipe(&sock).is_err()
{
-Errno::last().raw()
} else {
self.register_fd(slot.raw(), sock.into_owned());
0
};
self.complete_create(ud, result, slot)
}
fn submit_connect(
&mut self,
slot: FdSlot,
addr_ptr: *const sockaddr,
addr_len: u32,
ud: Token,
) -> bool {
let Some(raw) = self.raw_fd(slot) else {
self.push_pending(PendingCompletion::Write {
ud,
result: -EBADF,
});
return true;
};
let rc = unsafe { connect(raw, addr_ptr, addr_len as socklen_t) };
if rc == 0 {
self.push_pending(PendingCompletion::Write { ud, result: 0 });
return true;
}
let errno = Errno::last();
if errno.raw() == EINPROGRESS || errno.is_block() {
return self.arm_write_retry(raw, ud, WriteKind::Connect { addr_ptr, addr_len });
}
self.push_pending(PendingCompletion::Write {
ud,
result: -errno.raw(),
});
true
}
unsafe fn submit_send_msg_tagged_inner(
&mut self,
ud: Token,
slot: FdSlot,
msg: *const msghdr,
) -> bool {
let Some(raw) = self.raw_fd(slot) else {
self.push_pending(PendingCompletion::Write {
ud,
result: -EBADF,
});
return true;
};
if self.write_retry_fd.contains_key(&(raw as usize)) {
return false;
}
let n = unsafe { sendmsg(raw, msg, 0) };
if n >= 0 {
self.push_pending(PendingCompletion::Write {
ud,
result: n as i32,
});
return true;
}
let errno = Errno::last();
if errno.is_block() {
return self.arm_write_retry(raw, ud, WriteKind::SendMsg { msg });
}
self.push_pending(PendingCompletion::Write {
ud,
result: -errno.raw(),
});
true
}
}