throttled_waker.rs

use std::marker::PhantomData;
use std::sync::atomic::{self, AtomicBool, AtomicU8, Ordering};
use std::sync::Mutex;

struct Parker;

impl Parker {
    fn register(&mut self) {}
    fn unregister(&mut self) {}
    fn notify_one(&mut self) {}
}

const NOTIFIED: u8 = 0b001;
const TRANSFER: u8 = 0b010;
const IDLE: u8 = 0b100;

struct Channel<T> {
    parker: Mutex<Parker>,
    state: AtomicU8,
    _t: PhantomData<T>,
}

impl<T> Channel<T> {
    fn recv(&self) -> T {
        let mut waking = false;
        let mut registered = false;

        loop {
            if let Some(x) = self.try_recv() {
                if waking {
                    let state = self.state.fetch_and(!NOTIFIED, Ordering::SeqCst);

                    if state & IDLE == 0 {
                        self.notify();
                    }
                }

                if registered {
                    self.parker.lock().unwrap().unregister();
                }

                return x;
            }

            if !registered {
                self.parker.lock().unwrap().register();
                registered = true;
            }

            if waking {
                self.state.fetch_and(!(IDLE | NOTIFIED), Ordering::SeqCst);
            } else {
                self.state.fetch_and(!IDLE, Ordering::SeqCst);
            }

            if !self.is_empty() {
                std::thread::park();
            }

            waking = self.state.fetch_and(!TRANSFER, Ordering::Release) & TRANSFER != 0;
        }
    }

    fn send(&self, x: T) {
        self.send_val(x);

        let mut state = self.state.load(Ordering::SeqCst);

        if state & (IDLE | NOTIFIED) != 0 {
            return;
        }

        self.notify();
    }

    fn notify(&self) {
        if self.state.fetch_or(NOTIFIED, Ordering::SeqCst) & (NOTIFIED | IDLE) == 0 {
            self.state.fetch_or(TRANSFER, Ordering::Release);
            self.parker.lock().unwrap().notify_one();
            return;
        }
    }

    fn send_val(&self, _val: T) {
        unimplemented!()
    }

    fn try_recv(&self) -> Option<T> {
        unimplemented!()
    }

    fn is_empty(&self) -> bool {
        unimplemented!()
    }
}