#![forbid(missing_docs)]
use parking_lot_core as plc;
use parking_lot_core::ParkResult;
use std::convert::Infallible;
use std::sync::atomic::{AtomicU8, Ordering};
use std::time::{Duration, Instant};
#[cfg(test)]
mod tests;
const AVAILABLE_BIT: u8 = 0x01;
const WAITING_BIT: u8 = 0x02;
#[doc(hidden)]
pub struct RawEvent(AtomicU8);
#[derive(Clone, Debug, PartialEq)]
#[repr(u8)]
pub enum EventState {
Set,
Unset,
}
impl RawEvent {
const fn new(state: u8) -> RawEvent {
RawEvent(AtomicU8::new(state))
}
#[inline]
fn try_unlock_one(&self) -> bool {
self.0.compare_exchange_weak(AVAILABLE_BIT, 0, Ordering::Acquire, Ordering::Relaxed)
.is_ok()
}
#[cfg(any(test, miri))]
fn test_try_unlock_one(&self) -> bool {
self.0.compare_exchange(AVAILABLE_BIT, 0, Ordering::Acquire, Ordering::Relaxed)
.is_ok()
}
#[inline]
fn try_unlock_all(&self) -> bool {
(self.0.load(Ordering::Acquire) & AVAILABLE_BIT) != 0
}
unsafe fn suspend_one(&self, timeout: Option<Duration>) -> bool {
let timeout = timeout.map(|duration| Instant::now() + duration);
let mut state = self.0.load(Ordering::Relaxed);
loop {
if (state & AVAILABLE_BIT) != 0 {
match self.0.compare_exchange_weak(
state, state & !AVAILABLE_BIT, Ordering::Acquire, Ordering::Relaxed,
) {
Ok(_) => {
return true;
}
Err(s) => {
state = s;
continue;
}
}
} else if (state & WAITING_BIT) == 0 {
match self.0.compare_exchange_weak(
state, state | WAITING_BIT, Ordering::Relaxed, Ordering::Relaxed,
) {
Ok(_) => {
}
Err(s) => {
state = s;
continue;
}
}
} else {
}
let before_suspend = || -> bool {
self.0.load(Ordering::Relaxed) == WAITING_BIT
};
let on_timeout = |_, last_thread| {
if last_thread {
self.0.fetch_and(!WAITING_BIT, Ordering::Relaxed);
}
};
match plc::park(
self as *const RawEvent as usize, before_suspend,
|| {}, on_timeout,
plc::DEFAULT_PARK_TOKEN,
timeout,
) {
ParkResult::Invalid => state = self.0.load(Ordering::Relaxed),
ParkResult::TimedOut => return false,
ParkResult::Unparked(_) => return true,
}
}
}
unsafe fn suspend_all(&self, timeout: Option<Duration>) -> bool {
let timeout = timeout.map(|duration| Instant::now() + duration);
let mut state = self.0.load(Ordering::Relaxed);
loop {
if (state & AVAILABLE_BIT) != 0 {
return true;
} else if (state & WAITING_BIT) == 0 {
match self.0.compare_exchange_weak(
state, state | WAITING_BIT, Ordering::Relaxed, Ordering::Relaxed,
) {
Ok(_) => {
}
Err(s) => {
state = s;
continue;
}
}
} else {
}
let before_suspend = || -> bool {
self.0.load(Ordering::Relaxed) == WAITING_BIT
};
let mut timeout_result = false;
let on_timeout = |_, last_thread| {
if last_thread {
if (self.0.swap(0, Ordering::Relaxed) & AVAILABLE_BIT) != 0 {
timeout_result = true;
}
}
};
match plc::park(
self as *const RawEvent as usize, before_suspend,
|| {}, on_timeout,
plc::DEFAULT_PARK_TOKEN,
timeout,
) {
ParkResult::Invalid => state = self.0.load(Ordering::Relaxed),
ParkResult::TimedOut => return timeout_result,
ParkResult::Unparked(_) => return true,
}
}
}
fn set_one(&self) {
let mut state = match self.0.compare_exchange(
0, AVAILABLE_BIT, Ordering::Release, Ordering::Relaxed,
) {
Ok(_) => return,
Err(s) => s,
};
loop {
match state {
0b00 => {
match self.0.compare_exchange_weak(
0, AVAILABLE_BIT, Ordering::Release, Ordering::Relaxed,
) {
Ok(_) => return,
Err(s) => {
state = s;
continue;
}
}
}
0b01 => {
match self.0.compare_exchange_weak(
state, state, Ordering::Release, Ordering::Relaxed,
) {
Ok(_) => return,
Err(s) => {
state = s;
continue;
}
}
}
0b10 => {
break;
}
0b11 => {
#[cfg(any(test, miri))]
assert!(false, "AVAILABLE and WAITING bits set!");
break;
}
_ => {
unsafe { core::hint::unreachable_unchecked() }
}
}
}
unsafe {
plc::unpark_one(self as *const RawEvent as usize, |unpark_result| {
if unpark_result.unparked_threads == 0 {
self.0.store(AVAILABLE_BIT, Ordering::Release);
} else if !unpark_result.have_more_threads {
self.0.store(0, Ordering::Release);
} else {
self.0.store(WAITING_BIT, Ordering::Release);
}
plc::DEFAULT_UNPARK_TOKEN
})
};
}
fn set_all(&self) {
let prev_state = self.0.swap(AVAILABLE_BIT, Ordering::Release);
if (prev_state & WAITING_BIT) == 0 {
return;
}
let _unparked = unsafe {
plc::unpark_all(self as *const RawEvent as usize, plc::DEFAULT_UNPARK_TOKEN)
};
}
fn unlock_one(&self) {
if !self.try_unlock_one() {
unsafe {
self.suspend_one(None);
}
}
}
fn unlock_all(&self) {
if !self.try_unlock_all() {
unsafe {
self.suspend_all(None);
}
}
}
fn reset(&self) {
self.0.fetch_and(!AVAILABLE_BIT, Ordering::Relaxed);
}
fn wait_one_for(&self, limit: Duration) -> bool {
if self.try_unlock_one() {
return true;
}
unsafe {self.suspend_one(Some(limit)) }
}
fn wait_all_for(&self, limit: Duration) -> bool {
if self.try_unlock_all() {
return true;
}
unsafe { self.suspend_all(Some(limit)) }
}
}
#[doc(hidden)]
pub type State = EventState;
#[derive(Debug, Copy, Clone)]
pub struct TimeoutError;
impl std::fmt::Display for TimeoutError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("The wait call timed out")
}
}
impl std::error::Error for TimeoutError {}
impl AwaitableError for TimeoutError {
type UnboundedError = std::convert::Infallible;
}
impl std::convert::From<Infallible> for TimeoutError {
fn from(_: Infallible) -> Self {
unsafe { core::hint::unreachable_unchecked() }
}
}
mod sealed {
use crate::{Awaitable, AwaitableError};
pub trait InfallibleUnboundedWait {}
impl<E> InfallibleUnboundedWait for E where
E: AwaitableError<UnboundedError = std::convert::Infallible>
{
}
pub trait VoidAwaitable {}
impl<T, E> VoidAwaitable for T
where
T: for<'a> Awaitable<'a, T = (), Error = E>,
E: std::error::Error,
{
}
}
pub trait AwaitableError: std::error::Error {
type UnboundedError: std::error::Error;
}
pub trait Awaitable<'a> {
type T;
type Error: AwaitableError;
fn try_wait(&'a self) -> Result<Self::T, <Self::Error as AwaitableError>::UnboundedError>;
fn try_wait_for(&'a self, limit: Duration) -> Result<Self::T, Self::Error>;
fn try_wait0(&'a self) -> Result<Self::T, Self::Error> {
self.try_wait_for(Duration::ZERO)
}
fn wait(&'a self) -> Self::T
where
Self::Error: sealed::InfallibleUnboundedWait,
{
self.try_wait()
.expect("try_wait() is not allowed to return TimeoutError!")
}
fn wait_for(&'a self, limit: Duration) -> bool
where
Self: sealed::VoidAwaitable,
Self::Error: sealed::InfallibleUnboundedWait,
{
match self.try_wait_for(limit) {
Ok(_) => true,
Err(_) => false,
}
}
fn wait0(&'a self) -> bool
where
Self: sealed::VoidAwaitable,
Self::Error: sealed::InfallibleUnboundedWait,
{
match self.try_wait0() {
Ok(_) => true,
Err(_) => false,
}
}
}
pub struct AutoResetEvent {
event: RawEvent,
}
impl AutoResetEvent {
pub const fn new(state: EventState) -> AutoResetEvent {
Self {
event: RawEvent::new(match state {
EventState::Set => AVAILABLE_BIT,
EventState::Unset => 0,
}),
}
}
pub fn set(&self) {
self.event.set_one()
}
pub fn reset(&self) {
self.event.reset()
}
}
impl Awaitable<'_> for AutoResetEvent {
type T = ();
type Error = TimeoutError;
fn try_wait(&self) -> Result<Self::T, Infallible> {
Ok(self.event.unlock_one())
}
fn try_wait_for(&self, limit: Duration) -> Result<(), TimeoutError> {
if self.event.wait_one_for(limit) {
Ok(())
} else {
Err(TimeoutError)
}
}
fn try_wait0(&self) -> Result<(), TimeoutError> {
#[cfg(any(test, miri))]
return match self.event.test_try_unlock_one() {
true => Ok(()),
false => Err(TimeoutError),
};
#[cfg(not(any(test, miri)))]
return match self.event.try_unlock_one() {
true => Ok(()),
false => Err(TimeoutError),
};
}
}
pub struct ManualResetEvent {
event: RawEvent,
}
impl ManualResetEvent {
pub const fn new(state: EventState) -> ManualResetEvent {
Self {
event: RawEvent::new(match state {
EventState::Set => AVAILABLE_BIT,
EventState::Unset => 0,
}),
}
}
pub fn set(&self) {
self.event.set_all()
}
pub fn reset(&self) {
self.event.reset()
}
}
impl Awaitable<'_> for ManualResetEvent {
type T = ();
type Error = TimeoutError;
fn try_wait(&self) -> Result<(), Infallible> {
Ok(self.event.unlock_all())
}
fn try_wait_for(&self, limit: Duration) -> Result<(), TimeoutError> {
match self.event.wait_all_for(limit) {
true => Ok(()),
false => Err(TimeoutError),
}
}
fn try_wait0(&self) -> Result<(), TimeoutError> {
match self.event.try_unlock_all() {
true => Ok(()),
false => Err(TimeoutError),
}
}
}