use slotmap::{new_key_type, SlotMap};
use std::cmp::Reverse;
use std::collections::{BinaryHeap, HashMap};
use std::io;
use std::os::unix::io::RawFd;
use thiserror::Error;
pub const E_READ: u32 = 0x1;
pub const E_WRITE: u32 = 0x2;
pub const EPSILON: f64 = 1.0e-3;
pub const FANOUT_UNLIMITED: i32 = -1;
#[derive(Error, Debug)]
pub enum EngineError {
#[error("Operation aborted (kill={kill})")]
Abort { kill: bool },
#[error("Operation timed out")]
Timeout,
#[error("Illegal operation: {0}")]
IllegalOperation(String),
#[error("Engine is already running")]
AlreadyRunning,
#[error("Not supported: {0}")]
NotSupported(String),
#[error("I/O error: {0}")]
Io(#[from] io::Error),
}
new_key_type! { pub struct TimerId; }
#[derive(Debug, Clone)]
pub struct EngineBaseTimer {
fire_delay: f64,
interval: f64,
autoclose: bool,
}
impl EngineBaseTimer {
pub fn new(fire_delay: f64, interval: f64, autoclose: bool) -> Self {
Self {
fire_delay,
interval,
autoclose,
}
}
pub fn invalidate(&mut self) {
self.fire_delay = -1.0;
}
pub fn is_valid(&self) -> bool {
self.fire_delay >= 0.0
}
pub fn set_nextfire(&mut self, fire_delay: f64, interval: f64) {
self.fire_delay = fire_delay;
self.interval = interval;
}
pub fn fire_delay(&self) -> f64 {
self.fire_delay
}
pub fn interval(&self) -> f64 {
self.interval
}
pub fn autoclose(&self) -> bool {
self.autoclose
}
}
#[derive(Debug, Clone, PartialEq)]
struct TimerEntry {
fire_date: f64,
timer_id: TimerId,
armed: bool,
}
impl Eq for TimerEntry {}
impl Ord for TimerEntry {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.fire_date
.partial_cmp(&other.fire_date)
.unwrap_or(std::cmp::Ordering::Equal)
.then_with(|| self.timer_id.cmp(&other.timer_id))
}
}
impl PartialOrd for TimerEntry {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
#[derive(Debug)]
pub struct TimerQueue {
timers: BinaryHeap<Reverse<TimerEntry>>,
timer_data: SlotMap<TimerId, EngineBaseTimer>,
armed_count: usize,
current_time: f64,
}
impl TimerQueue {
pub fn new() -> Self {
Self {
timers: BinaryHeap::new(),
timer_data: SlotMap::with_capacity_and_key(16),
armed_count: 0,
current_time: 0.0,
}
}
pub fn schedule(&mut self, timer_id: TimerId) {
if let Some(timer) = self.timer_data.get(timer_id) {
if timer.fire_delay >= 0.0 {
let fire_date = self.current_time + timer.fire_delay;
let entry = TimerEntry {
fire_date,
timer_id,
armed: true,
};
self.timers.push(Reverse(entry));
self.armed_count += 1;
}
}
}
pub fn reschedule(&mut self, timer_id: TimerId) {
self.invalidate(timer_id);
self.schedule(timer_id);
}
pub fn invalidate(&mut self, timer_id: TimerId) {
let mut found = false;
let mut new_heap = BinaryHeap::new();
for Reverse(entry) in self.timers.drain() {
if entry.timer_id == timer_id {
found = true;
if let Some(timer) = self.timer_data.get_mut(timer_id) {
timer.invalidate();
}
} else {
new_heap.push(Reverse(entry));
}
}
self.timers = new_heap;
if found {
self.armed_count = self.armed_count.saturating_sub(1);
}
}
pub fn fire_expired(&mut self) -> Vec<(TimerId, bool)> {
let mut fired = Vec::new();
let now = self.current_time;
while let Some(Reverse(entry)) = self.timers.peek() {
if entry.fire_date > now + EPSILON {
break;
}
let entry = self.timers.pop().unwrap().0;
let timer_id = entry.timer_id;
if let Some(timer) = self.timer_data.get(timer_id) {
if entry.armed && timer.fire_delay >= 0.0 {
fired.push((timer_id, timer.autoclose));
if timer.interval >= 0.0 {
let new_fire_date = now + timer.interval;
let new_entry = TimerEntry {
fire_date: new_fire_date,
timer_id,
armed: true,
};
self.timers.push(Reverse(new_entry));
} else {
if let Some(t) = self.timer_data.get_mut(timer_id) {
t.invalidate();
}
self.armed_count = self.armed_count.saturating_sub(1);
}
}
}
}
fired
}
pub fn nextfire_delay(&self) -> Option<f64> {
if let Some(Reverse(entry)) = self.timers.peek() {
let delay = entry.fire_date - self.current_time;
if delay > EPSILON {
Some(delay)
} else {
Some(0.0)
}
} else {
None
}
}
pub fn clear(&mut self) {
self.timers.clear();
self.timer_data.clear();
self.armed_count = 0;
}
pub fn add_timer(&mut self, timer: EngineBaseTimer) -> TimerId {
let timer_id = self.timer_data.insert(timer);
timer_id
}
pub fn remove_timer(&mut self, timer_id: TimerId) {
self.invalidate(timer_id);
self.timer_data.remove(timer_id);
}
pub fn armed_count(&self) -> usize {
self.armed_count
}
pub fn update_time(&mut self, time: f64) {
self.current_time = time;
}
pub fn is_valid(&self, timer_id: TimerId) -> bool {
self.timer_data
.get(timer_id)
.map(|t| t.is_valid())
.unwrap_or(false)
}
}
impl Default for TimerQueue {
fn default() -> Self {
Self::new()
}
}
pub struct EngineTimer {
base: EngineBaseTimer,
handler: Box<dyn FnMut() + Send>,
}
impl EngineTimer {
pub fn new(
fire_delay: f64,
interval: f64,
autoclose: bool,
handler: impl FnMut() + Send + 'static,
) -> Self {
Self {
base: EngineBaseTimer::new(fire_delay, interval, autoclose),
handler: Box::new(handler),
}
}
pub fn fire(&mut self) {
(self.handler)();
}
pub fn base(&self) -> &EngineBaseTimer {
&self.base
}
pub fn base_mut(&mut self) -> &mut EngineBaseTimer {
&mut self.base
}
}
impl std::fmt::Debug for EngineTimer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("EngineTimer")
.field("base", &self.base)
.field("handler", &"<fn>")
.finish()
}
}
pub trait EngineClient: Send {
fn start(&mut self) -> io::Result<()>;
fn close(&mut self, abort: bool, timeout: bool);
fn fd(&self) -> Option<RawFd>;
fn events(&self) -> u32;
fn is_delayable(&self) -> bool;
fn is_registered(&self) -> bool;
fn set_registered(&mut self, val: bool);
fn autoclose(&self) -> bool;
fn handle_read(&mut self) -> io::Result<usize> {
Ok(0)
}
fn handle_write(&mut self) -> io::Result<usize> {
Ok(0)
}
}
new_key_type! { pub struct ClientId; }
pub trait PollBackend: Send + Sync {
fn register(&mut self, fd: RawFd, events: u32, client_id: ClientId) -> io::Result<()>;
fn modify(&mut self, fd: RawFd, events: u32, client_id: ClientId) -> io::Result<()>;
fn unregister(&mut self, fd: RawFd) -> io::Result<()>;
fn wait(&mut self, timeout: Option<f64>) -> io::Result<Vec<(RawFd, u32, ClientId)>>;
fn name(&self) -> &str;
}
pub struct Engine<B: PollBackend> {
clients: SlotMap<ClientId, Box<dyn EngineClient>>,
timerq: TimerQueue,
reg_clifds: HashMap<RawFd, ClientId>,
poll_backend: B,
fanout: usize,
running: bool,
exited: bool,
evlooprefcnt: i32,
}
impl<B: PollBackend> Engine<B> {
pub fn with_backend(backend: B, fanout: usize) -> Self {
Self {
clients: SlotMap::with_key(),
timerq: TimerQueue::new(),
reg_clifds: HashMap::new(),
poll_backend: backend,
fanout,
running: false,
exited: false,
evlooprefcnt: 0,
}
}
pub fn add_timer(&mut self, timer: EngineBaseTimer) -> TimerId {
let timer_id = self.timerq.add_timer(timer);
self.timerq.schedule(timer_id);
timer_id
}
pub fn remove_timer(&mut self, id: TimerId) {
self.timerq.remove_timer(id);
}
pub fn fire_timers(&mut self) {
let _fired = self.timerq.fire_expired();
}
pub fn run(&mut self, timeout: Option<f64>) -> Result<(), EngineError> {
if self.running {
return Err(EngineError::AlreadyRunning);
}
self.running = true;
self.exited = false;
let start_time = self.timerq.current_time;
let max_time = timeout.map(|t| start_time + t);
while self.running {
self.timerq.update_time(self.timerq.current_time + 0.001);
self.fire_timers();
if !self.running {
break;
}
let next_timer_delay = self.timerq.nextfire_delay();
let poll_timeout = match (max_time, next_timer_delay) {
(Some(max), Some(timer_delay)) => {
let remaining = max - self.timerq.current_time;
if remaining <= EPSILON {
return Err(EngineError::Timeout);
}
Some(if remaining < timer_delay {
remaining
} else {
timer_delay
})
}
(Some(max), None) => {
let remaining = max - self.timerq.current_time;
if remaining <= EPSILON {
return Err(EngineError::Timeout);
}
Some(remaining)
}
(None, Some(timer_delay)) => Some(timer_delay),
(None, None) => None,
};
let events = self.poll_backend.wait(poll_timeout.map(|t| t.max(0.0)))?;
for (_fd, events_mask, client_id) in events {
if let Some(client) = self.clients.get_mut(client_id) {
if events_mask & E_READ != 0 {
let _ = client.handle_read();
}
if events_mask & E_WRITE != 0 {
let _ = client.handle_write();
}
}
}
if let Some(max) = max_time {
if self.timerq.current_time >= max {
return Err(EngineError::Timeout);
}
}
}
self.exited = true;
self.running = false;
Ok(())
}
pub fn abort(&mut self, kill: bool) {
if kill {
for (_, client) in self.clients.iter_mut() {
client.close(true, false);
}
}
self.running = false;
}
pub fn is_running(&self) -> bool {
self.running
}
pub fn exited(&self) -> bool {
self.exited
}
pub fn add_client(&mut self, client: Box<dyn EngineClient>) -> Result<ClientId, EngineError> {
let client_id = self.clients.insert(client);
if let Some(client) = self.clients.get(client_id) {
if let Some(fd) = client.fd() {
if fd >= 0 {
let events = client.events();
self.poll_backend.register(fd, events, client_id)?;
self.reg_clifds.insert(fd, client_id);
}
}
}
Ok(client_id)
}
pub fn remove_client(&mut self, id: ClientId) {
if let Some(client) = self.clients.get(id) {
if let Some(fd) = client.fd() {
if fd >= 0 {
let _ = self.poll_backend.unregister(fd);
self.reg_clifds.remove(&fd);
}
}
}
self.clients.remove(id);
}
pub fn fanout(&self) -> usize {
self.fanout
}
pub fn update_client_events(&mut self, client_id: ClientId) -> Result<(), EngineError> {
if let Some(client) = self.clients.get(client_id) {
if let Some(fd) = client.fd() {
if fd >= 0 {
let events = client.events();
if self.reg_clifds.contains_key(&fd) {
self.poll_backend.modify(fd, events, client_id)?;
} else {
self.poll_backend.register(fd, events, client_id)?;
self.reg_clifds.insert(fd, client_id);
}
}
}
}
Ok(())
}
pub fn stop(&mut self) {
self.running = false;
}
pub fn evloop_acquire(&mut self) {
self.evlooprefcnt += 1;
}
pub fn evloop_release(&mut self) {
self.evlooprefcnt -= 1;
if self.evlooprefcnt <= 0 {
self.running = false;
}
}
pub fn client_count(&self) -> usize {
self.clients.len()
}
pub fn timerq(&self) -> &TimerQueue {
&self.timerq
}
pub fn timerq_mut(&mut self) -> &mut TimerQueue {
&mut self.timerq
}
pub fn start_all_clients(&mut self) -> io::Result<()> {
let ids: Vec<ClientId> = self.clients.keys().collect();
for id in ids {
if let Some(client) = self.clients.get_mut(id) {
client.start()?;
}
}
Ok(())
}
}
#[derive(Debug, Default)]
pub struct NullBackend;
impl PollBackend for NullBackend {
fn register(&mut self, _fd: RawFd, _events: u32, _client_id: ClientId) -> io::Result<()> {
Ok(())
}
fn modify(&mut self, _fd: RawFd, _events: u32, _client_id: ClientId) -> io::Result<()> {
Ok(())
}
fn unregister(&mut self, _fd: RawFd) -> io::Result<()> {
Ok(())
}
fn wait(&mut self, _timeout: Option<f64>) -> io::Result<Vec<(RawFd, u32, ClientId)>> {
Ok(Vec::new())
}
fn name(&self) -> &str {
"null"
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_engine_constants() {
assert_eq!(E_READ, 0x1);
assert_eq!(E_WRITE, 0x2);
assert!((EPSILON - 1.0e-3).abs() < EPSILON);
assert_eq!(FANOUT_UNLIMITED, -1);
}
#[test]
fn test_engine_base_timer() {
let mut timer = EngineBaseTimer::new(5.0, -1.0, false);
assert!(timer.is_valid());
assert_eq!(timer.fire_delay(), 5.0);
assert_eq!(timer.interval(), -1.0);
timer.invalidate();
assert!(!timer.is_valid());
assert_eq!(timer.fire_delay(), -1.0);
timer.set_nextfire(10.0, 2.0);
assert!(timer.is_valid());
assert_eq!(timer.fire_delay(), 10.0);
assert_eq!(timer.interval(), 2.0);
}
#[test]
fn test_timer_queue_scheduling() {
let mut timerq = TimerQueue::new();
let timer1 = timerq.add_timer(EngineBaseTimer::new(1.0, -1.0, false));
let timer2 = timerq.add_timer(EngineBaseTimer::new(2.0, -1.0, false));
let timer3 = timerq.add_timer(EngineBaseTimer::new(0.5, -1.0, false));
timerq.schedule(timer1);
timerq.schedule(timer2);
timerq.schedule(timer3);
assert_eq!(timerq.armed_count(), 3);
let delay = timerq.nextfire_delay();
assert!(delay.is_some());
assert!((delay.unwrap() - 0.5).abs() < EPSILON);
}
#[test]
fn test_timer_queue_firing() {
let mut timerq = TimerQueue::new();
let timer1 = timerq.add_timer(EngineBaseTimer::new(0.1, -1.0, false));
let timer2 = timerq.add_timer(EngineBaseTimer::new(0.2, -1.0, false));
timerq.schedule(timer1);
timerq.schedule(timer2);
timerq.update_time(0.3);
let fired = timerq.fire_expired();
assert_eq!(fired.len(), 2);
assert!(!timerq.is_valid(timer1));
assert!(!timerq.is_valid(timer2));
}
#[test]
fn test_timer_queue_invalidation() {
let mut timerq = TimerQueue::new();
let timer1 = timerq.add_timer(EngineBaseTimer::new(1.0, -1.0, false));
let timer2 = timerq.add_timer(EngineBaseTimer::new(2.0, -1.0, false));
timerq.schedule(timer1);
timerq.schedule(timer2);
timerq.invalidate(timer1);
assert_eq!(timerq.armed_count(), 1);
let delay = timerq.nextfire_delay();
assert!(delay.is_some());
assert!((delay.unwrap() - 2.0).abs() < EPSILON);
}
#[test]
fn test_timer_queue_repeat() {
let mut timerq = TimerQueue::new();
let timer = timerq.add_timer(EngineBaseTimer::new(0.1, 0.1, false));
timerq.schedule(timer);
timerq.update_time(0.15);
let fired = timerq.fire_expired();
assert_eq!(fired.len(), 1);
assert!(timerq.is_valid(timer));
timerq.update_time(0.25);
let fired = timerq.fire_expired();
assert_eq!(fired.len(), 1);
assert!(timerq.is_valid(timer));
}
#[test]
fn test_engine_error_types() {
let err = EngineError::Abort { kill: true };
assert!(matches!(err, EngineError::Abort { kill: true }));
let err = EngineError::Timeout;
assert!(matches!(err, EngineError::Timeout));
let err = EngineError::IllegalOperation("test".to_string());
assert!(matches!(err, EngineError::IllegalOperation(_)));
let err = EngineError::AlreadyRunning;
assert!(matches!(err, EngineError::AlreadyRunning));
let err = EngineError::NotSupported("test".to_string());
assert!(matches!(err, EngineError::NotSupported(_)));
let io_err = io::Error::new(io::ErrorKind::NotFound, "not found");
let err = EngineError::Io(io_err);
assert!(matches!(err, EngineError::Io(_)));
}
#[test]
fn test_null_backend_name() {
let nb = NullBackend;
assert_eq!(nb.name(), "null");
}
#[test]
fn test_null_backend_wait_returns_empty() {
let mut nb = NullBackend;
let events = nb.wait(Some(0.0)).unwrap();
assert!(events.is_empty());
}
#[test]
fn test_null_backend_register_modify_unregister() {
let mut nb = NullBackend;
let mut sm: SlotMap<ClientId, ()> = SlotMap::with_key();
let cid = sm.insert(());
nb.register(0, E_READ, cid).unwrap();
nb.modify(0, E_WRITE, cid).unwrap();
nb.unregister(0).unwrap();
}
struct DummyClient {
registered: bool,
started: bool,
}
impl DummyClient {
fn new() -> Self {
Self {
registered: false,
started: false,
}
}
}
impl EngineClient for DummyClient {
fn start(&mut self) -> io::Result<()> {
self.started = true;
Ok(())
}
fn close(&mut self, _abort: bool, _timeout: bool) {}
fn fd(&self) -> Option<RawFd> {
None }
fn events(&self) -> u32 {
E_READ
}
fn is_delayable(&self) -> bool {
false
}
fn is_registered(&self) -> bool {
self.registered
}
fn set_registered(&mut self, val: bool) {
self.registered = val;
}
fn autoclose(&self) -> bool {
false
}
}
#[test]
fn test_engine_creation() {
let engine = Engine::with_backend(NullBackend, 0);
assert!(!engine.is_running());
assert!(!engine.exited());
assert_eq!(engine.fanout(), 0);
assert_eq!(engine.client_count(), 0);
}
#[test]
fn test_engine_add_remove_client() {
let mut engine = Engine::with_backend(NullBackend, 0);
let cid = engine.add_client(Box::new(DummyClient::new())).unwrap();
assert_eq!(engine.client_count(), 1);
engine.remove_client(cid);
assert_eq!(engine.client_count(), 0);
}
#[test]
fn test_engine_add_timer() {
let mut engine = Engine::with_backend(NullBackend, 0);
let tid = engine.add_timer(EngineBaseTimer::new(1.0, -1.0, false));
assert!(engine.timerq().is_valid(tid));
engine.remove_timer(tid);
assert!(!engine.timerq().is_valid(tid));
}
#[test]
fn test_engine_fire_timers() {
let mut engine = Engine::with_backend(NullBackend, 0);
let tid = engine.add_timer(EngineBaseTimer::new(0.1, -1.0, false));
assert!(engine.timerq().is_valid(tid));
engine.timerq_mut().update_time(1.0);
engine.fire_timers();
assert!(!engine.timerq().is_valid(tid));
}
#[test]
fn test_engine_abort() {
let mut engine = Engine::with_backend(NullBackend, 0);
engine.abort(false);
assert!(!engine.is_running());
}
#[test]
fn test_engine_evloop_refcnt() {
let mut engine = Engine::with_backend(NullBackend, 0);
engine.evloop_acquire();
engine.evloop_acquire();
engine.evloop_release();
engine.evloop_release();
assert!(!engine.is_running());
}
#[test]
fn test_engine_stop() {
let mut engine = Engine::with_backend(NullBackend, 0);
engine.stop();
assert!(!engine.is_running());
}
#[test]
fn test_engine_start_all_clients() {
let mut engine = Engine::with_backend(NullBackend, 0);
engine.add_client(Box::new(DummyClient::new())).unwrap();
engine.add_client(Box::new(DummyClient::new())).unwrap();
engine.start_all_clients().unwrap();
assert_eq!(engine.client_count(), 2);
}
#[test]
fn test_engine_update_client_events() {
let mut engine = Engine::with_backend(NullBackend, 0);
let cid = engine.add_client(Box::new(DummyClient::new())).unwrap();
engine.update_client_events(cid).unwrap();
}
#[test]
fn test_engine_timer_creation() {
let timer = EngineTimer::new(1.0, -1.0, false, || {});
assert!(timer.base().is_valid());
assert_eq!(timer.base().fire_delay(), 1.0);
assert_eq!(timer.base().interval(), -1.0);
assert!(!timer.base().autoclose());
}
#[test]
fn test_engine_timer_fire_calls_handler() {
use std::sync::{
atomic::{AtomicUsize, Ordering},
Arc,
};
let counter = Arc::new(AtomicUsize::new(0));
let counter_clone = counter.clone();
let mut timer = EngineTimer::new(0.5, -1.0, false, move || {
counter_clone.fetch_add(1, Ordering::SeqCst);
});
assert_eq!(counter.load(Ordering::SeqCst), 0);
timer.fire();
assert_eq!(counter.load(Ordering::SeqCst), 1);
timer.fire();
assert_eq!(counter.load(Ordering::SeqCst), 2);
}
#[test]
fn test_engine_timer_invalidate() {
let mut timer = EngineTimer::new(1.0, -1.0, false, || {});
assert!(timer.base().is_valid());
timer.base_mut().invalidate();
assert!(!timer.base().is_valid());
}
#[test]
fn test_engine_timer_debug() {
let timer = EngineTimer::new(1.0, -1.0, false, || {});
let debug_str = format!("{:?}", timer);
assert!(debug_str.contains("EngineTimer"));
assert!(debug_str.contains("<fn>"));
}
#[test]
fn test_engine_timer_with_autoclose() {
let timer = EngineTimer::new(2.0, 0.5, true, || {});
assert!(timer.base().autoclose());
assert_eq!(timer.base().interval(), 0.5);
}
}