use super::{
event_list_new, timer_rbtree_new, BoxedPollImpl, Event, Poll, PollEvent, PollImpl, Scheduler,
Timer, Waker, POLLIN, POLLOUT,
};
use crate::{time, CacheLineAligned, Result};
use core::marker::PhantomData;
use core::ptr::{self, NonNull};
use core::time::Duration;
use hicollections::{List, RbTree};
use hioff::{container_of, container_of_mut};
use hipool::{Allocator, Boxed};
struct StopEvent {
waker: Waker,
event: Event,
stopped: bool,
}
struct Sched<'a, A: Allocator> {
mark: PhantomData<&'a A>,
}
#[repr(C)]
pub struct SchedImpl<'a, A: Allocator> {
sched: Sched<'a, A>,
poll: BoxedPollImpl<'a, A>,
events: List<Event>,
hi_pos: Option<NonNull<Event>>,
timers: RbTree<Timer>,
private_data: *const (),
now: Duration,
stop: StopEvent,
}
impl<'a, A: Allocator + 'a> SchedImpl<'a, A> {
pub fn new_in(pool: &'a A) -> Result<Boxed<'a, Self, A>> {
let waker = Waker::new()?;
let mut boxed = Boxed::new_in(
pool,
CacheLineAligned::new(Self {
sched: Sched { mark: PhantomData },
stop: StopEvent {
event: Event::new(Self::stop_event_handle),
waker,
stopped: false,
},
poll: PollImpl::new_in(pool)?,
events: event_list_new(),
hi_pos: None,
timers: timer_rbtree_new(),
now: time::now(),
private_data: ptr::null(),
}),
)?;
boxed.active_stop_event();
Ok(unsafe { boxed.cast_unchecked::<Self>() })
}
}
impl<'a, A: Allocator + 'a> SchedImpl<'a, A> {
fn active_stop_event(&mut self) {
let _ = self.poll.add_event(
self.stop.waker.fd.fd(),
POLLIN,
ptr::addr_of!(self.stop.event) as u64,
);
}
fn stop_event_handle(event: &Event, _events: u32, _flags: u16, _sched: &mut dyn Scheduler) {
let this = unsafe { container_of_mut!(event, Self, stop.event) };
if this.stop.waker.awaken() {
this.stop.stopped = true;
}
}
fn wait_timeout(&self) -> i32 {
if !self.events.empty() {
0
} else if let Some(first) = self.timers.first() {
let timeout = first.timeout.get();
if timeout <= self.now {
0
} else {
(timeout - self.now).as_millis() as i32
}
} else {
-1
}
}
fn dispatch_events(&mut self) {
let mut list = event_list_new();
self.events.move_head(&mut list);
self.hi_pos = None;
while let Some(first) = list.first() {
unsafe { list.del(first) };
first.handle(0, 0, self);
}
}
fn dispatch_timers(timers: &mut RbTree<Timer>, sched: &mut dyn Scheduler, now: Duration) {
while let Some(first) = timers.first() {
if first.timeout.get() <= now {
timers.remove(first);
first.handle(sched)
} else {
break;
}
}
}
}
impl<'a, A: Allocator + 'a> SchedImpl<'a, A> {
pub fn run(&mut self) {
while !self.stop.stopped {
self.now = time::now();
let timeout = self.wait_timeout();
let iter = self.poll.wait(timeout);
if timeout > 0 {
self.now = time::now();
}
Self::dispatch_timers(&mut self.timers, &mut self.sched, self.now);
for PollEvent(events, data) in iter {
let mut events = events & (POLLIN | POLLOUT);
if events == 0 {
events = POLLIN | POLLOUT;
}
let event = unsafe { &mut *((data & !0xFFFF_0000_0000_0000) as *mut Event) };
let flags = (data >> 48) as u16;
event.handle(events, flags, &mut self.sched);
}
self.dispatch_events();
}
}
}
impl<'a, A: Allocator> Scheduler for SchedImpl<'a, A> {
fn now(&self) -> Duration {
self.now
}
fn stop(&self) {
self.stop.waker.wake();
}
fn private_data(&self) -> *const () {
self.private_data
}
fn set_private_data(&mut self, data: *const ()) {
self.private_data = data;
}
unsafe fn add_event(&mut self, e: &Event, priority: i32) {
self.del_event(e);
if priority == 0 {
self.events.add_tail(e);
} else if let Some(hi_pos) = self.hi_pos {
self.events.add_after(e, hi_pos.as_ref());
self.hi_pos = Some(e.into());
} else {
self.events.add_head(e);
self.hi_pos = Some(e.into());
}
}
unsafe fn del_event(&mut self, e: &Event) {
if self.hi_pos == Some(e.into()) {
self.hi_pos = self.events.rev_iter_from(e).nth(1).map(|e| e.into());
}
self.events.del(e);
}
unsafe fn add_fd_event(&mut self, e: &Event, events: u32, fd: i32, flags: u16) -> Result<()> {
self.poll.add_event(fd, events, e as *const Event as u64 | ((flags as u64) << 48))
}
unsafe fn mod_fd_event(&mut self, e: &Event, events: u32, fd: i32, flags: u16) -> Result<()> {
self.poll.mod_event(fd, events, e as *const Event as u64 | ((flags as u64) << 48))
}
unsafe fn del_fd_event(&mut self, fd: i32) -> Result<()> {
self.poll.del_event(fd)
}
unsafe fn set_timer(&mut self, t: &Timer, micros: u32) {
self.timers.remove(t);
t.timeout
.set(self.now + Duration::from_micros(micros as u64));
self.timers.insert(t, false);
}
unsafe fn del_timer(&mut self, t: &Timer) {
self.timers.remove(t);
}
}
impl<'a, A: Allocator> Scheduler for Sched<'a, A> {
fn now(&self) -> Duration {
let this = unsafe { container_of!(self, SchedImpl<'a, A>, sched) };
this.now()
}
fn stop(&self) {
let this = unsafe { container_of!(self, SchedImpl<'a, A>, sched) };
this.stop()
}
fn private_data(&self) -> *const () {
let this = unsafe { container_of!(self, SchedImpl<'a, A>, sched) };
this.private_data()
}
fn set_private_data(&mut self, private: *const ()) {
let this = unsafe { container_of_mut!(self, SchedImpl<'a, A>, sched) };
this.set_private_data(private)
}
unsafe fn add_event(&mut self, e: &Event, priority: i32) {
let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
this.add_event(e, priority)
}
unsafe fn del_event(&mut self, e: &Event) {
let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
this.del_event(e)
}
unsafe fn add_fd_event(&mut self, e: &Event, events: u32, fd: i32, flags: u16) -> Result<()> {
let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
this.add_fd_event(e, events, fd, flags)
}
unsafe fn mod_fd_event(&mut self, e: &Event, events: u32, fd: i32, flags: u16) -> Result<()> {
let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
this.mod_fd_event(e, events, fd, flags)
}
unsafe fn del_fd_event(&mut self, fd: i32) -> Result<()> {
let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
this.del_fd_event(fd)
}
unsafe fn set_timer(&mut self, t: &Timer, msecs: u32) {
let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
this.set_timer(t, msecs)
}
unsafe fn del_timer(&mut self, t: &Timer) {
let this = container_of_mut!(self, SchedImpl<'a, A>, sched);
this.del_timer(t)
}
}