//! 基于预分配数组的有界通道。
//!
//! 这种味道具有固定的、积极的能力。
//!
//! 该实现基于 Dmitry Vyukov 的有界 MPMC 队列。
//!
//! Source:
//!   - <http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue>
//!   - <https://docs.google.com/document/d/1yIAYmbvL3JxOKOjuCyon7JhW4cSv1wy5hC0ApeGMV9s/pub>

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

use crate::cell::UnsafeCell;
use crate::mem::MaybeUninit;
use crate::ptr;
use crate::sync::atomic::{self, AtomicUsize, Ordering};
use crate::time::Instant;

/// 通道中的一个插槽。
struct Slot<T> {
    /// 当前邮票。
    stamp: AtomicUsize,

    /// 此插槽中的消息。
    /// 要么在 `read` 中读出，要么通过 `discard_all_messages` 丢弃。
    msg: UnsafeCell<MaybeUninit<T>>,
}

/// 数组风格的 token 类型。
#[derive(Debug)]
pub(crate) struct ArrayToken {
    /// 要读取或写入的插槽。
    slot: *const u8,

    /// 读或写后戳记存入槽中。
    stamp: usize,
}

impl Default for ArrayToken {
    #[inline]
    fn default() -> Self {
        ArrayToken { slot: ptr::null(), stamp: 0 }
    }
}

/// 基于预分配数组的有界通道。
pub(crate) struct Channel<T> {
    /// 通道之首。
    ///
    /// 这个值是一个 "stamp"，由一个缓冲区索引、一个标记位和一个圈组成，但是打包成一个 `usize`。
    /// 低位代表索引，而高位代表圈数。
    /// 头部的标记位始终为零。
    ///
    /// 消息从通道的头部弹出。
    head: CachePadded<AtomicUsize>,

    /// 通道的尾巴。
    ///
    /// 这个值是一个 "stamp"，由一个缓冲区索引、一个标记位和一个圈组成，但是打包成一个 `usize`。
    /// 低位代表索引，而高位代表圈数。
    /// 标志位表示通道断开。
    ///
    /// 消息被推入通道的尾部。
    tail: CachePadded<AtomicUsize>,

    /// 缓冲区保持槽。
    buffer: Box<[Slot<T>]>,

    /// 通道容量。
    cap: usize,

    /// 值 `{ lap: 1, mark: 0, index: 0 }` 的邮票。
    one_lap: usize,

    /// 如果这个位在尾部被置位，则表示通道断开。
    mark_bit: usize,

    /// 发送者在通道已满时等待。
    senders: SyncWaker,

    /// 接收器在通道为空且未断开连接时等待。
    receivers: SyncWaker,
}

impl<T> Channel<T> {
    /// 创建容量为 `cap` 的有界通道。
    pub(crate) fn with_capacity(cap: usize) -> Self {
        assert!(cap > 0, "capacity must be positive");

        // 计算常量 `mark_bit` 和 `one_lap`。
        let mark_bit = (cap + 1).next_power_of_two();
        let one_lap = mark_bit * 2;

        // Head 被初始化为 `{ lap: 0, mark: 0, index: 0 }`。
        let head = 0;
        // Tail 初始化为 `{ lap: 0, mark: 0, index: 0 }`。
        let tail = 0;

        // 分配一个由标记初始化的 `cap` 插槽组成的缓冲区。
        //
        let buffer: Box<[Slot<T>]> = (0..cap)
            .map(|i| {
                // 将图章设置为 `{ lap: 0, mark: 0, index: i }`。
                Slot { stamp: AtomicUsize::new(i), msg: UnsafeCell::new(MaybeUninit::uninit()) }
            })
            .collect();

        Channel {
            buffer,
            cap,
            one_lap,
            mark_bit,
            head: CachePadded::new(AtomicUsize::new(head)),
            tail: CachePadded::new(AtomicUsize::new(tail)),
            senders: SyncWaker::new(),
            receivers: SyncWaker::new(),
        }
    }

    /// 尝试为发送消息保留一个槽。
    fn start_send(&self, token: &mut Token) -> bool {
        let backoff = Backoff::new();
        let mut tail = self.tail.load(Ordering::Relaxed);

        loop {
            // 检查通道是否断开。
            if tail & self.mark_bit != 0 {
                token.array.slot = ptr::null();
                token.array.stamp = 0;
                return true;
            }

            // 解构尾巴。
            let index = tail & (self.mark_bit - 1);
            let lap = tail & !(self.one_lap - 1);

            // 检查相应的插槽。
            debug_assert!(index < self.buffer.len());
            let slot = unsafe { self.buffer.get_unchecked(index) };
            let stamp = slot.stamp.load(Ordering::Acquire);

            // 如果尾巴和邮票匹配，我们可能会尝试推动。
            if tail == stamp {
                let new_tail = if index + 1 < self.cap {
                    // 同一个 lap，增加索引。
                    // 设置为 `{ lap: lap, mark: 0, index: index + 1 }`。
                    tail + 1
                } else {
                    // 向前一圈，索引环绕到零。
                    // 设置为 `{ lap: lap.wrapping_add(1), mark: 0, index: 0 }`。
                    lap.wrapping_add(self.one_lap)
                };

                // 尝试移动尾巴。
                match self.tail.compare_exchange_weak(
                    tail,
                    new_tail,
                    Ordering::SeqCst,
                    Ordering::Relaxed,
                ) {
                    Ok(_) => {
                        // 准备 token，用于后续调用 `write`。
                        token.array.slot = slot as *const Slot<T> as *const u8;
                        token.array.stamp = tail + 1;
                        return true;
                    }
                    Err(_) => {
                        backoff.spin_light();
                        tail = self.tail.load(Ordering::Relaxed);
                    }
                }
            } else if stamp.wrapping_add(self.one_lap) == tail + 1 {
                atomic::fence(Ordering::SeqCst);
                let head = self.head.load(Ordering::Relaxed);

                // 如果头部也落后尾巴一圈...
                if head.wrapping_add(self.one_lap) == tail {
                    // ...那么通道就满了。
                    return false;
                }

                backoff.spin_light();
                tail = self.tail.load(Ordering::Relaxed);
            } else {
                // 暂停，因为我们需要等待邮票更新。
                backoff.spin_heavy();
                tail = self.tail.load(Ordering::Relaxed);
            }
        }
    }

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

        let slot: &Slot<T> = &*(token.array.slot as *const Slot<T>);

        // 将消息写入插槽并更新戳记。
        slot.msg.get().write(MaybeUninit::new(msg));
        slot.stamp.store(token.array.stamp, Ordering::Release);

        // 唤醒沉睡的接收者。
        self.receivers.notify();
        Ok(())
    }

    /// 尝试为接收消息保留一个槽。
    fn start_recv(&self, token: &mut Token) -> bool {
        let backoff = Backoff::new();
        let mut head = self.head.load(Ordering::Relaxed);

        loop {
            // 解构头部。
            let index = head & (self.mark_bit - 1);
            let lap = head & !(self.one_lap - 1);

            // 检查相应的插槽。
            debug_assert!(index < self.buffer.len());
            let slot = unsafe { self.buffer.get_unchecked(index) };
            let stamp = slot.stamp.load(Ordering::Acquire);

            // 如果戳记领先于头部 1，我们可能会尝试弹出。
            if head + 1 == stamp {
                let new = if index + 1 < self.cap {
                    // 同一个 lap，增加索引。
                    // 设置为 `{ lap: lap, mark: 0, index: index + 1 }`。
                    head + 1
                } else {
                    // 向前一圈，索引环绕到零。
                    // 设置为 `{ lap: lap.wrapping_add(1), mark: 0, index: 0 }`。
                    lap.wrapping_add(self.one_lap)
                };

                // 尝试移动头部。
                match self.head.compare_exchange_weak(
                    head,
                    new,
                    Ordering::SeqCst,
                    Ordering::Relaxed,
                ) {
                    Ok(_) => {
                        // 准备 token，用于后续调用 `read`。
                        token.array.slot = slot as *const Slot<T> as *const u8;
                        token.array.stamp = head.wrapping_add(self.one_lap);
                        return true;
                    }
                    Err(_) => {
                        backoff.spin_light();
                        head = self.head.load(Ordering::Relaxed);
                    }
                }
            } else if stamp == head {
                atomic::fence(Ordering::SeqCst);
                let tail = self.tail.load(Ordering::Relaxed);

                // 如果尾巴等于头，那就意味着通道是空的。
                if (tail & !self.mark_bit) == head {
                    // 如果通信断开...
                    if tail & self.mark_bit != 0 {
                        // ...然后收到错误。
                        token.array.slot = ptr::null();
                        token.array.stamp = 0;
                        return true;
                    } else {
                        // 否则，接收操作未就绪。
                        return false;
                    }
                }

                backoff.spin_light();
                head = self.head.load(Ordering::Relaxed);
            } else {
                // 暂停，因为我们需要等待邮票更新。
                backoff.spin_heavy();
                head = self.head.load(Ordering::Relaxed);
            }
        }
    }

    /// 读取来自通道的消息。
    pub(crate) unsafe fn read(&self, token: &mut Token) -> Result<T, ()> {
        if token.array.slot.is_null() {
            // 通讯已断开。
            return Err(());
        }

        let slot: &Slot<T> = &*(token.array.slot as *const Slot<T>);

        // 从插槽中读取消息并更新标记。
        let msg = slot.msg.get().read().assume_init();
        slot.stamp.store(token.array.stamp, Ordering::Release);

        // 唤醒沉睡的发送者。
        self.senders.notify();
        Ok(msg)
    }

    /// 尝试向通道发送消息。
    pub(crate) fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
        let token = &mut Token::default();
        if self.start_send(token) {
            unsafe { self.write(token, msg).map_err(TrySendError::Disconnected) }
        } else {
            Err(TrySendError::Full(msg))
        }
    }

    /// 向通道发送消息。
    pub(crate) fn send(
        &self,
        msg: T,
        deadline: Option<Instant>,
    ) -> Result<(), SendTimeoutError<T>> {
        let token = &mut Token::default();
        loop {
            // 尝试发送消息。
            if self.start_send(token) {
                let res = unsafe { self.write(token, msg) };
                return res.map_err(SendTimeoutError::Disconnected);
            }

            if let Some(d) = deadline {
                if Instant::now() >= d {
                    return Err(SendTimeoutError::Timeout(msg));
                }
            }

            Context::with(|cx| {
                // 准备阻塞，直到接收者叫醒我们。
                let oper = Operation::hook(token);
                self.senders.register(oper, cx);

                // 通道刚刚准备好了吗?
                if !self.is_full() || self.is_disconnected() {
                    let _ = cx.try_select(Selected::Aborted);
                }

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

                match sel {
                    Selected::Waiting => unreachable!(),
                    Selected::Aborted | Selected::Disconnected => {
                        self.senders.unregister(oper).unwrap();
                    }
                    Selected::Operation(_) => {}
                }
            });
        }
    }

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

        if self.start_recv(token) {
            unsafe { self.read(token).map_err(|_| TryRecvError::Disconnected) }
        } else {
            Err(TryRecvError::Empty)
        }
    }

    /// 收到来自通道的消息。
    pub(crate) fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
        let token = &mut Token::default();
        loop {
            // 尝试接收消息。
            if self.start_recv(token) {
                let res = unsafe { self.read(token) };
                return res.map_err(|_| RecvTimeoutError::Disconnected);
            }

            if let Some(d) = deadline {
                if Instant::now() >= d {
                    return Err(RecvTimeoutError::Timeout);
                }
            }

            Context::with(|cx| {
                // 准备阻塞，直到发送者唤醒我们。
                let oper = Operation::hook(token);
                self.receivers.register(oper, cx);

                // 通道刚刚准备好了吗?
                if !self.is_empty() || self.is_disconnected() {
                    let _ = cx.try_select(Selected::Aborted);
                }

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

                match sel {
                    Selected::Waiting => unreachable!(),
                    Selected::Aborted | Selected::Disconnected => {
                        self.receivers.unregister(oper).unwrap();
                        // 如果通信断开，我们仍然需要检查剩余消息。
                        //
                    }
                    Selected::Operation(_) => {}
                }
            });
        }
    }

    /// 返回通道内的当前消息数。
    pub(crate) fn len(&self) -> usize {
        loop {
            // 加载尾部，然后加载头部。
            let tail = self.tail.load(Ordering::SeqCst);
            let head = self.head.load(Ordering::SeqCst);

            // 如果尾巴没有改变，我们就可以使用一致的值。
            if self.tail.load(Ordering::SeqCst) == tail {
                let hix = head & (self.mark_bit - 1);
                let tix = tail & (self.mark_bit - 1);

                return if hix < tix {
                    tix - hix
                } else if hix > tix {
                    self.cap - hix + tix
                } else if (tail & !self.mark_bit) == head {
                    0
                } else {
                    self.cap
                };
            }
        }
    }

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

    /// 断开发送者并唤醒所有被阻止的接收者。
    ///
    /// 如果此调用断开了通信，则返回 `true`。
    pub(crate) fn disconnect_senders(&self) -> bool {
        let tail = self.tail.fetch_or(self.mark_bit, Ordering::SeqCst);

        if tail & self.mark_bit == 0 {
            self.receivers.disconnect();
            true
        } else {
            false
        }
    }

    /// 断开接收者并唤醒所有被阻止的发送者。
    ///
    /// 如果此调用断开了通信，则返回 `true`。
    ///
    /// # Safety
    /// 丢弃最后一个接收者只能被称为一次。
    /// 必须通过获取命令或更强的命令观察到所有其他接收器的销毁。
    ///
    pub(crate) unsafe fn disconnect_receivers(&self) -> bool {
        let tail = self.tail.fetch_or(self.mark_bit, Ordering::SeqCst);
        let disconnected = if tail & self.mark_bit == 0 {
            self.senders.disconnect();
            true
        } else {
            false
        };

        self.discard_all_messages(tail);
        disconnected
    }

    /// 丢弃所有消息。
    ///
    /// `tail` 应该是 `tail` 的当前 (也是最后一个) 值。
    ///
    /// # Panicking
    /// 如果析构函数出现 panic，则剩余的消息将被泄露，与无界通道的行为相匹配。
    ///
    ///
    /// # Safety
    /// 此方法只能在丢弃最后一个接收者时调用。
    /// 必须通过获取命令或更强的命令观察到所有其他接收器的销毁。
    ///
    unsafe fn discard_all_messages(&self, tail: usize) {
        debug_assert!(self.is_disconnected());

        // 只有接收者修改 `head`，所以由于我们是最后一个，所以这个值不会改变，也不会被观察到 (因为断开后不能发送新消息)。
        //
        //
        let mut head = self.head.load(Ordering::Relaxed);
        let tail = tail & !self.mark_bit;

        let backoff = Backoff::new();
        loop {
            // 解构头部。
            let index = head & (self.mark_bit - 1);
            let lap = head & !(self.one_lap - 1);

            // 检查相应的插槽。
            debug_assert!(index < self.buffer.len());
            let slot = unsafe { self.buffer.get_unchecked(index) };
            let stamp = slot.stamp.load(Ordering::Acquire);

            // 如果戳记在头部前面 1，我们可以丢弃该消息。
            if head + 1 == stamp {
                head = if index + 1 < self.cap {
                    // 同一个 lap，增加索引。
                    // 设置为 `{ lap: lap, mark: 0, index: index + 1 }`。
                    head + 1
                } else {
                    // 向前一圈，索引环绕到零。
                    // 设置为 `{ lap: lap.wrapping_add(1), mark: 0, index: 0 }`。
                    lap.wrapping_add(self.one_lap)
                };

                unsafe {
                    (*slot.msg.get()).assume_init_drop();
                }
            // 如果尾巴等于头，那就意味着通道是空的。
            } else if tail == head {
                return;
            // 否则，发送者即将写入槽，因此我们需要等待它更新标记。
            //
            } else {
                backoff.spin_heavy();
            }
        }
    }

    /// 如果通信断开，则返回 `true`。
    pub(crate) fn is_disconnected(&self) -> bool {
        self.tail.load(Ordering::SeqCst) & self.mark_bit != 0
    }

    /// 如果通道为空，则返回 `true`。
    pub(crate) fn is_empty(&self) -> bool {
        let head = self.head.load(Ordering::SeqCst);
        let tail = self.tail.load(Ordering::SeqCst);

        // 尾巴等于头吗?
        //
        // Note: 如果在我们加载尾部之前头部发生变化，这意味着有一段时间通道不为空，因此返回 `false` 是安全的。
        //
        (tail & !self.mark_bit) == head
    }

    /// 如果通道已满，则返回 `true`。
    pub(crate) fn is_full(&self) -> bool {
        let tail = self.tail.load(Ordering::SeqCst);
        let head = self.head.load(Ordering::SeqCst);

        // 头部是否比尾部滞后一圈?
        //
        // Note: 如果尾巴在我们加载头部之前发生变化，这意味着有一段时间通道没有满，所以返回 `false` 是安全的。
        //
        head.wrapping_add(self.one_lap) == tail & !self.mark_bit
    }
}