#![allow(deprecated)]
mod atomic_stack;
mod atomic_waker;
mod entry;
mod handle;
mod now;
pub(crate) mod park;
mod registration;
mod stack;
pub use self::handle::{set_default, Handle};
pub use self::now::{Now, SystemNow};
pub(crate) use self::atomic_waker::AtomicWaker;
pub(crate) use self::handle::HandlePriv;
pub(crate) use self::registration::Registration;
use self::atomic_stack::AtomicStack;
use self::entry::Entry;
use self::park::{Park, ParkThread, Unpark};
use self::stack::Stack;
use crate::timer::atomic::AtomicU64;
use crate::timer::wheel;
use crate::timer::Error;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::usize;
use std::{cmp, fmt};
#[derive(Debug)]
pub struct Timer<T, N = SystemNow> {
inner: Arc<Inner>,
wheel: wheel::Wheel<Stack>,
park: T,
now: N,
}
#[derive(Debug)]
pub struct Turn(());
pub(crate) struct Inner {
start: Instant,
elapsed: AtomicU64,
num: AtomicUsize,
process: AtomicStack,
unpark: Box<dyn Unpark>,
}
const MAX_TIMEOUTS: usize = usize::MAX >> 1;
impl<T> Timer<T>
where
T: Park,
{
pub fn new(park: T) -> Self {
Timer::new_with_now(park, SystemNow::new())
}
}
impl<T, N> Timer<T, N> {
pub fn get_park(&self) -> &T {
&self.park
}
pub fn get_park_mut(&mut self) -> &mut T {
&mut self.park
}
}
impl<T, N> Timer<T, N>
where
T: Park,
N: Now,
{
pub fn new_with_now(park: T, mut now: N) -> Self {
let unpark = Box::new(park.unpark());
Timer {
inner: Arc::new(Inner::new(now.now(), unpark)),
wheel: wheel::Wheel::new(),
park,
now,
}
}
pub fn handle(&self) -> Handle {
Handle::new(Arc::downgrade(&self.inner))
}
pub fn turn(&mut self, max_wait: Option<Duration>) -> Result<Turn, T::Error> {
match max_wait {
Some(timeout) => self.park_timeout(timeout)?,
None => self.park()?,
}
Ok(Turn(()))
}
fn expiration_instant(&self, when: u64) -> Instant {
self.inner.start + Duration::from_millis(when)
}
fn process(&mut self) {
let now = crate::timer::ms(self.now.now() - self.inner.start, crate::timer::Round::Down);
let mut poll = wheel::Poll::new(now);
while let Some(entry) = self.wheel.poll(&mut poll, &mut ()) {
let when = entry.when_internal().expect("invalid internal entry state");
entry.fire(when);
entry.set_when_internal(None);
}
self.inner.elapsed.store(self.wheel.elapsed(), SeqCst);
}
fn process_queue(&mut self) {
for entry in self.inner.process.take() {
match (entry.when_internal(), entry.load_state()) {
(None, None) => {
}
(Some(_), None) => {
self.clear_entry(&entry);
}
(None, Some(when)) => {
self.add_entry(entry, when);
}
(Some(_), Some(next)) => {
self.clear_entry(&entry);
self.add_entry(entry, next);
}
}
}
}
fn clear_entry(&mut self, entry: &Arc<Entry>) {
self.wheel.remove(entry, &mut ());
entry.set_when_internal(None);
}
fn add_entry(&mut self, entry: Arc<Entry>, when: u64) {
use crate::timer::wheel::InsertError;
entry.set_when_internal(Some(when));
match self.wheel.insert(when, entry, &mut ()) {
Ok(_) => {}
Err((entry, InsertError::Elapsed)) => {
entry.set_when_internal(None);
entry.fire(when);
}
Err((entry, InsertError::Invalid)) => {
entry.set_when_internal(None);
entry.error();
}
}
}
}
impl Default for Timer<ParkThread, SystemNow> {
fn default() -> Self {
Timer::new(ParkThread::new())
}
}
impl<T, N> Park for Timer<T, N>
where
T: Park,
N: Now,
{
type Unpark = T::Unpark;
type Error = T::Error;
fn unpark(&self) -> Self::Unpark {
self.park.unpark()
}
fn park(&mut self) -> Result<(), Self::Error> {
self.process_queue();
match self.wheel.poll_at() {
Some(when) => {
let now = self.now.now();
let deadline = self.expiration_instant(when);
if deadline > now {
self.park.park_timeout(deadline - now)?;
} else {
self.park.park_timeout(Duration::from_secs(0))?;
}
}
None => {
self.park.park()?;
}
}
self.process();
Ok(())
}
fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> {
self.process_queue();
match self.wheel.poll_at() {
Some(when) => {
let now = self.now.now();
let deadline = self.expiration_instant(when);
if deadline > now {
self.park.park_timeout(cmp::min(deadline - now, duration))?;
} else {
self.park.park_timeout(Duration::from_secs(0))?;
}
}
None => {
self.park.park_timeout(duration)?;
}
}
self.process();
Ok(())
}
}
impl<T, N> Drop for Timer<T, N> {
fn drop(&mut self) {
use std::u64;
self.inner.process.shutdown();
let mut poll = wheel::Poll::new(u64::MAX);
while let Some(entry) = self.wheel.poll(&mut poll, &mut ()) {
entry.error();
}
}
}
impl Inner {
fn new(start: Instant, unpark: Box<dyn Unpark>) -> Inner {
Inner {
num: AtomicUsize::new(0),
elapsed: AtomicU64::new(0),
process: AtomicStack::new(),
start,
unpark,
}
}
fn elapsed(&self) -> u64 {
self.elapsed.load(SeqCst)
}
fn increment(&self) -> Result<(), Error> {
let mut curr = self.num.load(SeqCst);
loop {
if curr == MAX_TIMEOUTS {
return Err(Error::at_capacity());
}
let actual = self.num.compare_and_swap(curr, curr + 1, SeqCst);
if curr == actual {
return Ok(());
}
curr = actual;
}
}
fn decrement(&self) {
let prev = self.num.fetch_sub(1, SeqCst);
debug_assert!(prev <= MAX_TIMEOUTS);
}
fn queue(&self, entry: &Arc<Entry>) -> Result<(), Error> {
if self.process.push(entry)? {
self.unpark.unpark();
}
Ok(())
}
fn normalize_deadline(&self, deadline: Instant) -> u64 {
if deadline < self.start {
return 0;
}
crate::timer::ms(deadline - self.start, crate::timer::Round::Up)
}
}
impl fmt::Debug for Inner {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Inner").finish()
}
}