//! 零容量通道。
//!
//! 这种通道也称为 *rendezvous* 通道。

use super::context::Context;
use super::error::*;
use super::select::{Operation, Selected, Token};
use super::utils::Backoff;
use super::waker::Waker;

use crate::cell::UnsafeCell;
use crate::marker::PhantomData;
use crate::sync::atomic::{AtomicBool, Ordering};
use crate::sync::Mutex;
use crate::time::Instant;
use crate::{fmt, ptr};

/// 指向数据包的指针。
pub(crate) struct ZeroToken(*mut ());

impl Default for ZeroToken {
    fn default() -> Self {
        Self(ptr::null_mut())
    }
}

impl fmt::Debug for ZeroToken {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Debug::fmt(&(self.0 as usize), f)
    }
}

/// 用于将一条消息从发送者传递给接收者的插槽。
struct Packet<T> {
    /// 如果数据包分配在栈上，则等于 `true`。
    on_stack: bool,

    /// 一旦数据包准备好读取或写入，就等于 `true`。
    ready: AtomicBool,

    /// 消息。
    msg: UnsafeCell<Option<T>>,
}

impl<T> Packet<T> {
    /// 在栈上创建一个空包。
    fn empty_on_stack() -> Packet<T> {
        Packet { on_stack: true, ready: AtomicBool::new(false), msg: UnsafeCell::new(None) }
    }

    /// 在栈上创建一个包含消息的数据包。
    fn message_on_stack(msg: T) -> Packet<T> {
        Packet { on_stack: true, ready: AtomicBool::new(false), msg: UnsafeCell::new(Some(msg)) }
    }

    /// 等待数据包准备好读取或写入。
    fn wait_ready(&self) {
        let backoff = Backoff::new();
        while !self.ready.load(Ordering::Acquire) {
            backoff.spin_heavy();
        }
    }
}

/// 零容量通道的内部表示。
struct Inner {
    /// 等待与接收操作配对的发送者。
    senders: Waker,

    /// 等待与发送操作配对的接收器。
    receivers: Waker,

    /// 当通道断开时等于 `true`。
    is_disconnected: bool,
}

/// 零容量通道。
pub(crate) struct Channel<T> {
    /// 通道的内部表示。
    inner: Mutex<Inner>,

    /// 表示丢弃 `Channel<T>` 可以丢弃 `T` 类型的值。
    _marker: PhantomData<T>,
}

impl<T> Channel<T> {
    /// 构建一个新的零容量通道。
    pub(crate) fn new() -> Self {
        Channel {
            inner: Mutex::new(Inner {
                senders: Waker::new(),
                receivers: Waker::new(),
                is_disconnected: false,
            }),
            _marker: PhantomData,
        }
    }

    /// 将消息写入数据包。
    pub(crate) unsafe fn write(&self, token: &mut Token, msg: T) -> Result<(), T> {
        // 如果没有数据包，则通信断开。
        if token.zero.0.is_null() {
            return Err(msg);
        }

        let packet = &*(token.zero.0 as *const Packet<T>);
        packet.msg.get().write(Some(msg));
        packet.ready.store(true, Ordering::Release);
        Ok(())
    }

    /// 从数据包中读取消息。
    pub(crate) unsafe fn read(&self, token: &mut Token) -> Result<T, ()> {
        // 如果没有数据包，则通信断开。
        if token.zero.0.is_null() {
            return Err(());
        }

        let packet = &*(token.zero.0 as *const Packet<T>);

        if packet.on_stack {
            // 消息从一开始就在数据包中，所以不需要等待。
            // 但是，在读取消息后，我们需要将 `ready` 设置为 `true`，以表示该数据包可以被销毁。
            //
            let msg = packet.msg.get().replace(None).unwrap();
            packet.ready.store(true, Ordering::Release);
            Ok(msg)
        } else {
            // 等到消息可用，然后读取它并销毁堆分配的数据包。
            //
            packet.wait_ready();
            let msg = packet.msg.get().replace(None).unwrap();
            drop(Box::from_raw(token.zero.0 as *mut Packet<T>));
            Ok(msg)
        }
    }

    /// 尝试向通道发送消息。
    pub(crate) fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
        let token = &mut Token::default();
        let mut inner = self.inner.lock().unwrap();

        // 如果有等待的接收者，就与它配对。
        if let Some(operation) = inner.receivers.try_select() {
            token.zero.0 = operation.packet;
            drop(inner);
            unsafe {
                self.write(token, msg).ok().unwrap();
            }
            Ok(())
        } else if inner.is_disconnected {
            Err(TrySendError::Disconnected(msg))
        } else {
            Err(TrySendError::Full(msg))
        }
    }

    /// 向通道发送消息。
    pub(crate) fn send(
        &self,
        msg: T,
        deadline: Option<Instant>,
    ) -> Result<(), SendTimeoutError<T>> {
        let token = &mut Token::default();
        let mut inner = self.inner.lock().unwrap();

        // 如果有等待的接收者，就与它配对。
        if let Some(operation) = inner.receivers.try_select() {
            token.zero.0 = operation.packet;
            drop(inner);
            unsafe {
                self.write(token, msg).ok().unwrap();
            }
            return Ok(());
        }

        if inner.is_disconnected {
            return Err(SendTimeoutError::Disconnected(msg));
        }

        Context::with(|cx| {
            // 准备阻塞，直到接收者叫醒我们。
            let oper = Operation::hook(token);
            let mut packet = Packet::<T>::message_on_stack(msg);
            inner.senders.register_with_packet(oper, &mut packet as *mut Packet<T> as *mut (), cx);
            inner.receivers.notify();
            drop(inner);

            // 阻塞当前线程。
            let sel = cx.wait_until(deadline);

            match sel {
                Selected::Waiting => unreachable!(),
                Selected::Aborted => {
                    self.inner.lock().unwrap().senders.unregister(oper).unwrap();
                    let msg = unsafe { packet.msg.get().replace(None).unwrap() };
                    Err(SendTimeoutError::Timeout(msg))
                }
                Selected::Disconnected => {
                    self.inner.lock().unwrap().senders.unregister(oper).unwrap();
                    let msg = unsafe { packet.msg.get().replace(None).unwrap() };
                    Err(SendTimeoutError::Disconnected(msg))
                }
                Selected::Operation(_) => {
                    // 等到消息被读取，然后丢弃数据包。
                    packet.wait_ready();
                    Ok(())
                }
            }
        })
    }

    /// 尝试在不阻塞的情况下接收消息。
    pub(crate) fn try_recv(&self) -> Result<T, TryRecvError> {
        let token = &mut Token::default();
        let mut inner = self.inner.lock().unwrap();

        // 如果有等待发送者，请与它配对。
        if let Some(operation) = inner.senders.try_select() {
            token.zero.0 = operation.packet;
            drop(inner);
            unsafe { self.read(token).map_err(|_| TryRecvError::Disconnected) }
        } else if inner.is_disconnected {
            Err(TryRecvError::Disconnected)
        } else {
            Err(TryRecvError::Empty)
        }
    }

    /// 收到来自通道的消息。
    pub(crate) fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
        let token = &mut Token::default();
        let mut inner = self.inner.lock().unwrap();

        // 如果有等待发送者，请与它配对。
        if let Some(operation) = inner.senders.try_select() {
            token.zero.0 = operation.packet;
            drop(inner);
            unsafe {
                return self.read(token).map_err(|_| RecvTimeoutError::Disconnected);
            }
        }

        if inner.is_disconnected {
            return Err(RecvTimeoutError::Disconnected);
        }

        Context::with(|cx| {
            // 准备阻塞，直到发送者唤醒我们。
            let oper = Operation::hook(token);
            let mut packet = Packet::<T>::empty_on_stack();
            inner.receivers.register_with_packet(
                oper,
                &mut packet as *mut Packet<T> as *mut (),
                cx,
            );
            inner.senders.notify();
            drop(inner);

            // 阻塞当前线程。
            let sel = cx.wait_until(deadline);

            match sel {
                Selected::Waiting => unreachable!(),
                Selected::Aborted => {
                    self.inner.lock().unwrap().receivers.unregister(oper).unwrap();
                    Err(RecvTimeoutError::Timeout)
                }
                Selected::Disconnected => {
                    self.inner.lock().unwrap().receivers.unregister(oper).unwrap();
                    Err(RecvTimeoutError::Disconnected)
                }
                Selected::Operation(_) => {
                    // 等到消息出现，然后阅读它。
                    packet.wait_ready();
                    unsafe { Ok(packet.msg.get().replace(None).unwrap()) }
                }
            }
        })
    }

    /// 断开通信并唤醒所有被阻止的发送者和接收者。
    ///
    /// 如果此调用断开了通信，则返回 `true`。
    pub(crate) fn disconnect(&self) -> bool {
        let mut inner = self.inner.lock().unwrap();

        if !inner.is_disconnected {
            inner.is_disconnected = true;
            inner.senders.disconnect();
            inner.receivers.disconnect();
            true
        } else {
            false
        }
    }

    /// 返回通道内的当前消息数。
    pub(crate) fn len(&self) -> usize {
        0
    }

    /// 返回通道的容量。
    #[allow(clippy::unnecessary_wraps)] // 这是故意的。
    pub(crate) fn capacity(&self) -> Option<usize> {
        Some(0)
    }

    /// 如果通道为空，则返回 `true`。
    pub(crate) fn is_empty(&self) -> bool {
        true
    }

    /// 如果通道已满，则返回 `true`。
    pub(crate) fn is_full(&self) -> bool {
        true
    }
}