pub mod browser;
pub mod interest;
pub mod lab;
mod registration;
pub mod source;
pub mod token;
#[cfg(any(target_os = "linux", target_os = "android"))]
pub mod epoll;
#[cfg(any(target_os = "linux", target_os = "android"))]
#[path = "io_uring.rs"]
pub mod uring;
#[cfg(any(
target_os = "macos",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly"
))]
pub mod kqueue;
#[cfg(target_os = "windows")]
pub mod windows;
pub use browser::{BrowserReactor, BrowserReactorConfig};
pub use interest::Interest;
pub use lab::{FaultConfig, LabReactor};
#[allow(unused_imports)]
pub(crate) use registration::ReactorHandle;
pub use registration::Registration;
pub use source::{Source, SourceId, SourceWrapper, next_source_id};
pub use token::{SlabToken, TokenSlab};
#[cfg(any(target_os = "linux", target_os = "android"))]
pub use epoll::EpollReactor;
#[cfg(target_os = "windows")]
pub use windows::IocpReactor;
#[cfg(any(
target_os = "macos",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly"
))]
pub use kqueue::KqueueReactor;
use std::io;
use std::sync::Arc;
use std::time::Duration;
#[cfg(any(target_os = "linux", target_os = "android"))]
pub use uring::IoUringReactor;
use smallvec::SmallVec;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct Token(pub usize);
impl Token {
#[must_use]
pub const fn new(val: usize) -> Self {
Self(val)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Event {
pub token: Token,
pub ready: Interest,
}
impl Event {
#[must_use]
pub const fn new(token: Token, ready: Interest) -> Self {
Self { token, ready }
}
#[must_use]
pub const fn readable(token: Token) -> Self {
Self {
token,
ready: Interest::READABLE,
}
}
#[must_use]
pub const fn writable(token: Token) -> Self {
Self {
token,
ready: Interest::WRITABLE,
}
}
#[must_use]
pub const fn errored(token: Token) -> Self {
Self {
token,
ready: Interest::ERROR,
}
}
#[must_use]
pub const fn hangup(token: Token) -> Self {
Self {
token,
ready: Interest::HUP,
}
}
#[must_use]
pub const fn is_readable(&self) -> bool {
self.ready.is_readable()
}
#[must_use]
pub const fn is_writable(&self) -> bool {
self.ready.is_writable()
}
#[must_use]
pub const fn is_error(&self) -> bool {
self.ready.is_error()
}
#[must_use]
pub const fn is_hangup(&self) -> bool {
self.ready.is_hup()
}
}
#[derive(Debug, Default)]
pub struct Events {
inner: SmallVec<[Event; 16]>,
capacity: usize,
}
impl Events {
#[must_use]
pub fn with_capacity(capacity: usize) -> Self {
Self {
inner: SmallVec::with_capacity(capacity),
capacity,
}
}
pub fn clear(&mut self) {
self.inner.clear();
}
pub(crate) fn push(&mut self, event: Event) {
self.inner.push(event);
self.capacity = self.capacity.max(self.inner.len());
}
#[must_use]
pub fn len(&self) -> usize {
self.inner.len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.inner.is_empty()
}
#[must_use]
pub fn capacity(&self) -> usize {
self.capacity
}
pub fn iter(&self) -> std::slice::Iter<'_, Event> {
self.inner.iter()
}
}
impl<'a> IntoIterator for &'a Events {
type Item = &'a Event;
type IntoIter = std::slice::Iter<'a, Event>;
fn into_iter(self) -> Self::IntoIter {
self.iter()
}
}
impl IntoIterator for Events {
type Item = Event;
type IntoIter = smallvec::IntoIter<[Event; 16]>;
fn into_iter(self) -> Self::IntoIter {
self.inner.into_iter()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum IoReactorBackend {
Unavailable,
IoUring,
Epoll,
Kqueue,
Iocp,
Browser,
Injected,
}
impl IoReactorBackend {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Unavailable => "unavailable",
Self::IoUring => "io_uring",
Self::Epoll => "epoll",
Self::Kqueue => "kqueue",
Self::Iocp => "iocp",
Self::Browser => "browser",
Self::Injected => "injected",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[repr(u8)]
pub enum IoUringCapability {
FixedBuffers = 0,
ProvidedGroups = 1,
MappedBufferRing = 2,
MultishotAccept = 3,
MultishotRecv = 4,
SqPoll = 5,
}
impl IoUringCapability {
pub const ALL: [Self; 6] = [
Self::FixedBuffers,
Self::ProvidedGroups,
Self::MappedBufferRing,
Self::MultishotAccept,
Self::MultishotRecv,
Self::SqPoll,
];
#[must_use]
pub const fn id(self) -> &'static str {
match self {
Self::FixedBuffers => "URING-CAP-FIXED-BUFFERS",
Self::ProvidedGroups => "URING-CAP-PROVIDED-GROUPS",
Self::MappedBufferRing => "URING-CAP-MAPPED-BUFFER-RING",
Self::MultishotAccept => "URING-CAP-MULTISHOT-ACCEPT",
Self::MultishotRecv => "URING-CAP-MULTISHOT-RECV",
Self::SqPoll => "URING-CAP-SQPOLL",
}
}
const fn index(self) -> usize {
self as usize
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum IoUringSupportState {
Supported,
Unsupported,
NotProbed,
}
impl IoUringSupportState {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Supported => "SUPPORTED",
Self::Unsupported => "UNSUPPORTED",
Self::NotProbed => "NOT_PROBED",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum IoUringProbeOutcome {
Supported,
Unsupported,
Permission,
Resource,
Dependency,
Error,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct IoUringCapabilityPolicy {
requested: u8,
forced_off: u8,
}
impl IoUringCapabilityPolicy {
#[must_use]
pub const fn new() -> Self {
Self {
requested: 0,
forced_off: 0,
}
}
const fn bit(capability: IoUringCapability) -> u8 {
1 << capability.index()
}
#[must_use]
pub const fn with_requested(mut self, capability: IoUringCapability, requested: bool) -> Self {
let bit = Self::bit(capability);
if requested {
self.requested |= bit;
} else {
self.requested &= !bit;
}
self
}
#[must_use]
pub const fn with_forced_off(
mut self,
capability: IoUringCapability,
forced_off: bool,
) -> Self {
let bit = Self::bit(capability);
if forced_off {
self.forced_off |= bit;
} else {
self.forced_off &= !bit;
}
self
}
#[must_use]
pub const fn is_requested(self, capability: IoUringCapability) -> bool {
self.requested & Self::bit(capability) != 0
}
#[must_use]
pub const fn is_forced_off(self, capability: IoUringCapability) -> bool {
self.forced_off & Self::bit(capability) != 0
}
#[must_use]
pub const fn decide(
self,
capability: IoUringCapability,
probe: Option<IoUringProbeOutcome>,
) -> IoUringCapabilityDecision {
if !self.is_requested(capability) {
return IoUringCapabilityDecision::not_requested(capability);
}
if self.is_forced_off(capability) {
return IoUringCapabilityDecision::new(
capability,
true,
IoUringSupportState::NotProbed,
false,
IoUringFallbackReason::ForcedOff,
);
}
match probe {
Some(IoUringProbeOutcome::Supported) => IoUringCapabilityDecision::new(
capability,
true,
IoUringSupportState::Supported,
true,
IoUringFallbackReason::None,
),
Some(IoUringProbeOutcome::Unsupported) => IoUringCapabilityDecision::new(
capability,
true,
IoUringSupportState::Unsupported,
false,
IoUringFallbackReason::OperationUnsupported,
),
Some(IoUringProbeOutcome::Permission) => IoUringCapabilityDecision::new(
capability,
true,
IoUringSupportState::NotProbed,
false,
IoUringFallbackReason::Permission,
),
Some(IoUringProbeOutcome::Resource) => IoUringCapabilityDecision::new(
capability,
true,
IoUringSupportState::NotProbed,
false,
IoUringFallbackReason::Resource,
),
Some(IoUringProbeOutcome::Dependency) => IoUringCapabilityDecision::new(
capability,
true,
IoUringSupportState::NotProbed,
false,
IoUringFallbackReason::Dependency,
),
Some(IoUringProbeOutcome::Error) | None => IoUringCapabilityDecision::new(
capability,
true,
IoUringSupportState::NotProbed,
false,
IoUringFallbackReason::ProbeError,
),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum IoUringFallbackReason {
None,
NotRequested,
FeatureDisabled,
TargetUnsupported,
ForcedOff,
RingCreate,
ReactorUnavailable,
OperationUnsupported,
Permission,
Resource,
ProbeError,
Dependency,
NoWin,
Unknown,
}
impl IoUringFallbackReason {
#[must_use]
pub const fn id(self) -> &'static str {
match self {
Self::None => "URING-FB-NONE",
Self::NotRequested => "URING-FB-NOT-REQUESTED",
Self::FeatureDisabled => "URING-FB-FEATURE-DISABLED",
Self::TargetUnsupported => "URING-FB-TARGET-UNSUPPORTED",
Self::ForcedOff => "URING-FB-FORCED-OFF",
Self::RingCreate => "URING-FB-RING-CREATE",
Self::ReactorUnavailable => "URING-FB-REACTOR-UNAVAILABLE",
Self::OperationUnsupported => "URING-FB-OP-UNSUPPORTED",
Self::Permission => "URING-FB-PERMISSION",
Self::Resource => "URING-FB-RESOURCE",
Self::ProbeError => "URING-FB-PROBE-ERROR",
Self::Dependency => "URING-FB-DEPENDENCY",
Self::NoWin => "URING-FB-NO-WIN",
Self::Unknown => "URING-FB-UNKNOWN",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct IoUringCapabilityDecision {
capability: IoUringCapability,
requested: bool,
supported: IoUringSupportState,
active: bool,
fallback_reason: IoUringFallbackReason,
}
impl IoUringCapabilityDecision {
#[must_use]
pub const fn new(
capability: IoUringCapability,
requested: bool,
supported: IoUringSupportState,
active: bool,
fallback_reason: IoUringFallbackReason,
) -> Self {
let valid = match fallback_reason {
IoUringFallbackReason::None => {
requested && matches!(supported, IoUringSupportState::Supported) && active
}
IoUringFallbackReason::NotRequested => {
!requested && matches!(supported, IoUringSupportState::NotProbed) && !active
}
IoUringFallbackReason::TargetUnsupported
| IoUringFallbackReason::OperationUnsupported => {
requested && matches!(supported, IoUringSupportState::Unsupported) && !active
}
IoUringFallbackReason::NoWin => {
requested && matches!(supported, IoUringSupportState::Supported) && !active
}
IoUringFallbackReason::FeatureDisabled
| IoUringFallbackReason::ForcedOff
| IoUringFallbackReason::RingCreate
| IoUringFallbackReason::ReactorUnavailable
| IoUringFallbackReason::Permission
| IoUringFallbackReason::Resource
| IoUringFallbackReason::ProbeError
| IoUringFallbackReason::Dependency => {
requested && matches!(supported, IoUringSupportState::NotProbed) && !active
}
IoUringFallbackReason::Unknown => !active,
};
Self {
capability,
requested,
supported,
active: if valid { active } else { false },
fallback_reason: if valid {
fallback_reason
} else {
IoUringFallbackReason::Unknown
},
}
}
#[must_use]
pub const fn not_requested(capability: IoUringCapability) -> Self {
Self::new(
capability,
false,
IoUringSupportState::NotProbed,
false,
IoUringFallbackReason::NotRequested,
)
}
#[must_use]
pub const fn capability(self) -> IoUringCapability {
self.capability
}
#[must_use]
pub const fn requested(self) -> bool {
self.requested
}
#[must_use]
pub const fn supported(self) -> IoUringSupportState {
self.supported
}
#[must_use]
pub const fn active(self) -> bool {
self.active
}
#[must_use]
pub const fn fallback_reason(self) -> IoUringFallbackReason {
self.fallback_reason
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IoReactorCapabilitySnapshot {
backend: IoReactorBackend,
selection_fallback: Option<IoUringFallbackReason>,
decisions: [IoUringCapabilityDecision; 6],
}
impl IoReactorCapabilitySnapshot {
#[must_use]
pub fn from_policy(
backend: IoReactorBackend,
selection_fallback: Option<IoUringFallbackReason>,
policy: IoUringCapabilityPolicy,
probes: [Option<IoUringProbeOutcome>; 6],
) -> Self {
let decisions = IoUringCapability::ALL
.map(|capability| policy.decide(capability, probes[capability.index()]));
Self {
backend,
selection_fallback,
decisions,
}
}
fn unavailable(
backend: IoReactorBackend,
selection_fallback: Option<IoUringFallbackReason>,
policy: IoUringCapabilityPolicy,
reason: IoUringFallbackReason,
) -> Self {
let decisions = IoUringCapability::ALL.map(|capability| {
if !policy.is_requested(capability) {
return IoUringCapabilityDecision::not_requested(capability);
}
if policy.is_forced_off(capability) {
return policy.decide(capability, None);
}
let supported = if matches!(reason, IoUringFallbackReason::TargetUnsupported) {
IoUringSupportState::Unsupported
} else {
IoUringSupportState::NotProbed
};
IoUringCapabilityDecision::new(capability, true, supported, false, reason)
});
Self {
backend,
selection_fallback,
decisions,
}
}
#[must_use]
pub const fn not_requested(
backend: IoReactorBackend,
selection_fallback: Option<IoUringFallbackReason>,
) -> Self {
Self {
backend,
selection_fallback,
decisions: [
IoUringCapabilityDecision::not_requested(IoUringCapability::FixedBuffers),
IoUringCapabilityDecision::not_requested(IoUringCapability::ProvidedGroups),
IoUringCapabilityDecision::not_requested(IoUringCapability::MappedBufferRing),
IoUringCapabilityDecision::not_requested(IoUringCapability::MultishotAccept),
IoUringCapabilityDecision::not_requested(IoUringCapability::MultishotRecv),
IoUringCapabilityDecision::not_requested(IoUringCapability::SqPoll),
],
}
}
#[must_use]
pub const fn injected() -> Self {
Self::not_requested(IoReactorBackend::Injected, None)
}
#[must_use]
pub const fn no_reactor() -> Self {
Self::not_requested(IoReactorBackend::Unavailable, None)
}
#[must_use]
pub(crate) fn reactor_unavailable(policy: IoUringCapabilityPolicy) -> Self {
Self::unavailable(
IoReactorBackend::Unavailable,
Some(IoUringFallbackReason::ReactorUnavailable),
policy,
IoUringFallbackReason::ReactorUnavailable,
)
}
#[must_use]
pub const fn backend(self) -> IoReactorBackend {
self.backend
}
#[must_use]
pub const fn selection_fallback(self) -> Option<IoUringFallbackReason> {
self.selection_fallback
}
#[must_use]
pub const fn decision(self, capability: IoUringCapability) -> IoUringCapabilityDecision {
self.decisions[capability.index()]
}
#[must_use]
pub const fn decisions(self) -> [IoUringCapabilityDecision; 6] {
self.decisions
}
}
pub trait Reactor: Send + Sync {
#[must_use]
fn capability_snapshot(&self) -> IoReactorCapabilitySnapshot {
IoReactorCapabilitySnapshot::injected()
}
fn register(&self, source: &dyn Source, token: Token, interest: Interest) -> io::Result<()>;
fn modify(&self, token: Token, interest: Interest) -> io::Result<()>;
fn deregister(&self, token: Token) -> io::Result<()>;
fn poll(&self, events: &mut Events, timeout: Option<Duration>) -> io::Result<usize>;
fn wake(&self) -> io::Result<()>;
fn registration_count(&self) -> usize;
fn is_empty(&self) -> bool {
self.registration_count() == 0
}
}
struct ObservedReactor {
inner: Arc<dyn Reactor>,
snapshot: IoReactorCapabilitySnapshot,
}
impl ObservedReactor {
fn wrap(inner: Arc<dyn Reactor>, snapshot: IoReactorCapabilitySnapshot) -> Arc<dyn Reactor> {
Arc::new(Self { inner, snapshot })
}
}
impl Reactor for ObservedReactor {
fn capability_snapshot(&self) -> IoReactorCapabilitySnapshot {
self.snapshot
}
fn register(&self, source: &dyn Source, token: Token, interest: Interest) -> io::Result<()> {
self.inner.register(source, token, interest)
}
fn modify(&self, token: Token, interest: Interest) -> io::Result<()> {
self.inner.modify(token, interest)
}
fn deregister(&self, token: Token) -> io::Result<()> {
self.inner.deregister(token)
}
fn poll(&self, events: &mut Events, timeout: Option<Duration>) -> io::Result<usize> {
self.inner.poll(events, timeout)
}
fn wake(&self) -> io::Result<()> {
self.inner.wake()
}
fn registration_count(&self) -> usize {
self.inner.registration_count()
}
}
#[cfg(any(target_os = "linux", target_os = "android"))]
pub fn create_reactor() -> io::Result<Arc<dyn Reactor>> {
create_reactor_with_policy(IoUringCapabilityPolicy::default())
}
#[cfg(any(target_os = "linux", target_os = "android"))]
pub fn create_reactor_with_policy(policy: IoUringCapabilityPolicy) -> io::Result<Arc<dyn Reactor>> {
#[cfg(feature = "io-uring")]
{
if let Ok(reactor) = IoUringReactor::new() {
let probes = reactor.capability_probes(policy);
return Ok(ObservedReactor::wrap(
Arc::new(reactor),
IoReactorCapabilitySnapshot::from_policy(
IoReactorBackend::IoUring,
None,
policy,
probes,
),
));
}
Ok(ObservedReactor::wrap(
Arc::new(EpollReactor::new()?),
IoReactorCapabilitySnapshot::unavailable(
IoReactorBackend::Epoll,
Some(IoUringFallbackReason::RingCreate),
policy,
IoUringFallbackReason::RingCreate,
),
))
}
#[cfg(not(feature = "io-uring"))]
Ok(ObservedReactor::wrap(
Arc::new(EpollReactor::new()?),
IoReactorCapabilitySnapshot::unavailable(
IoReactorBackend::Epoll,
Some(IoUringFallbackReason::FeatureDisabled),
policy,
IoUringFallbackReason::FeatureDisabled,
),
))
}
#[cfg(any(
target_os = "macos",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly"
))]
pub fn create_reactor() -> io::Result<Arc<dyn Reactor>> {
create_reactor_with_policy(IoUringCapabilityPolicy::default())
}
#[cfg(any(
target_os = "macos",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly"
))]
pub fn create_reactor_with_policy(policy: IoUringCapabilityPolicy) -> io::Result<Arc<dyn Reactor>> {
Ok(ObservedReactor::wrap(
Arc::new(KqueueReactor::new()?),
IoReactorCapabilitySnapshot::unavailable(
IoReactorBackend::Kqueue,
None,
policy,
IoUringFallbackReason::TargetUnsupported,
),
))
}
#[cfg(target_os = "windows")]
pub fn create_reactor() -> io::Result<Arc<dyn Reactor>> {
create_reactor_with_policy(IoUringCapabilityPolicy::default())
}
#[cfg(target_os = "windows")]
pub fn create_reactor_with_policy(policy: IoUringCapabilityPolicy) -> io::Result<Arc<dyn Reactor>> {
Ok(ObservedReactor::wrap(
Arc::new(IocpReactor::new()?),
IoReactorCapabilitySnapshot::unavailable(
IoReactorBackend::Iocp,
None,
policy,
IoUringFallbackReason::TargetUnsupported,
),
))
}
#[cfg(target_arch = "wasm32")]
pub fn create_reactor() -> io::Result<Arc<dyn Reactor>> {
create_reactor_with_policy(IoUringCapabilityPolicy::default())
}
#[cfg(target_arch = "wasm32")]
pub fn create_reactor_with_policy(policy: IoUringCapabilityPolicy) -> io::Result<Arc<dyn Reactor>> {
Ok(ObservedReactor::wrap(
Arc::new(BrowserReactor::default()),
IoReactorCapabilitySnapshot::unavailable(
IoReactorBackend::Browser,
None,
policy,
IoUringFallbackReason::TargetUnsupported,
),
))
}
#[cfg(not(any(
target_os = "linux",
target_os = "android",
target_os = "macos",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly",
target_os = "windows",
target_arch = "wasm32"
)))]
pub fn create_reactor() -> io::Result<Arc<dyn Reactor>> {
create_reactor_with_policy(IoUringCapabilityPolicy::default())
}
#[cfg(not(any(
target_os = "linux",
target_os = "android",
target_os = "macos",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly",
target_os = "windows",
target_arch = "wasm32"
)))]
pub fn create_reactor_with_policy(
_policy: IoUringCapabilityPolicy,
) -> io::Result<Arc<dyn Reactor>> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
"no supported reactor backend for this platform",
))
}
#[cfg(test)]
mod tests {
#![allow(
clippy::pedantic,
clippy::nursery,
clippy::expect_fun_call,
clippy::map_unwrap_or,
clippy::cast_possible_wrap,
clippy::future_not_send
)]
use super::*;
use crate::test_utils::init_test_logging;
fn init_test(name: &str) {
init_test_logging();
crate::test_phase!(name);
}
#[test]
#[cfg(any(
target_os = "linux",
target_os = "android",
target_os = "macos",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly",
target_os = "windows"
))]
fn create_reactor_factory() {
init_test("create_reactor_factory");
let reactor = create_reactor().expect("failed to create reactor");
let snapshot = reactor.capability_snapshot();
crate::assert_with_log!(
reactor.is_empty(),
"reactor empty",
true,
reactor.is_empty()
);
crate::assert_with_log!(
reactor.registration_count() == 0,
"registration count",
0usize,
reactor.registration_count()
);
assert_ne!(snapshot.backend(), IoReactorBackend::Injected);
for capability in IoUringCapability::ALL {
assert_eq!(
snapshot.decision(capability),
IoUringCapabilityDecision::not_requested(capability)
);
}
crate::test_complete!("create_reactor_factory");
}
#[test]
fn io_uring_capability_ids_and_fallback_ids_are_stable() {
assert_eq!(IoReactorBackend::Unavailable.as_str(), "unavailable");
assert_eq!(
IoUringCapability::ALL.map(IoUringCapability::id),
[
"URING-CAP-FIXED-BUFFERS",
"URING-CAP-PROVIDED-GROUPS",
"URING-CAP-MAPPED-BUFFER-RING",
"URING-CAP-MULTISHOT-ACCEPT",
"URING-CAP-MULTISHOT-RECV",
"URING-CAP-SQPOLL",
]
);
assert_eq!(IoUringFallbackReason::None.id(), "URING-FB-NONE");
assert_eq!(
IoUringFallbackReason::NotRequested.id(),
"URING-FB-NOT-REQUESTED"
);
assert_eq!(
IoUringFallbackReason::FeatureDisabled.id(),
"URING-FB-FEATURE-DISABLED"
);
assert_eq!(
IoUringFallbackReason::TargetUnsupported.id(),
"URING-FB-TARGET-UNSUPPORTED"
);
assert_eq!(IoUringFallbackReason::ForcedOff.id(), "URING-FB-FORCED-OFF");
assert_eq!(
IoUringFallbackReason::RingCreate.id(),
"URING-FB-RING-CREATE"
);
assert_eq!(
IoUringFallbackReason::ReactorUnavailable.id(),
"URING-FB-REACTOR-UNAVAILABLE"
);
assert_eq!(
IoUringFallbackReason::OperationUnsupported.id(),
"URING-FB-OP-UNSUPPORTED"
);
assert_eq!(
IoUringFallbackReason::Permission.id(),
"URING-FB-PERMISSION"
);
assert_eq!(IoUringFallbackReason::Resource.id(), "URING-FB-RESOURCE");
assert_eq!(
IoUringFallbackReason::ProbeError.id(),
"URING-FB-PROBE-ERROR"
);
assert_eq!(
IoUringFallbackReason::Dependency.id(),
"URING-FB-DEPENDENCY"
);
assert_eq!(IoUringFallbackReason::NoWin.id(), "URING-FB-NO-WIN");
assert_eq!(IoUringFallbackReason::Unknown.id(), "URING-FB-UNKNOWN");
}
#[test]
fn terminal_reactor_snapshot_preserves_policy_without_probing() {
let policy = IoUringCapabilityPolicy::new()
.with_requested(IoUringCapability::FixedBuffers, true)
.with_requested(IoUringCapability::SqPoll, true)
.with_forced_off(IoUringCapability::SqPoll, true);
let snapshot = IoReactorCapabilitySnapshot::reactor_unavailable(policy);
assert_eq!(snapshot.backend(), IoReactorBackend::Unavailable);
assert_eq!(
snapshot.selection_fallback(),
Some(IoUringFallbackReason::ReactorUnavailable)
);
let requested = snapshot.decision(IoUringCapability::FixedBuffers);
assert!(requested.requested());
assert_eq!(requested.supported(), IoUringSupportState::NotProbed);
assert!(!requested.active());
assert_eq!(
requested.fallback_reason(),
IoUringFallbackReason::ReactorUnavailable
);
let forced_off = snapshot.decision(IoUringCapability::SqPoll);
assert!(forced_off.requested());
assert_eq!(forced_off.supported(), IoUringSupportState::NotProbed);
assert!(!forced_off.active());
assert_eq!(
forced_off.fallback_reason(),
IoUringFallbackReason::ForcedOff
);
let unrequested = snapshot.decision(IoUringCapability::MultishotAccept);
assert!(!unrequested.requested());
assert_eq!(unrequested.supported(), IoUringSupportState::NotProbed);
assert!(!unrequested.active());
assert_eq!(
unrequested.fallback_reason(),
IoUringFallbackReason::NotRequested
);
}
#[test]
fn io_uring_capability_decision_fails_inconsistent_tuples_closed() {
let capability = IoUringCapability::FixedBuffers;
let valid = IoUringCapabilityDecision::new(
capability,
true,
IoUringSupportState::Supported,
true,
IoUringFallbackReason::None,
);
assert!(valid.active());
assert_eq!(valid.fallback_reason(), IoUringFallbackReason::None);
let invalid = IoUringCapabilityDecision::new(
capability,
false,
IoUringSupportState::Supported,
true,
IoUringFallbackReason::None,
);
assert_eq!(invalid.capability(), capability);
assert!(!invalid.requested());
assert_eq!(invalid.supported(), IoUringSupportState::Supported);
assert!(!invalid.active());
assert_eq!(invalid.fallback_reason(), IoUringFallbackReason::Unknown);
}
#[test]
fn io_uring_capability_policy_precedence_is_deterministic() {
let requested = IoUringCapability::FixedBuffers;
let unrequested = IoUringCapability::ProvidedGroups;
let policy = IoUringCapabilityPolicy::new()
.with_requested(requested, true)
.with_forced_off(requested, true);
let forced = policy.decide(requested, Some(IoUringProbeOutcome::Supported));
assert!(forced.requested());
assert_eq!(forced.supported(), IoUringSupportState::NotProbed);
assert!(!forced.active());
assert_eq!(forced.fallback_reason(), IoUringFallbackReason::ForcedOff);
let skipped = policy.decide(unrequested, Some(IoUringProbeOutcome::Supported));
assert_eq!(
skipped,
IoUringCapabilityDecision::not_requested(unrequested)
);
let active = IoUringCapabilityPolicy::new()
.with_requested(requested, true)
.decide(requested, Some(IoUringProbeOutcome::Supported));
assert!(active.active());
assert_eq!(active.fallback_reason(), IoUringFallbackReason::None);
let dependent = IoUringCapability::MultishotRecv;
let dependency_blocked = IoUringCapabilityPolicy::new()
.with_requested(dependent, true)
.decide(dependent, Some(IoUringProbeOutcome::Dependency));
assert!(dependency_blocked.requested());
assert_eq!(
dependency_blocked.supported(),
IoUringSupportState::NotProbed
);
assert!(!dependency_blocked.active());
assert_eq!(
dependency_blocked.fallback_reason(),
IoUringFallbackReason::Dependency
);
}
#[test]
fn event_new() {
init_test("event_new");
let event = Event::new(Token::new(42), Interest::READABLE | Interest::WRITABLE);
crate::assert_with_log!(event.token.0 == 42, "token id", 42usize, event.token.0);
crate::assert_with_log!(
event.is_readable(),
"readable flag",
true,
event.is_readable()
);
crate::assert_with_log!(
event.is_writable(),
"writable flag",
true,
event.is_writable()
);
crate::assert_with_log!(
!event.is_error(),
"error flag unset",
false,
event.is_error()
);
crate::assert_with_log!(
!event.is_hangup(),
"hangup flag unset",
false,
event.is_hangup()
);
crate::test_complete!("event_new");
}
#[test]
fn event_readable() {
init_test("event_readable");
let event = Event::readable(Token::new(1));
crate::assert_with_log!(
event.is_readable(),
"readable flag",
true,
event.is_readable()
);
crate::assert_with_log!(
!event.is_writable(),
"writable flag unset",
false,
event.is_writable()
);
crate::assert_with_log!(
!event.is_error(),
"error flag unset",
false,
event.is_error()
);
crate::assert_with_log!(
!event.is_hangup(),
"hangup flag unset",
false,
event.is_hangup()
);
crate::test_complete!("event_readable");
}
#[test]
fn event_writable() {
init_test("event_writable");
let event = Event::writable(Token::new(2));
crate::assert_with_log!(
!event.is_readable(),
"readable flag unset",
false,
event.is_readable()
);
crate::assert_with_log!(
event.is_writable(),
"writable flag",
true,
event.is_writable()
);
crate::assert_with_log!(
!event.is_error(),
"error flag unset",
false,
event.is_error()
);
crate::assert_with_log!(
!event.is_hangup(),
"hangup flag unset",
false,
event.is_hangup()
);
crate::test_complete!("event_writable");
}
#[test]
fn event_errored() {
init_test("event_errored");
let event = Event::errored(Token::new(3));
crate::assert_with_log!(
!event.is_readable(),
"readable flag unset",
false,
event.is_readable()
);
crate::assert_with_log!(
!event.is_writable(),
"writable flag unset",
false,
event.is_writable()
);
crate::assert_with_log!(event.is_error(), "error flag", true, event.is_error());
crate::assert_with_log!(
!event.is_hangup(),
"hangup flag unset",
false,
event.is_hangup()
);
crate::test_complete!("event_errored");
}
#[test]
fn event_hangup() {
init_test("event_hangup");
let event = Event::hangup(Token::new(4));
crate::assert_with_log!(
!event.is_readable(),
"readable flag unset",
false,
event.is_readable()
);
crate::assert_with_log!(
!event.is_writable(),
"writable flag unset",
false,
event.is_writable()
);
crate::assert_with_log!(
!event.is_error(),
"error flag unset",
false,
event.is_error()
);
crate::assert_with_log!(event.is_hangup(), "hangup flag", true, event.is_hangup());
crate::test_complete!("event_hangup");
}
#[test]
fn event_combined_flags() {
init_test("event_combined_flags");
let event = Event::new(
Token::new(5),
Interest::READABLE | Interest::ERROR | Interest::HUP,
);
crate::assert_with_log!(
event.is_readable(),
"readable flag",
true,
event.is_readable()
);
crate::assert_with_log!(
!event.is_writable(),
"writable flag unset",
false,
event.is_writable()
);
crate::assert_with_log!(event.is_error(), "error flag", true, event.is_error());
crate::assert_with_log!(event.is_hangup(), "hangup flag", true, event.is_hangup());
crate::test_complete!("event_combined_flags");
}
#[test]
fn events_with_capacity() {
init_test("events_with_capacity");
let events = Events::with_capacity(64);
crate::assert_with_log!(
events.capacity() == 64,
"capacity",
64usize,
events.capacity()
);
crate::assert_with_log!(events.is_empty(), "len", 0usize, events.len());
crate::assert_with_log!(events.is_empty(), "is_empty", true, events.is_empty());
crate::test_complete!("events_with_capacity");
}
#[test]
fn events_push_and_iterate() {
init_test("events_push_and_iterate");
let mut events = Events::with_capacity(10);
events.push(Event::readable(Token::new(1)));
events.push(Event::writable(Token::new(2)));
events.push(Event::errored(Token::new(3)));
crate::assert_with_log!(events.len() == 3, "len", 3usize, events.len());
crate::assert_with_log!(!events.is_empty(), "not empty", false, events.is_empty());
let tokens: Vec<usize> = events.iter().map(|e| e.token.0).collect();
crate::assert_with_log!(
tokens == vec![1, 2, 3],
"tokens order",
vec![1, 2, 3],
tokens
);
crate::test_complete!("events_push_and_iterate");
}
#[test]
fn events_clear() {
init_test("events_clear");
let mut events = Events::with_capacity(10);
events.push(Event::readable(Token::new(1)));
events.push(Event::readable(Token::new(2)));
crate::assert_with_log!(events.len() == 2, "len before clear", 2usize, events.len());
events.clear();
crate::assert_with_log!(events.is_empty(), "len after clear", 0usize, events.len());
crate::assert_with_log!(
events.is_empty(),
"empty after clear",
true,
events.is_empty()
);
crate::assert_with_log!(
events.capacity() == 10,
"capacity maintained",
10usize,
events.capacity()
);
crate::test_complete!("events_clear");
}
#[test]
fn events_grow_beyond_capacity() {
init_test("events_grow_beyond_capacity");
let mut events = Events::with_capacity(3);
events.push(Event::readable(Token::new(1)));
events.push(Event::readable(Token::new(2)));
events.push(Event::readable(Token::new(3)));
events.push(Event::readable(Token::new(4)));
events.push(Event::readable(Token::new(5)));
crate::assert_with_log!(events.len() == 5, "len grew", 5usize, events.len());
crate::assert_with_log!(
events.capacity() >= events.len(),
"capacity tracks growth",
true,
events.capacity()
);
let tokens: Vec<usize> = events.iter().map(|e| e.token.0).collect();
crate::assert_with_log!(
tokens == vec![1, 2, 3, 4, 5],
"all tokens retained",
vec![1, 2, 3, 4, 5],
tokens
);
crate::test_complete!("events_grow_beyond_capacity");
}
#[test]
fn events_into_iter_ref() {
init_test("events_into_iter_ref");
let mut events = Events::with_capacity(10);
events.push(Event::readable(Token::new(1)));
events.push(Event::writable(Token::new(2)));
let mut count = 0;
for event in &events {
let ok = event.is_readable() || event.is_writable();
crate::assert_with_log!(ok, "event readable or writable", true, ok);
count += 1;
}
crate::assert_with_log!(count == 2, "iter count", 2usize, count);
crate::test_complete!("events_into_iter_ref");
}
#[test]
fn events_into_iter_owned() {
init_test("events_into_iter_owned");
let mut events = Events::with_capacity(10);
events.push(Event::readable(Token::new(1)));
events.push(Event::writable(Token::new(2)));
let collected: Vec<Event> = events.into_iter().collect();
crate::assert_with_log!(
collected.len() == 2,
"collected len",
2usize,
collected.len()
);
crate::assert_with_log!(
collected[0].is_readable(),
"first readable",
true,
collected[0].is_readable()
);
crate::assert_with_log!(
collected[1].is_writable(),
"second writable",
true,
collected[1].is_writable()
);
crate::test_complete!("events_into_iter_owned");
}
#[test]
fn events_zero_capacity() {
init_test("events_zero_capacity");
let mut events = Events::with_capacity(0);
crate::assert_with_log!(
events.capacity() == 0,
"capacity zero",
0usize,
events.capacity()
);
crate::assert_with_log!(events.is_empty(), "len zero", 0usize, events.len());
events.push(Event::readable(Token::new(1)));
crate::assert_with_log!(events.len() == 1, "len grew", 1usize, events.len());
crate::test_complete!("events_zero_capacity");
}
#[test]
fn token_new() {
init_test("token_new");
let token = Token::new(123);
crate::assert_with_log!(token.0 == 123, "token id", 123usize, token.0);
crate::test_complete!("token_new");
}
#[test]
fn token_equality() {
init_test("token_equality");
let t1 = Token::new(1);
let t2 = Token::new(1);
let t3 = Token::new(2);
crate::assert_with_log!(t1 == t2, "t1 == t2", t2, t1);
crate::assert_with_log!(t1 != t3, "t1 != t3", true, t1 != t3);
crate::test_complete!("token_equality");
}
#[test]
fn token_ordering() {
init_test("token_ordering");
let t1 = Token::new(1);
let t2 = Token::new(2);
crate::assert_with_log!(t1 < t2, "t1 < t2", true, t1 < t2);
crate::assert_with_log!(t2 > t1, "t2 > t1", true, t2 > t1);
crate::test_complete!("token_ordering");
}
fn assert_reactor_trait_bounds<R: Reactor + Send + Sync>() {}
#[test]
fn reactor_trait_bounds_epoll() {
init_test("reactor_trait_bounds_epoll");
#[cfg(any(target_os = "linux", target_os = "android"))]
assert_reactor_trait_bounds::<super::EpollReactor>();
crate::test_complete!("reactor_trait_bounds_epoll");
}
#[test]
fn reactor_trait_bounds_lab() {
init_test("reactor_trait_bounds_lab");
assert_reactor_trait_bounds::<super::LabReactor>();
crate::test_complete!("reactor_trait_bounds_lab");
}
#[test]
fn reactor_trait_bounds_browser() {
init_test("reactor_trait_bounds_browser");
assert_reactor_trait_bounds::<super::BrowserReactor>();
crate::test_complete!("reactor_trait_bounds_browser");
}
#[cfg(any(target_os = "linux", target_os = "android"))]
#[test]
fn reactor_trait_bounds_io_uring() {
init_test("reactor_trait_bounds_io_uring");
assert_reactor_trait_bounds::<super::IoUringReactor>();
crate::test_complete!("reactor_trait_bounds_io_uring");
}
fn compliance_check_empty_state(reactor: &dyn Reactor, name: &str) {
crate::assert_with_log!(
reactor.is_empty(),
&format!("{name} starts empty"),
true,
reactor.is_empty()
);
crate::assert_with_log!(
reactor.registration_count() == 0,
&format!("{name} starts with zero registrations"),
0usize,
reactor.registration_count()
);
}
fn compliance_check_wake(reactor: &dyn Reactor, name: &str) {
let result = reactor.wake();
crate::assert_with_log!(
result.is_ok(),
&format!("{name} wake succeeds"),
true,
result.is_ok()
);
}
fn compliance_check_poll_nonblocking(reactor: &dyn Reactor, name: &str) {
let mut events = Events::with_capacity(16);
let result = reactor.poll(&mut events, Some(std::time::Duration::ZERO));
crate::assert_with_log!(
result.is_ok(),
&format!("{name} non-blocking poll succeeds"),
true,
result.is_ok()
);
crate::assert_with_log!(
events.is_empty(),
&format!("{name} no events on empty reactor"),
true,
events.is_empty()
);
}
fn compliance_check_deregister_unknown(reactor: &dyn Reactor, name: &str) {
let result = reactor.deregister(Token::new(99999));
crate::assert_with_log!(
result.is_err(),
&format!("{name} deregister unknown token fails"),
true,
result.is_err()
);
let kind = result.expect_err("checked above").kind();
crate::assert_with_log!(
kind == io::ErrorKind::NotFound,
&format!("{name} deregister unknown token reports NotFound"),
io::ErrorKind::NotFound,
kind
);
}
fn compliance_check_modify_unknown(reactor: &dyn Reactor, name: &str) {
let result = reactor.modify(Token::new(99999), Interest::READABLE);
crate::assert_with_log!(
result.is_err(),
&format!("{name} modify unknown token fails"),
true,
result.is_err()
);
let kind = result.expect_err("checked above").kind();
crate::assert_with_log!(
kind == io::ErrorKind::NotFound,
&format!("{name} modify unknown token reports NotFound"),
io::ErrorKind::NotFound,
kind
);
}
#[cfg(any(target_os = "linux", target_os = "android"))]
#[test]
fn cross_reactor_compliance_epoll() {
init_test("cross_reactor_compliance_epoll");
let reactor = super::EpollReactor::new().expect("failed to create epoll reactor");
let name = "EpollReactor";
compliance_check_empty_state(&reactor, name);
compliance_check_wake(&reactor, name);
compliance_check_poll_nonblocking(&reactor, name);
compliance_check_deregister_unknown(&reactor, name);
compliance_check_modify_unknown(&reactor, name);
crate::test_complete!("cross_reactor_compliance_epoll");
}
#[test]
fn cross_reactor_compliance_lab() {
init_test("cross_reactor_compliance_lab");
let reactor = super::LabReactor::new();
let name = "LabReactor";
compliance_check_empty_state(&reactor, name);
compliance_check_wake(&reactor, name);
compliance_check_poll_nonblocking(&reactor, name);
compliance_check_deregister_unknown(&reactor, name);
compliance_check_modify_unknown(&reactor, name);
crate::test_complete!("cross_reactor_compliance_lab");
}
#[test]
fn cross_reactor_compliance_browser() {
init_test("cross_reactor_compliance_browser");
let reactor = super::BrowserReactor::default();
let name = "BrowserReactor";
compliance_check_empty_state(&reactor, name);
compliance_check_wake(&reactor, name);
compliance_check_poll_nonblocking(&reactor, name);
compliance_check_deregister_unknown(&reactor, name);
compliance_check_modify_unknown(&reactor, name);
crate::test_complete!("cross_reactor_compliance_browser");
}
#[cfg(any(target_os = "linux", target_os = "android"))]
#[test]
fn cross_reactor_compliance_io_uring() {
init_test("cross_reactor_compliance_io_uring");
let reactor = match super::IoUringReactor::new() {
Ok(r) => r,
Err(e) => {
eprintln!("Skipping io_uring compliance test: {e}");
return;
}
};
let name = "IoUringReactor";
compliance_check_empty_state(&reactor, name);
compliance_check_wake(&reactor, name);
compliance_check_poll_nonblocking(&reactor, name);
compliance_check_deregister_unknown(&reactor, name);
compliance_check_modify_unknown(&reactor, name);
crate::test_complete!("cross_reactor_compliance_io_uring");
}
#[test]
fn reactor_as_trait_object() {
init_test("reactor_as_trait_object");
let lab = super::LabReactor::new();
let reactor: &dyn Reactor = &lab;
crate::assert_with_log!(
reactor.is_empty(),
"trait object is_empty",
true,
reactor.is_empty()
);
crate::assert_with_log!(
reactor.registration_count() == 0,
"trait object registration_count",
0usize,
reactor.registration_count()
);
crate::assert_with_log!(
reactor.wake().is_ok(),
"trait object wake",
true,
reactor.wake().is_ok()
);
crate::test_complete!("reactor_as_trait_object");
}
#[test]
fn reactor_arc_shared_access() {
init_test("reactor_arc_shared_access");
let reactor = std::sync::Arc::new(super::LabReactor::new());
let reactor_clone = std::sync::Arc::clone(&reactor);
crate::assert_with_log!(
reactor.is_empty(),
"arc reactor empty",
true,
reactor.is_empty()
);
crate::assert_with_log!(
reactor_clone.is_empty(),
"arc clone reactor empty",
true,
reactor_clone.is_empty()
);
crate::assert_with_log!(
reactor_clone.wake().is_ok(),
"wake from arc clone",
true,
reactor_clone.wake().is_ok()
);
crate::test_complete!("reactor_arc_shared_access");
}
#[test]
fn token_debug_clone_copy_hash_ord_eq() {
use std::collections::HashSet;
let t = Token::new(42);
let dbg = format!("{t:?}");
assert!(dbg.contains("42"), "{dbg}");
let copied: Token = t;
let cloned = t;
assert_eq!(copied, cloned);
assert!(Token::new(1) < Token::new(2));
let mut set = HashSet::new();
set.insert(Token::new(1));
set.insert(Token::new(2));
assert_eq!(set.len(), 2);
}
#[test]
fn event_debug_clone_copy_eq() {
let e = Event::new(Token::new(1), Interest::READABLE);
let dbg = format!("{e:?}");
assert!(dbg.contains("Event"), "{dbg}");
let copied: Event = e;
let cloned = e;
assert_eq!(copied, cloned);
assert_ne!(e, Event::new(Token::new(2), Interest::WRITABLE));
}
}