use std::cell::UnsafeCell;
use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex as StdMutex};
use std::task::{Context, Poll, Waker};

use crate::chaos::{self, ChaosOperation};

struct WaitList {
    waiters: VecDeque<(usize, Waker)>,
    next_id: usize,
}

impl WaitList {
    fn new() -> Self {
        Self {
            waiters: VecDeque::new(),
            next_id: 0,
        }
    }

    fn push(&mut self, waker: Waker) -> usize {
        let id = self.next_id;
        self.next_id = self.next_id.wrapping_add(1);
        self.waiters.push_back((id, waker));
        id
    }

    fn remove(&mut self, id: usize) {
        if let Some(pos) = self.waiters.iter().position(|(i, _)| *i == id) {
            self.waiters.remove(pos);
        }
    }

    fn pop(&mut self) -> Option<Waker> {
        self.waiters.pop_front().map(|(_, w)| w)
    }

    fn push_back(&mut self, waker: Waker) -> usize {
        self.push(waker)
    }

    fn pop_front(&mut self) -> Option<Waker> {
        self.pop()
    }

    fn iter_mut(&mut self) -> impl Iterator<Item = &mut (usize, Waker)> {
        self.waiters.iter_mut()
    }
}

pub struct Mutex<T: ?Sized> {
    state: StdMutex<MutexState>,
    data: UnsafeCell<T>,
}

struct MutexState {
    locked: bool,
    waiters: WaitList,
}

unsafe impl<T: Send + ?Sized> Send for Mutex<T> {}
unsafe impl<T: Send + ?Sized> Sync for Mutex<T> {}

impl<T> Mutex<T> {
    pub fn new(data: T) -> Self {
        Self {
            state: StdMutex::new(MutexState {
                locked: false,
                waiters: WaitList::new()
            }),
            data: UnsafeCell::new(data),
        }
    }

    pub fn into_inner(self) -> T {
        self.data.into_inner()
    }
}

impl<T: ?Sized> Mutex<T> {
    pub async fn lock(&self) -> MutexGuard<'_, T> {
        if chaos::should_fail(ChaosOperation::MutexDeadlock) {
            std::future::pending::<()>().await;
        }

        struct LockFuture<'a, T: ?Sized> { mutex: &'a Mutex<T> }
        impl<T: ?Sized> Future for LockFuture<'_, T> {
            type Output = ();
            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
                if chaos::should_fail(ChaosOperation::MutexLockTimeout) {
                    return Poll::Pending;
                }

                let mut state = self.mutex.state.lock().unwrap();
                if !state.locked {
                    state.locked = true;
                    Poll::Ready(())
                }
                else {
                    state.waiters.push_back(cx.waker().clone());
                    Poll::Pending
                }
            }
        }
        LockFuture { mutex: self }.await;
        MutexGuard { lock: self }
    }

    pub async fn lock_owned(self: Arc<Self>) -> OwnedMutexGuard<T> {
        if chaos::should_fail(ChaosOperation::MutexDeadlock) {
            std::future::pending::<()>().await;
        }

        struct LockOwnedFuture<T: ?Sized> { mutex: Arc<Mutex<T>> }
        impl<T: ?Sized> Future for LockOwnedFuture<T> {
            type Output = ();
            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
                if chaos::should_fail(ChaosOperation::MutexLockTimeout) {
                    return Poll::Pending;
                }

                let mut state = self.mutex.state.lock().unwrap();
                if !state.locked {
                    state.locked = true;
                    Poll::Ready(())
                }
                else {
                    state.waiters.push_back(cx.waker().clone());
                    Poll::Pending
                }
            }
        }
        LockOwnedFuture { mutex: self.clone() }.await;
        OwnedMutexGuard { lock: self }
    }

    pub fn try_lock(&self) -> Result<MutexGuard<'_, T>, ()> {
        if chaos::should_fail(ChaosOperation::MutexTryLock) {
            return Err(());
        }

        let mut state = self.state.lock().unwrap();
        if !state.locked {
            state.locked = true;
            Ok(MutexGuard { lock: self })
        } else {
            Err(())
        }
    }

    pub fn blocking_lock(&self) -> MutexGuard<'_, T> {
        loop {
            if let Ok(guard) = self.try_lock() {
                return guard;
            }
            std::thread::yield_now();
        }
    }

    pub fn blocking_lock_owned(self: Arc<Self>) -> OwnedMutexGuard<T> {
        loop {
            let locked = {
                let mut state = self.state.lock().unwrap();
                if !state.locked {
                    state.locked = true;
                    true
                } else {
                    false
                }
            };

            if locked {
                return OwnedMutexGuard { lock: self };
            }

            std::thread::yield_now();
        }
    }
}

pub struct LockFuture<'a, T: ?Sized> {
    mutex: &'a Mutex<T>,
    waiter_id: Option<usize>,
}

impl<'a, T: ?Sized> Future for LockFuture<'a, T> {
    type Output = MutexGuard<'a, T>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if chaos::should_fail(ChaosOperation::MutexDeadlock) {
            return Poll::Pending;
        }

        let mut state = self.mutex.state.lock().unwrap();

        if !state.locked {
            state.locked = true;
            
            if let Some(id) = self.waiter_id.take() {
                state.waiters.remove(id);
            }
            
            return Poll::Ready(MutexGuard { lock: self.mutex });
        }

        match self.waiter_id {
            None => {
                let id = state.waiters.push(cx.waker().clone());
                self.waiter_id = Some(id);
            }
            Some(id) => {
                if let Some(entry) = state.waiters.iter_mut().find(|(i, _)| *i == id) {
                    entry.1 = cx.waker().clone();
                }
            }
        }

        if chaos::should_fail(ChaosOperation::MutexLockTimeout) {
            // Wrap future with timeout from time mod
        }

        Poll::Pending
    }
}

impl<'a, T: ?Sized> Drop for LockFuture<'a, T> {
    fn drop(&mut self) {
        if let Some(id) = self.waiter_id {
            let mut state = self.mutex.state.lock().unwrap();
            state.waiters.remove(id);

            // Propagate wake later
        }
    }
}

pub struct MutexGuard<'a, T: ?Sized> { lock: &'a Mutex<T> }

impl<T: ?Sized> Drop for MutexGuard<'_, T> {
    fn drop(&mut self) {
        let mut state = self.lock.state.lock().unwrap();
        state.locked = false;
        if let Some(waker) = state.waiters.pop() {
            waker.wake();
        }
    }
}

impl<T: ?Sized> std::ops::Deref for MutexGuard<'_, T> {
    type Target = T;
    fn deref(&self) -> &Self::Target { unsafe { &*self.lock.data.get() } }
}
impl<T: ?Sized> std::ops::DerefMut for MutexGuard<'_, T> {
    fn deref_mut(&mut self) -> &mut Self::Target { unsafe { &mut *self.lock.data.get() } }
}

pub struct OwnedMutexGuard<T: ?Sized> {
    lock: Arc<Mutex<T>>
}

impl<T: ?Sized> Drop for OwnedMutexGuard<T> {
    fn drop(&mut self) {
        let mut state = self.lock.state.lock().unwrap();
        state.locked = false;
        if let Some(waker) = state.waiters.pop() {
            waker.wake();
        }
    }
}

impl<T: ?Sized> std::ops::Deref for OwnedMutexGuard<T> {
    type Target = T;
    fn deref(&self) -> &Self::Target { unsafe { &*self.lock.data.get() } }
}

impl<T: ?Sized> std::ops::DerefMut for OwnedMutexGuard<T> {
    fn deref_mut(&mut self) -> &mut Self::Target { unsafe { &mut *self.lock.data.get() } }
}

pub struct RwLock<T: ?Sized> {
    state: StdMutex<RwLockState>,
    data: UnsafeCell<T>,
}

struct RwLockState {
    readers: usize,
    writer: bool,
    writer_waiters: WaitList,
    reader_waiters: WaitList
}

unsafe impl<T: Send + Sync + ?Sized> Send for RwLock<T> {}
unsafe impl<T: Send + Sync + ?Sized> Sync for RwLock<T> {}

impl<T> RwLock<T> {
    pub fn new(data: T) -> Self {
        Self {
            state: StdMutex::new(RwLockState { 
                readers: 0, 
                writer: false, 
                writer_waiters: WaitList::new(), 
                reader_waiters: WaitList::new() 
            }),
            data: UnsafeCell::new(data),
        }
    }
}

impl<T: ?Sized> RwLock<T> {
    pub async fn read(&self) -> RwLockReadGuard<'_, T> {
        if chaos::should_fail(ChaosOperation::RwLockDeadlock) {
            std::future::pending::<()>().await;
        }

        struct ReadFuture<'a, T: ?Sized> { lock: &'a RwLock<T> }
        impl<T: ?Sized> Future for ReadFuture<'_, T> {
            type Output = ();
            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
                if chaos::should_fail(ChaosOperation::RwLockRead) {
                    return Poll::Pending;
                }

                let mut state = self.lock.state.lock().unwrap();
                if !state.writer {
                    state.readers += 1;
                    Poll::Ready(())
                } else {
                    state.reader_waiters.push_back(cx.waker().clone());
                    Poll::Pending
                }
            }
        }
        ReadFuture { lock: self }.await;
        RwLockReadGuard { lock: self }
    }

    pub async fn write(&self) -> RwLockWriteGuard<'_, T> {
        if chaos::should_fail(ChaosOperation::RwLockDeadlock) {
            std::future::pending::<()>().await;
        }

        struct WriteFuture<'a, T: ?Sized> { lock: &'a RwLock<T> }
        impl<T: ?Sized> Future for WriteFuture<'_, T> {
            type Output = ();
            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
                if chaos::should_fail(ChaosOperation::RwLockWrite) {
                    return Poll::Pending;
                }

                let mut state = self.lock.state.lock().unwrap();
                if !state.writer && state.readers == 0 {
                    state.writer = true;
                    Poll::Ready(())
                } else {
                    state.writer_waiters.push_back(cx.waker().clone());
                    Poll::Pending
                }
            }
        }
        WriteFuture { lock: self }.await;
        RwLockWriteGuard { lock: self }
    }

    pub fn try_read(&self) -> Result<RwLockReadGuard<'_, T>, ()> {
        if chaos::should_fail(ChaosOperation::RwLockRead) {
            return Err(());
        }

        let mut state = self.state.lock().unwrap();
        if !state.writer {
            state.readers += 1;
            Ok(RwLockReadGuard { lock: self })
        } else {
            Err(())
        }
    }

    pub fn try_write(&self) -> Result<RwLockWriteGuard<'_, T>, ()> {
        if chaos::should_fail(ChaosOperation::RwLockWrite) {
            return Err(());
        }

        let mut state = self.state.lock().unwrap();
        if !state.writer && state.readers == 0 {
            state.writer = true;
            Ok(RwLockWriteGuard { lock: self })
        } else {
            Err(())
        }
    }

    pub fn blocking_read(&self) -> RwLockReadGuard<'_, T> {
        loop {
            if let Ok(guard) = self.try_read() {
                return guard;
            }
            std::thread::yield_now();
        }
    }

    pub fn blocking_write(&self) -> RwLockWriteGuard<'_, T> {
        loop {
            if let Ok(guard) = self.try_write() {
                return guard;
            }
            std::thread::yield_now();
        }
    }
}

pub struct RwLockReadGuard<'a, T: ?Sized> { lock: &'a RwLock<T> }
impl<T: ?Sized> Drop for RwLockReadGuard<'_, T> {
    fn drop(&mut self) {
        let mut state = self.lock.state.lock().unwrap();
        state.readers -= 1;
        if state.readers == 0 {
            if let Some(waker) = state.writer_waiters.pop_front() {
                waker.wake();
            }
        }
    }
}
impl<T: ?Sized> std::ops::Deref for RwLockReadGuard<'_, T> {
    type Target = T;
    fn deref(&self) -> &Self::Target { unsafe { &*self.lock.data.get() } }
}

pub struct RwLockWriteGuard<'a, T: ?Sized> { lock: &'a RwLock<T> }
impl<T: ?Sized> Drop for RwLockWriteGuard<'_, T> {
    fn drop(&mut self) {
        let mut state = self.lock.state.lock().unwrap();
        state.writer = false;
        if let Some(waker) = state.writer_waiters.pop_front() {
            waker.wake();
        }
        else {
            while let Some(waker) = state.reader_waiters.pop_front() {
                waker.wake();
            }
        }
    }
}
impl<T: ?Sized> std::ops::Deref for RwLockWriteGuard<'_, T> {
    type Target = T;
    fn deref(&self) -> &Self::Target { unsafe { &*self.lock.data.get() } }
}
impl<T: ?Sized> std::ops::DerefMut for RwLockWriteGuard<'_, T> {
    fn deref_mut(&mut self) -> &mut Self::Target { unsafe { &mut *self.lock.data.get() } }
}

impl<'a, T: ?Sized> RwLockWriteGuard<'a, T> {
    pub fn downgrade(self) -> RwLockReadGuard<'a, T> {
        let mut state = self.lock.state.lock().unwrap();
        state.writer = false;
        state.readers = 1;

        while let Some(waker) = state.reader_waiters.pop_front() {
            waker.wake();
        }

        let lock = self.lock;
        std::mem::forget(self);

        RwLockReadGuard { lock }
    }
}

pub struct Notify {
    state: StdMutex<NotifyState>
}

struct NotifyState {
    waiters: WaitList,
    notified: bool,
}

impl Notify {
    pub fn new() -> Self {
        Self { 
            state: StdMutex::new(NotifyState { 
                waiters: WaitList::new(), 
                notified: false 
            }) 
        }
    }

    pub fn notify_one(&self) {
        if chaos::should_fail(ChaosOperation::NotifyMissed) {
            return;
        }

        let mut state = self.state.lock().unwrap();
        if let Some(waker) = state.waiters.pop() {
            waker.wake();
        } else {
            state.notified = true;
        }
    }

    pub fn notify_waiters(&self) {
        if chaos::should_fail(ChaosOperation::NotifyMissed) {
            return;
        }

        let mut state = self.state.lock().unwrap();

        while let Some(waker) = state.waiters.pop() {
            waker.wake();
        }

        state.notified = true;
    }

    pub fn notified(&self) -> Notified<'_> {
        Notified {
            notify: self,
            waiter_id: None
        }
    }
}

pub struct Notified<'a> {
    notify: &'a Notify,
    waiter_id: Option<usize>
}

impl Future for Notified<'_> {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut state = self.notify.state.lock().unwrap();

        if state.notified {
            state.notified = false;
            if let Some(id) = self.waiter_id.take() {
                state.waiters.remove(id);
            }
            return Poll::Ready(());
        }

        match self.waiter_id {
            None => {
                let id = state.waiters.push(cx.waker().clone());
                self.waiter_id = Some(id);
            }
            Some(id) => {
                if let Some(entry) = state.waiters.iter_mut().find(|(i, _)| *i == id) {
                    entry.1 = cx.waker().clone();
                }
            }
        }
        
        Poll::Pending
    }
}

impl Drop for Notified<'_> {
    fn drop(&mut self) {
        if let Some(id) = self.waiter_id {
            let mut state = self.notify.state.lock().unwrap();
            state.waiters.remove(id);
        }
    }
}

pub struct Semaphore {
    permits: StdMutex<usize>,
    waiters: StdMutex<SemaphoreWaitList>,
    closed: StdMutex<bool>,
}

struct SemaphoreWaitList {
    waiters: VecDeque<(usize, usize, Waker)>,
    next_id: usize,
}

impl SemaphoreWaitList {
    fn new() -> Self {
        Self { waiters: VecDeque::new(), next_id: 0 }
    }
    
    fn push(&mut self, permits: usize, waker: Waker) -> usize {
        let id = self.next_id;
        self.next_id = self.next_id.wrapping_add(1);
        self.waiters.push_back((id, permits, waker));
        id
    }

    fn remove(&mut self, id: usize) {
        if let Some(pos) = self.waiters.iter().position(|(i, _, _)| *i == id) {
            self.waiters.remove(pos);
        }
    }

    fn push_back(&mut self, item: (usize, Waker)) -> usize {
        self.push(item.0, item.1)
    }
}

impl Semaphore {
    pub fn new(permits: usize) -> Self {
        Self {
            permits: StdMutex::new(permits),
            waiters: StdMutex::new(SemaphoreWaitList::new()),
            closed: StdMutex::new(false),
        }
    }

    pub fn close(&self) {
        *self.closed.lock().unwrap() = true;

        let mut waiters = self.waiters.lock().unwrap();
        while let Some((_, _, waker)) = waiters.waiters.pop_front() {
            waker.wake();
        }
    }

    pub fn is_closed(&self) -> bool {
        *self.closed.lock().unwrap()
    }

    pub async fn acquire(&self) -> Result<SemaphorePermit<'_>, ()> {
        self.acquire_many(1).await
    }

    pub async fn acquire_many(&self, n: u32) -> Result<SemaphorePermit<'_>, ()> {
        if chaos::should_fail(ChaosOperation::SemaphoreAcquire) {
            return Err(());
        }

        if *self.closed.lock().unwrap() {
            return Err(());
        }

        let n = n as usize;
        struct AcquireFuture<'a> { sem: &'a Semaphore, n: usize }
        impl Future for AcquireFuture<'_> {
            type Output = Result<(), ()>;
            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), ()>> {
                if chaos::should_fail(ChaosOperation::SemaphoreAcquireTimeout) {
                    return Poll::Pending;
                }

                if *self.sem.closed.lock().unwrap() {
                    return Poll::Ready(Err(()));
                }

                let mut permits = self.sem.permits.lock().unwrap();
                if *permits >= self.n {
                    *permits -= self.n;
                    Poll::Ready(Ok(()))
                } else {
                    self.sem.waiters.lock().unwrap().push_back((self.n, cx.waker().clone()));
                    Poll::Pending
                }
            }
        }
        AcquireFuture { sem: self, n }.await?;
        Ok(SemaphorePermit { sem: self, permits: n })
    }

    pub fn add_permits(&self, n: usize) {
        let mut permits = self.permits.lock().unwrap();
        *permits += n;

        let mut waiters = self.waiters.lock().unwrap();

        // Have to improve this later
        while let Some((_id, needed, _)) = waiters.waiters.front() {
            if *permits >= *needed {
                let (_, _, waker) = waiters.waiters.pop_front().unwrap();
                waker.wake();
            }
            else {
                break;
            }
        }
    }

    pub fn try_acquire(&self) -> Result<SemaphorePermit<'_>, ()> {
        self.try_acquire_many(1)
    }

    pub fn try_acquire_many(&self, n: u32) -> Result<SemaphorePermit<'_>, ()> {
        if chaos::should_fail(ChaosOperation::SemaphoreAcquire) {
            return Err(());
        }

        if *self.closed.lock().unwrap() {
            return Err(());
        }

        let n = n as usize;
        let mut permits = self.permits.lock().unwrap();
        if *permits >= n {
            *permits -= n;
            Ok(SemaphorePermit { sem: self, permits: n })
        } else {
            Err(())
        }
    }

    pub fn available_permits(&self) -> usize {
        *self.permits.lock().unwrap()
    }
}

pub struct SemaphorePermit<'a> { sem: &'a Semaphore, permits: usize }
impl Drop for SemaphorePermit<'_> {
    fn drop(&mut self) {
        self.sem.add_permits(self.permits);
    }
}

pub struct Barrier {
    state: StdMutex<BarrierState>
}

struct BarrierState {
    n: usize,
    count: usize,
    generation: usize,
    waiters: VecDeque<Waker>
}

#[derive(Debug, Clone)]
pub struct BarrierWaitResult(bool);

impl BarrierWaitResult {
    pub fn is_leader(&self) -> bool { self.0 }
}

impl Barrier {
    pub fn new(n: usize) -> Self {
        Self {
            state: StdMutex::new(BarrierState {
                n,
                count: 0,
                generation: 0,
                waiters: VecDeque::new(),
            }),
        }
    }

    pub async fn wait(&self) -> BarrierWaitResult {
        if chaos::should_fail(ChaosOperation::BarrierWait) {
            std::future::pending::<()>().await;
        }

        struct WaitFuture<'a> {
            barrier: &'a Barrier,
            my_generation: Option<usize>,
        }

        impl Future for WaitFuture<'_> {
            type Output = BarrierWaitResult;

            fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
                let mut state = self.barrier.state.lock().unwrap();

                if self.my_generation.is_none() {
                    self.my_generation = Some(state.generation);
                    state.count += 1;

                    if state.count >= state.n {
                        state.count = 0;
                        state.generation += 1;
                        for waker in state.waiters.drain(..) {
                            waker.wake();
                        }
                        return Poll::Ready(BarrierWaitResult(true));
                    }
                }

                if state.generation != self.my_generation.unwrap() {
                    return Poll::Ready(BarrierWaitResult(false));
                }

                if chaos::should_fail(ChaosOperation::BarrierTimeout) {
                     return Poll::Pending;
                }

                state.waiters.push_back(cx.waker().clone());
                Poll::Pending
            }
        }

        WaitFuture { barrier: self, my_generation: None }.await
    }
}


pub mod mpsc {
    use super::*;

    #[derive(Debug)]
    pub struct SendError<T>(pub T);

    impl<T> std::fmt::Display for SendError<T> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "channel closed")
        }
    }

    impl<T: std::fmt::Debug> std::error::Error for SendError<T> {}

    #[derive(Debug)]
    pub enum TrySendError<T> {
        Full(T),
        Closed(T),
    }

    impl<T> std::fmt::Display for TrySendError<T> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            match self {
                TrySendError::Full(_) => write!(f, "channel full"),
                TrySendError::Closed(_) => write!(f, "channel closed"),
            }
        }
    }

    impl<T: std::fmt::Debug> std::error::Error for TrySendError<T> {}

    #[derive(Debug)]
    pub enum SendTimeoutError<T> {
        Timeout(T),
        Closed(T),
    }

    impl<T> std::fmt::Display for SendTimeoutError<T> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            match self {
                SendTimeoutError::Timeout(_) => write!(f, "timed out waiting on send operation"),
                SendTimeoutError::Closed(_) => write!(f, "channel closed"),
            }
        }
    }

    impl<T: std::fmt::Debug> std::error::Error for SendTimeoutError<T> {}

    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    pub enum TryRecvError {
        Empty,
        Disconnected,
    }

    impl std::fmt::Display for TryRecvError {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            match self {
                TryRecvError::Empty => write!(f, "channel empty"),
                TryRecvError::Disconnected => write!(f, "channel closed"),
            }
        }
    }

    impl std::error::Error for TryRecvError {}

    pub fn unbounded<T>() -> (UnboundedSender<T>, UnboundedReceiver<T>) {
        let shared = Arc::new(StdMutex::new(UnboundedState {
            queue: VecDeque::new(),
            waker: None,
            closed: false,
            rx_count: 1,
        }));
        (UnboundedSender { shared: shared.clone() }, UnboundedReceiver { shared })
    }

    struct UnboundedState<T> { queue: VecDeque<T>, waker: Option<Waker>, closed: bool, rx_count: usize }
    
    pub struct UnboundedSender<T> { shared: Arc<StdMutex<UnboundedState<T>>> }
    impl<T> UnboundedSender<T> {
        pub fn send(&self, value: T) -> Result<(), SendError<T>> {
            if chaos::should_fail(ChaosOperation::MpscSend) {
                return Err(SendError(value));
            }
            let mut s = self.shared.lock().unwrap();
            if s.closed { return Err(SendError(value)); }
            s.queue.push_back(value);
            if let Some(w) = s.waker.take() { w.wake(); }
            Ok(())
        }

        pub fn blocking_send(&self, value: T) -> Result<(), SendError<T>> {
            self.send(value)
        }

        pub fn is_closed(&self) -> bool {
            let s = self.shared.lock().unwrap();
            s.closed
        }

        pub fn same_channel(&self, other: &Self) -> bool {
            Arc::ptr_eq(&self.shared, &other.shared)
        }
    }
    impl<T> Clone for UnboundedSender<T> {
        fn clone(&self) -> Self { Self { shared: self.shared.clone() } }
    }

    pub struct UnboundedReceiver<T> { shared: Arc<StdMutex<UnboundedState<T>>> }
    impl<T> UnboundedReceiver<T> {
        pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
            if chaos::should_fail(ChaosOperation::MpscRecv) {
                return Poll::Ready(None);
            }
            let mut s = self.shared.lock().unwrap();
            if let Some(v) = s.queue.pop_front() { Poll::Ready(Some(v)) }
            else if s.closed { Poll::Ready(None) }
            else { s.waker = Some(cx.waker().clone()); Poll::Pending }
        }

        pub async fn recv(&mut self) -> Option<T> {
            struct RecvFuture<'a, T> { rx: &'a mut UnboundedReceiver<T> }
            impl<T> Future for RecvFuture<'_, T> {
                type Output = Option<T>;
                fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
                    self.rx.poll_recv(cx)
                }
            }
            RecvFuture { rx: self }.await
        }

        pub fn blocking_recv(&mut self) -> Option<T> {
            let mut s = self.shared.lock().unwrap();
            s.queue.pop_front().or_else(|| {
                if s.closed { None } else { s.queue.pop_front() }
            })
        }

        pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
            if chaos::should_fail(ChaosOperation::MpscRecv) {
                return Err(TryRecvError::Disconnected);
            }
            let mut s = self.shared.lock().unwrap();
            if let Some(v) = s.queue.pop_front() {
                Ok(v)
            } else if s.closed {
                Err(TryRecvError::Disconnected)
            } else {
                Err(TryRecvError::Empty)
            }
        }

        pub fn close(&mut self) {
            let mut s = self.shared.lock().unwrap();
            s.closed = true;
        }
    }
    impl<T> Drop for UnboundedReceiver<T> {
        fn drop(&mut self) {
            let mut s = self.shared.lock().unwrap();
            s.rx_count -= 1;
            if s.rx_count == 0 { s.closed = true; }
        }
    }
    
    pub fn channel<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
        assert!(capacity > 0, "capacity must be > 0");
        let state = Arc::new(StdMutex::new(BoundedState {
            queue: VecDeque::with_capacity(capacity),
            capacity,
            reserved: 0,
            tx_waiters: VecDeque::new(),
            rx_waker: None,
            closed: false,
            sender_count: 1,
        }));
        (Sender { state: state.clone() }, Receiver { state })
    }

    struct BoundedState<T> {
        queue: VecDeque<T>,
        capacity: usize,
        reserved: usize,
        tx_waiters: VecDeque<Waker>,
        rx_waker: Option<Waker>,
        closed: bool,
        sender_count: usize,
    }

    pub struct Sender<T> { state: Arc<StdMutex<BoundedState<T>>> }
    
    impl<T> Sender<T> {
        pub async fn send(&self, value: T) -> Result<(), SendError<T>> {
            #[pin_project::pin_project]
            struct SendFuture<'a, T> {
                sender: &'a Sender<T>,
                value: Option<T>
            }

            impl<T> Future for SendFuture<'_, T> {
                type Output = Result<(), SendError<T>>;

                fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
                    let this = self.project();

                    if chaos::should_fail(ChaosOperation::MpscSend) {
                        return Poll::Ready(Err(SendError(this.value.take().unwrap())));
                    }

                    let mut s = this.sender.state.lock().unwrap();

                    if s.closed {
                        return Poll::Ready(Err(SendError(this.value.take().unwrap())));
                    }

                    if s.queue.len() + s.reserved < s.capacity {
                        s.queue.push_back(this.value.take().unwrap());
                        if let Some(w) = s.rx_waker.take() { w.wake(); }
                        Poll::Ready(Ok(()))
                    }
                    else {
                        s.tx_waiters.push_back(cx.waker().clone());
                        Poll::Pending
                    }
                }
            }

            SendFuture { sender: self, value: Some(value) }.await
        }

        pub fn blocking_send(&self, value: T) -> Result<(), SendError<T>> {
            loop {
                let mut s = self.state.lock().unwrap();
                if s.closed {
                    return Err(SendError(value));
                }
                if s.queue.len() < s.capacity {
                    s.queue.push_back(value);
                    if let Some(w) = s.rx_waker.take() { w.wake(); }
                    return Ok(());
                }
                drop(s);
                std::thread::yield_now();
            }
        }

        pub fn try_send(&self, value: T) -> Result<(), TrySendError<T>> {
            if chaos::should_fail(ChaosOperation::MpscTrySend) {
                return Err(TrySendError::Full(value));
            }

            let mut s = self.state.lock().unwrap();
            if s.closed {
                return Err(TrySendError::Closed(value));
            }

            if s.queue.len() < s.capacity {
                s.queue.push_back(value);
                if let Some(w) = s.rx_waker.take() { w.wake(); }
                Ok(())
            }
            else {
                Err(TrySendError::Full(value))
            }
        }

        pub async fn send_timeout(&self, value: T, duration: std::time::Duration) -> Result<(), SendTimeoutError<T>> {
            use crate::time::sleep;

            #[pin_project::pin_project]
            struct SendTimeout<'a, T> {
                sender: &'a Sender<T>,
                #[pin]
                sleep: crate::time::Sleep,
                value: Option<T>,
            }

            impl<T> Future for SendTimeout<'_, T> {
                type Output = Result<(), SendTimeoutError<T>>;

                fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
                    let mut this = self.project();

                    if this.sleep.as_mut().poll(cx).is_ready() {
                        return Poll::Ready(Err(SendTimeoutError::Timeout(this.value.take().unwrap())));
                    }

                    if chaos::should_fail(ChaosOperation::MpscSend) {
                        return Poll::Ready(Err(SendTimeoutError::Closed(this.value.take().unwrap())));
                    }

                    let mut s = this.sender.state.lock().unwrap();

                    if s.closed {
                        return Poll::Ready(Err(SendTimeoutError::Closed(this.value.take().unwrap())));
                    }

                    if s.queue.len() + s.reserved < s.capacity {
                        s.queue.push_back(this.value.take().unwrap());
                        if let Some(w) = s.rx_waker.take() { w.wake(); }
                        Poll::Ready(Ok(()))
                    }
                    else {
                        s.tx_waiters.push_back(cx.waker().clone());
                        Poll::Pending
                    }
                }
            }

            SendTimeout {
                sender: self,
                sleep: sleep(duration),
                value: Some(value),
            }.await
        }

        pub async fn reserve(&self) -> Result<Permit<'_, T>, SendError<()>> {
            if chaos::should_fail(ChaosOperation::MpscPermitReserve) {
                return Err(SendError(()));
            }

            struct ReserveFuture<'a, T> { sender: &'a Sender<T> }

            impl<'a, T> Future for ReserveFuture<'a, T> {
                type Output = Result<Permit<'a, T>, SendError<()>>;

                fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
                    let mut s = self.sender.state.lock().unwrap();
                    if s.closed {
                        return Poll::Ready(Err(SendError(())));
                    }

                    if s.queue.len() + s.reserved < s.capacity {
                        s.reserved += 1;
                        Poll::Ready(Ok(Permit { sender: self.sender }))
                    } else {
                        s.tx_waiters.push_back(cx.waker().clone());
                        Poll::Pending
                    }
                }
            }

            ReserveFuture { sender: self }.await
        }

        pub async fn reserve_owned(self) -> Result<OwnedPermit<T>, SendError<()>> {
            if chaos::should_fail(ChaosOperation::MpscPermitReserve) {
                return Err(SendError(()));
            }

            struct ReserveOwnedFuture<T> { sender: Sender<T> }

            impl<T> Future for ReserveOwnedFuture<T> {
                type Output = Result<OwnedPermit<T>, SendError<()>>;

                fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
                    let mut s = self.sender.state.lock().unwrap();
                    if s.closed {
                        return Poll::Ready(Err(SendError(())));
                    }

                    if s.queue.len() + s.reserved < s.capacity {
                        s.reserved += 1;
                        Poll::Ready(Ok(OwnedPermit { state: self.sender.state.clone() }))
                    } else {
                        s.tx_waiters.push_back(cx.waker().clone());
                        Poll::Pending
                    }
                }
            }

            ReserveOwnedFuture { sender: self }.await
        }

        pub fn is_closed(&self) -> bool {
            let s = self.state.lock().unwrap();
            s.closed
        }

        pub fn same_channel(&self, other: &Self) -> bool {
            Arc::ptr_eq(&self.state, &other.state)
        }

        pub fn capacity(&self) -> usize {
            let s = self.state.lock().unwrap();
            s.capacity
        }
    }

    impl<T> Clone for Sender<T> {
        fn clone(&self) -> Self {
            let mut s = self.state.lock().unwrap();
            s.sender_count += 1;
            Self { state: self.state.clone() }
        }
    }

    impl<T> Drop for Sender<T> {
        fn drop(&mut self) {
            let mut s = self.state.lock().unwrap();
            s.sender_count -= 1;
            if s.sender_count == 0 {
                s.closed = true;
                if let Some(w) = s.rx_waker.take() { w.wake(); }
            }
        }
    }

    pub struct Receiver<T> { state: Arc<StdMutex<BoundedState<T>>> }

    impl<T> Receiver<T> {
        pub async fn recv(&mut self) -> Option<T> {
             struct RecvFuture<'a, T> { rx: &'a mut Receiver<T> }
            impl<T> Future for RecvFuture<'_, T> {
                type Output = Option<T>;
                fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
                    if chaos::should_fail(ChaosOperation::MpscRecv) {
                        return Poll::Ready(None);
                    }

                    let mut s = self.rx.state.lock().unwrap();
                    if let Some(v) = s.queue.pop_front() {
                        if let Some(w) = s.tx_waiters.pop_front() { w.wake(); }
                        Poll::Ready(Some(v))
                    }
                    else if s.closed && s.sender_count == 0 {
                        Poll::Ready(None)
                    }
                    else {
                        s.rx_waker = Some(cx.waker().clone());
                        Poll::Pending
                    }
                }
            }
            RecvFuture { rx: self }.await
        }

        pub fn blocking_recv(&mut self) -> Option<T> {
            loop {
                let mut s = self.state.lock().unwrap();
                if let Some(v) = s.queue.pop_front() {
                    if let Some(w) = s.tx_waiters.pop_front() { w.wake(); }
                    return Some(v);
                }
                if s.closed && s.sender_count == 0 {
                    return None;
                }
                drop(s);
                std::thread::yield_now();
            }
        }

        pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
            if chaos::should_fail(ChaosOperation::MpscRecv) {
                return Err(TryRecvError::Disconnected);
            }
            let mut s = self.state.lock().unwrap();
            if let Some(v) = s.queue.pop_front() {
                if let Some(w) = s.tx_waiters.pop_front() { w.wake(); }
                Ok(v)
            } else if s.closed && s.sender_count == 0 {
                Err(TryRecvError::Disconnected)
            } else {
                Err(TryRecvError::Empty)
            }
        }

        pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
            if chaos::should_fail(ChaosOperation::MpscRecv) {
                return Poll::Ready(None);
            }

            let mut s = self.state.lock().unwrap();
            if let Some(v) = s.queue.pop_front() {
                if let Some(w) = s.tx_waiters.pop_front() { w.wake(); }
                Poll::Ready(Some(v))
            }
            else if s.closed && s.sender_count == 0 {
                Poll::Ready(None)
            }
            else {
                s.rx_waker = Some(cx.waker().clone());
                Poll::Pending
            }
        }
    }

    pub struct Permit<'a, T> { sender: &'a Sender<T> }

    impl<'a, T> Permit<'a, T> {
        pub fn send(self, value: T) {
            let mut s = self.sender.state.lock().unwrap();
            s.reserved -= 1;
            s.queue.push_back(value);
            if let Some(w) = s.rx_waker.take() {
                w.wake();
            }
            
            std::mem::forget(self);
        }
    }

    impl<T> Drop for Permit<'_, T> {
        fn drop(&mut self) {
            let mut s = self.sender.state.lock().unwrap();
            s.reserved -= 1;
            if let Some(w) = s.tx_waiters.pop_front() {
                w.wake();
            }
        }
    }

    pub struct OwnedPermit<T> {
        state: Arc<StdMutex<BoundedState<T>>>
    }

    impl<T> OwnedPermit<T> {
        pub fn send(self, value: T) {
            {
                let mut s = self.state.lock().unwrap();
                s.reserved -= 1;
                s.queue.push_back(value);
                if let Some(w) = s.rx_waker.take() {
                    w.wake();
                }
            }

            std::mem::forget(self);
        }
    }

    impl<T> Drop for OwnedPermit<T> {
        fn drop(&mut self) {
            let mut s = self.state.lock().unwrap();
            s.reserved -= 1;
            if let Some(w) = s.tx_waiters.pop_front() {
                w.wake();
            }
        }
    }
}

pub mod oneshot {
    use super::*;

    #[derive(Debug, Copy, Clone, Eq, PartialEq)]
    pub struct RecvError;

    impl std::fmt::Display for RecvError {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "channel closed")
        }
    }

    impl std::error::Error for RecvError {}

    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    pub enum TryRecvError {
        Empty,
        Closed,
    }

    impl std::fmt::Display for TryRecvError {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            match self {
                TryRecvError::Empty => write!(f, "channel empty"),
                TryRecvError::Closed => write!(f, "channel closed"),
            }
        }
    }

    impl std::error::Error for TryRecvError {}

    pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
        let shared = Arc::new(StdMutex::new(State { value: None, waker: None, closed: false }));
        (Sender { shared: shared.clone() }, Receiver { shared })
    }
    struct State<T> { value: Option<T>, waker: Option<Waker>, closed: bool }

    pub struct Sender<T> { shared: Arc<StdMutex<State<T>>> }

    impl<T> Sender<T> {
        pub fn send(self, t: T) -> Result<(), T> {
            if chaos::should_fail(ChaosOperation::OneshotSend) {
                return Err(t);
            }
            let mut s = self.shared.lock().unwrap();
            if s.closed {
                return Err(t);
            }
            s.value = Some(t);
            if let Some(w) = s.waker.take() {
                w.wake();
            }
            Ok(())
        }
    }

    impl<T> Drop for Sender<T> {
        fn drop(&mut self) {
            let mut s = self.shared.lock().unwrap();
            s.closed = true;
            if let Some(w) = s.waker.take() {
                w.wake();
            }
        }
    }

    pub struct Receiver<T> { shared: Arc<StdMutex<State<T>>> }

    impl<T> Receiver<T> {
        pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
            if chaos::should_fail(ChaosOperation::OneshotRecv) {
                return Err(TryRecvError::Closed);
            }
            let mut s = self.shared.lock().unwrap();
            if let Some(v) = s.value.take() {
                Ok(v)
            } else if s.closed {
                Err(TryRecvError::Closed)
            } else {
                Err(TryRecvError::Empty)
            }
        }
    }

    impl<T> Future for Receiver<T> {
        type Output = Result<T, RecvError>;
        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if chaos::should_fail(ChaosOperation::OneshotRecv) {
                return Poll::Ready(Err(RecvError));
            }
            let mut s = self.shared.lock().unwrap();
            if let Some(v) = s.value.take() { Poll::Ready(Ok(v)) }
            else if s.closed {
                Poll::Ready(Err(RecvError))
            }
            else {
                s.waker = Some(cx.waker().clone());
                Poll::Pending
            }
        }
    }
    impl<T> Drop for Receiver<T> {
        fn drop(&mut self) {
            let mut s = self.shared.lock().unwrap();
            s.closed = true;
        }
    }
}

pub mod watch {
    use super::*;

    #[derive(Debug, Clone)]
    pub struct SendError<T>(pub T);

    impl<T> std::fmt::Display for SendError<T> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "channel closed")
        }
    }

    impl<T: std::fmt::Debug> std::error::Error for SendError<T> {}

    #[derive(Debug, Clone)]
    pub struct RecvError;

    impl std::fmt::Display for RecvError {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "channel closed")
        }
    }

    impl std::error::Error for RecvError {}

    pub fn channel<T: Clone + Send + Sync + 'static>(init: T) -> (Sender<T>, Receiver<T>) {
        let state = Arc::new(StdMutex::new(State {
            value: init.clone(),
            version: 0,
            waiters: Vec::new(),
            closed: false,
        }));
        (Sender { state: state.clone() }, Receiver { state, version: 0, value: init })
    }

    struct State<T> {
        value: T,
        version: u64,
        waiters: Vec<Waker>,
        closed: bool,
    }

    pub struct Sender<T> { state: Arc<StdMutex<State<T>>> }
    
    impl<T: Clone> Sender<T> {
        pub fn send(&self, value: T) -> Result<(), SendError<T>> {
            if chaos::should_fail(ChaosOperation::WatchSend) {
                return Err(SendError(value));
            }
            
            let mut s = self.state.lock().unwrap();
            if s.closed { return Err(SendError(value)); }
            
            s.value = value;
            s.version += 1;
            for w in s.waiters.drain(..) { w.wake(); }
            Ok(())
        }

        pub fn send_modify<F>(&self, modify: F) 
        where F: FnOnce(&mut T) {
            let mut s = self.state.lock().unwrap();
            if s.closed { return; }
            modify(&mut s.value);
            s.version += 1;
            for w in s.waiters.drain(..) { w.wake(); }
        }

        pub fn borrow(&self) -> Ref<'_, T> {
            let s = self.state.lock().unwrap();
            Ref {
                inner: s.value.clone(),
                _marker: std::marker::PhantomData
            }
        }
    }

    impl<T> Drop for Sender<T> {
        fn drop(&mut self) {
            let mut s = self.state.lock().unwrap();
            s.closed = true;
            for w in s.waiters.drain(..) { w.wake(); }
        }
    }

    // Placeholder for Ref implementation simplification
    pub struct Ref<'a, T> {
        inner: T,
        _marker: std::marker::PhantomData<&'a ()>
    } // Simplified, fix later

    #[derive(Clone)]
    pub struct Receiver<T> {
        state: Arc<StdMutex<State<T>>>,
        version: u64,
        value: T,
    }

    impl<T: Clone> Receiver<T> {
        pub fn borrow(&self) -> &T { &self.value }
        
        pub fn borrow_and_update(&mut self) -> &T {
            let s = self.state.lock().unwrap();
            self.value = s.value.clone();
            self.version = s.version;
            &self.value
        }

        pub async fn changed(&mut self) -> Result<(), RecvError> {
            struct ChangedFuture<'a, T> { rx: &'a mut Receiver<T> }
            impl<T: Clone> Future for ChangedFuture<'_, T> {
                type Output = Result<(), RecvError>;
                fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
                    if chaos::should_fail(ChaosOperation::WatchChanged) {
                        return Poll::Ready(Err(RecvError));
                    }

                    let mut s = self.rx.state.lock().unwrap();
                    if s.version > self.rx.version {
                        let new_value = s.value.clone();
                        let new_version = s.version;
                        drop(s);
                        self.rx.value = new_value;
                        self.rx.version = new_version;
                        return Poll::Ready(Ok(()));
                    }
                    if s.closed { return Poll::Ready(Err(RecvError)); }

                    s.waiters.push(cx.waker().clone());
                    Poll::Pending
                }
            }
            ChangedFuture { rx: self }.await
        }

        pub fn has_changed(&self) -> Result<bool, RecvError> {
            let s = self.state.lock().unwrap();
            if s.closed {
                return Err(RecvError);
            }
            Ok(s.version > self.version)
        }
    }
}

pub struct OnceCell<T> {
    value: StdMutex<Option<T>>,
    waiters: StdMutex<Vec<Waker>>,
}

impl<T> OnceCell<T> {
    pub fn new() -> Self {
        Self {
            value: StdMutex::new(None),
            waiters: StdMutex::new(Vec::new()),
        }
    }

    pub fn get(&self) -> Option<&T> {
        let guard = self.value.lock().unwrap();
        if guard.is_some() {
            unsafe {
                let ptr = &*guard as *const Option<T>;
                (*ptr).as_ref()
            }
        } else {
            None
        }
    }

    pub fn set(&self, value: T) -> Result<(), T> {
        let mut guard = self.value.lock().unwrap();
        if guard.is_some() {
            return Err(value);
        }
        *guard = Some(value);
        drop(guard);

        let mut waiters = self.waiters.lock().unwrap();
        for waker in waiters.drain(..) {
            waker.wake();
        }
        Ok(())
    }

    pub async fn get_or_init<F>(&self, f: F) -> &T
    where
        F: Future<Output = T>,
    {
        if let Some(value) = self.get() {
            return value;
        }

        let value = f.await;
        let _ = self.set(value);
        self.get().unwrap()
    }

    pub fn get_mut(&mut self) -> Option<&mut T> {
        self.value.get_mut().unwrap().as_mut()
    }

    pub fn into_inner(self) -> Option<T> {
        self.value.into_inner().unwrap()
    }
}

impl<T> Default for OnceCell<T> {
    fn default() -> Self {
        Self::new()
    }
}

pub mod broadcast {
    use super::*;

    #[derive(Debug, Clone)]
    pub struct SendError<T>(pub T);

    impl<T> std::fmt::Display for SendError<T> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "channel closed")
        }
    }

    impl<T: std::fmt::Debug> std::error::Error for SendError<T> {}

    #[derive(Debug, Clone, PartialEq, Eq)]
    pub enum RecvError {
        Closed,
        Lagged(u64),
    }

    impl std::fmt::Display for RecvError {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            match self {
                RecvError::Closed => write!(f, "channel closed"),
                RecvError::Lagged(n) => write!(f, "channel lagged by {} messages", n),
            }
        }
    }

    impl std::error::Error for RecvError {}

    #[derive(Debug, Clone, PartialEq, Eq)]
    pub enum TryRecvError {
        Empty,
        Closed,
        Lagged(u64),
    }

    impl std::fmt::Display for TryRecvError {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            match self {
                TryRecvError::Empty => write!(f, "channel empty"),
                TryRecvError::Closed => write!(f, "channel closed"),
                TryRecvError::Lagged(n) => write!(f, "channel lagged by {} messages", n),
            }
        }
    }

    impl std::error::Error for TryRecvError {}

    #[derive(Clone)]
    struct Message<T> {
        val: T,
        seq: u64,
    }

    pub fn channel<T: Clone + Send + 'static>(capacity: usize) -> (Sender<T>, Receiver<T>) {
        let state = Arc::new(StdMutex::new(State {
            buffer: VecDeque::with_capacity(capacity),
            capacity,
            tail_seq: 0,
            closed: false,
            senders: 1,
            waiters: Vec::new(),
        }));
        (Sender { state: state.clone() }, Receiver { state, next_seq: 0 })
    }

    struct State<T> {
        buffer: VecDeque<Message<T>>,
        capacity: usize,
        tail_seq: u64,
        closed: bool,
        senders: usize,
        waiters: Vec<Waker>,
    }

    pub struct Sender<T> { state: Arc<StdMutex<State<T>>> }

    impl<T: Clone> Sender<T> {
        pub fn send(&self, value: T) -> Result<usize, SendError<T>> {
             if chaos::should_fail(ChaosOperation::BroadcastSend) {
                return Err(SendError(value));
            }

             let mut s = self.state.lock().unwrap();
             if s.closed { return Err(SendError(value)); }

             if s.buffer.len() == s.capacity {
                 s.buffer.pop_front();
             }
             
             let seq = s.tail_seq;
             s.buffer.push_back(Message { val: value, seq });
             s.tail_seq += 1;

             for w in s.waiters.drain(..) { w.wake(); }
             
             Ok(1)
        }
    }

    impl<T> Clone for Sender<T> {
        fn clone(&self) -> Self {
            let mut s = self.state.lock().unwrap();
            s.senders += 1;
            Self { state: self.state.clone() }
        }
    }

    impl<T> Drop for Sender<T> {
        fn drop(&mut self) {
            let mut s = self.state.lock().unwrap();
            s.senders -= 1;
            if s.senders == 0 {
                s.closed = true;
                for w in s.waiters.drain(..) { w.wake(); }
            }
        }
    }

    pub struct Receiver<T> {
        state: Arc<StdMutex<State<T>>>,
        next_seq: u64,
    }

    impl<T: Clone> Receiver<T> {
        pub async fn recv(&mut self) -> Result<T, RecvError> {
            struct RecvFuture<'a, T> { rx: &'a mut Receiver<T> }
            impl<T: Clone> Future for RecvFuture<'_, T> {
                type Output = Result<T, RecvError>;
                fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
                    if chaos::should_fail(ChaosOperation::BroadcastRecv) {
                        return Poll::Ready(Err(RecvError::Closed));
                    }
                    
                    let mut s = self.rx.state.lock().unwrap();
                    
                    if let Some(head) = s.buffer.front() {
                        if self.rx.next_seq < head.seq {
                            let missed = head.seq - self.rx.next_seq;
                            let tail_seq = s.tail_seq;
                            drop(s);
                            self.rx.next_seq = tail_seq;
                            return Poll::Ready(Err(RecvError::Lagged(missed)));
                        }
                    }

                    for msg in &s.buffer {
                        if msg.seq == self.rx.next_seq {
                            let val = msg.val.clone();
                            drop(s);
                            self.rx.next_seq += 1;
                            return Poll::Ready(Ok(val));
                        }
                    }

                    if s.closed && s.senders == 0 {
                        return Poll::Ready(Err(RecvError::Closed));
                    }

                    s.waiters.push(cx.waker().clone());
                    Poll::Pending
                }
            }
            RecvFuture { rx: self }.await
        }

        pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
            if chaos::should_fail(ChaosOperation::BroadcastRecv) {
                return Err(TryRecvError::Closed);
            }

            let s = self.state.lock().unwrap();

            if !s.buffer.is_empty() {
                let head_seq = s.buffer.front().unwrap().seq;
                if self.next_seq < head_seq {
                    let missed = head_seq - self.next_seq;
                    self.next_seq = s.tail_seq;
                    return Err(TryRecvError::Lagged(missed));
                }
            }

            for msg in &s.buffer {
                if msg.seq == self.next_seq {
                    let val = msg.val.clone();
                    self.next_seq += 1;
                    return Ok(val);
                }
            }

            if s.closed && s.senders == 0 {
                Err(TryRecvError::Closed)
            } else {
                Err(TryRecvError::Empty)
            }
        }

        pub fn resubscribe(&self) -> Self {
             let s = self.state.lock().unwrap();
             Self { state: self.state.clone(), next_seq: s.tail_seq }
        }
    }
    
    impl<T> Clone for Receiver<T> {
        fn clone(&self) -> Self {
             Self { state: self.state.clone(), next_seq: self.next_seq }
        }
    }
}