use core::{ptr, task, hint, mem};
use core::cell::UnsafeCell;
use core::sync::atomic::{AtomicU8, Ordering};
mod noop {
use core::{ptr, task};
const VTABLE: task::RawWakerVTable = task::RawWakerVTable::new(clone, action, action, action);
const WAKER: task::RawWaker = task::RawWaker::new(ptr::null(), &VTABLE);
fn clone(_: *const()) -> task::RawWaker {
WAKER
}
fn action(_: *const ()) {
}
#[inline(always)]
pub fn waker() -> task::Waker {
unsafe {
task::Waker::from_raw(WAKER)
}
}
}
pub(crate) mod thread {
use std::thread::Thread;
use core::{task, mem};
const VTABLE: task::RawWakerVTable = task::RawWakerVTable::new(clone, wake, wake_by_ref, on_drop);
unsafe fn on_drop(thread: *const ()) {
let thread = Box::from_raw(thread as *mut Thread);
drop(thread);
}
unsafe fn clone(thread: *const()) -> task::RawWaker {
let thread = Box::from_raw(thread as *mut Thread);
let new_ptr = thread.clone();
mem::forget(thread);
task::RawWaker::new(Box::into_raw(new_ptr) as _, &VTABLE)
}
unsafe fn wake(thread: *const ()) {
let thread = Box::from_raw(thread as *mut () as *mut Thread);
thread.unpark();
}
unsafe fn wake_by_ref(thread: *const ()) {
let thread = &*(thread as *const Thread);
thread.unpark();
}
#[inline(always)]
pub fn waker(thread: Thread) -> task::Waker {
let thread = Box::new(thread);
unsafe {
task::Waker::from_raw(task::RawWaker::new(Box::into_raw(thread) as _, &VTABLE))
}
}
}
const WAITING: u8 = 0;
const REGISTERING: u8 = 0b01;
const WAKING: u8 = 0b10;
#[doc(hidden)]
pub struct AtomicWaker {
state: AtomicU8,
waker: UnsafeCell<task::Waker>,
}
struct StateRestore<F: Fn()>(F);
impl<F: Fn()> Drop for StateRestore<F> {
fn drop(&mut self) {
(self.0)()
}
}
macro_rules! impl_register {
($this:ident($waker:ident) { $($impl:tt)+ }) => {
match $this.state.compare_exchange(WAITING, REGISTERING, Ordering::Acquire, Ordering::Acquire).unwrap_or_else(|err| err) {
WAITING => {
let state_guard = StateRestore(|| {
$this.state.store(WAITING, Ordering::Release);
});
unsafe {
$(
$impl
)+
match $this.state.compare_exchange(REGISTERING, WAITING, Ordering::AcqRel, Ordering::Acquire) {
Ok(_) => {
mem::forget(state_guard);
}
Err(actual) => {
debug_assert_eq!(actual, REGISTERING | WAKING);
let mut waker = noop::waker();
ptr::swap($this.waker.get(), &mut waker);
drop(state_guard);
waker.wake();
}
}
}
}
WAKING => {
$waker.wake_by_ref();
hint::spin_loop();
}
state => {
debug_assert!(
state == REGISTERING ||
state == REGISTERING | WAKING
);
}
}
};
}
impl AtomicWaker {
pub fn new() -> Self {
Self {
state: AtomicU8::new(WAITING),
waker: UnsafeCell::new(noop::waker()),
}
}
pub fn register_ref(&self, waker: &task::Waker) {
impl_register!(self(waker) {
if !(*self.waker.get()).will_wake(waker) {
let mut waker = waker.clone();
ptr::swap(self.waker.get(), &mut waker);
}
});
}
pub fn wake(&self) {
match self.state.fetch_or(WAKING, Ordering::AcqRel) {
WAITING => {
let mut waker = noop::waker();
unsafe {
ptr::swap(self.waker.get(), &mut waker);
}
self.state.fetch_and(!WAKING, Ordering::Release);
waker.wake();
}
state => {
debug_assert!(
state == REGISTERING ||
state == REGISTERING | WAKING ||
state == WAKING
);
}
}
}
}
unsafe impl Send for AtomicWaker {}
unsafe impl Sync for AtomicWaker {}