//! 无界通道实现为链表。

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::marker::PhantomData;
use crate::mem::MaybeUninit;
use crate::ptr;
use crate::sync::atomic::{self, AtomicPtr, AtomicUsize, Ordering};
use crate::time::Instant;

// 指示插槽状态的位:
// * 如果消息已写入槽中，则设置 `WRITE`。
// * 如果从插槽中读取了一条消息，则设置 `READ`。
// * 如果块正在销毁，则设置 `DESTROY`。
const WRITE: usize = 1;
const READ: usize = 2;
const DESTROY: usize = 4;

// 每个块包含一个 "lap" 的索引。
const LAP: usize = 32;
// 一个块可以容纳的最大消息数。
const BLOCK_CAP: usize = LAP - 1;
// 为元数据保留了多少低位。
const SHIFT: usize = 1;
// 有两个不同的目的:
// * 如果设置在头部，表示该块不是最后一个。
// * 如果设置在 tail，表示通道断开。
const MARK_BIT: usize = 1;

/// 块中的一个插槽。
struct Slot<T> {
    /// 消息。
    msg: UnsafeCell<MaybeUninit<T>>,

    /// 插槽的状态。
    state: AtomicUsize,
}

impl<T> Slot<T> {
    /// 等待消息写入插槽。
    fn wait_write(&self) {
        let backoff = Backoff::new();
        while self.state.load(Ordering::Acquire) & WRITE == 0 {
            backoff.spin_heavy();
        }
    }
}

/// 链表中的块。
///
/// 列表中的每个块最多可以容纳 `BLOCK_CAP` 条消息。
struct Block<T> {
    /// 链表中的下一个块。
    next: AtomicPtr<Block<T>>,

    /// 消息槽。
    slots: [Slot<T>; BLOCK_CAP],
}

impl<T> Block<T> {
    /// 创建一个空块。
    fn new() -> Block<T> {
        // SAFETY: 这是安全的，因为：
        //  [1] `Block::next` (AtomicPtr) 可以安全地进行零初始化。
        //  由于 [3, 4]，[2] `Block::slots` (Array) 可以安全地进行零初始化。
        //  [3] `Slot::msg` (UnsafeCell) 可以安全地进行零初始化，因为它拥有一个 MaybeUninit。
        //
        //  [4] `Slot::state` (AtomicUsize) 可以安全地进行零初始化。
        unsafe { MaybeUninit::zeroed().assume_init() }
    }

    /// 等待直到设置下一个指针。
    fn wait_next(&self) -> *mut Block<T> {
        let backoff = Backoff::new();
        loop {
            let next = self.next.load(Ordering::Acquire);
            if !next.is_null() {
                return next;
            }
            backoff.spin_heavy();
        }
    }

    /// 在从 `start` 开始的插槽中设置 `DESTROY` 位并销毁该块。
    unsafe fn destroy(this: *mut Block<T>, start: usize) {
        // 没有必要在最后一个槽中设置 `DESTROY` 位，因为那个槽已经开始破坏块。
        //
        for i in start..BLOCK_CAP - 1 {
            let slot = (*this).slots.get_unchecked(i);

            // 如果线程仍在使用插槽，则标记 `DESTROY` 位。
            if slot.state.load(Ordering::Acquire) & READ == 0
                && slot.state.fetch_or(DESTROY, Ordering::AcqRel) & READ == 0
            {
                // 如果线程仍在使用该槽，它将继续销毁该块。
                return;
            }
        }

        // 没有线程正在使用该块，现在可以安全地销毁它。
        drop(Box::from_raw(this));
    }
}

/// 通道中的一个位置。
#[derive(Debug)]
struct Position<T> {
    /// 通道中的索引。
    index: AtomicUsize,

    /// 链表中的块。
    block: AtomicPtr<Block<T>>,
}

/// 列表风格的 token 类型。
#[derive(Debug)]
pub(crate) struct ListToken {
    /// 插槽块。
    block: *const u8,

    /// 块中的偏移量。
    offset: usize,
}

impl Default for ListToken {
    #[inline]
    fn default() -> Self {
        ListToken { block: ptr::null(), offset: 0 }
    }
}

/// 无界通道实现为链表。
///
/// 发送到通道的每条消息都分配有一个序列号，即索引。
/// 索引表示为 `usize` 类型的数字并在溢出时换行。
///
/// 连续的消息被分组到块中，以减轻分配器的压力并提高缓存效率。
///
pub(crate) struct Channel<T> {
    /// 通道之首。
    head: CachePadded<Position<T>>,

    /// 通道的尾巴。
    tail: CachePadded<Position<T>>,

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

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

impl<T> Channel<T> {
    /// 创建一个新的无限通道。
    pub(crate) fn new() -> Self {
        Channel {
            head: CachePadded::new(Position {
                block: AtomicPtr::new(ptr::null_mut()),
                index: AtomicUsize::new(0),
            }),
            tail: CachePadded::new(Position {
                block: AtomicPtr::new(ptr::null_mut()),
                index: AtomicUsize::new(0),
            }),
            receivers: SyncWaker::new(),
            _marker: PhantomData,
        }
    }

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

        loop {
            // 检查通道是否断开。
            if tail & MARK_BIT != 0 {
                token.list.block = ptr::null();
                return true;
            }

            // 计算索引在块中的偏移量。
            let offset = (tail >> SHIFT) % LAP;

            // 如果我们到达了块的末尾，请等到安装下一个块。
            if offset == BLOCK_CAP {
                backoff.spin_heavy();
                tail = self.tail.index.load(Ordering::Acquire);
                block = self.tail.block.load(Ordering::Acquire);
                continue;
            }

            // 如果我们要安装下一个块，请提前分配它，以使其他线程的等待时间尽可能短。
            //
            if offset + 1 == BLOCK_CAP && next_block.is_none() {
                next_block = Some(Box::new(Block::<T>::new()));
            }

            // 如果这是要发送到通道的第一条消息，我们需要分配第一个块并安装它。
            //
            if block.is_null() {
                let new = Box::into_raw(Box::new(Block::<T>::new()));

                if self
                    .tail
                    .block
                    .compare_exchange(block, new, Ordering::Release, Ordering::Relaxed)
                    .is_ok()
                {
                    self.head.block.store(new, Ordering::Release);
                    block = new;
                } else {
                    next_block = unsafe { Some(Box::from_raw(new)) };
                    tail = self.tail.index.load(Ordering::Acquire);
                    block = self.tail.block.load(Ordering::Acquire);
                    continue;
                }
            }

            let new_tail = tail + (1 << SHIFT);

            // 尝试向前推进尾巴。
            match self.tail.index.compare_exchange_weak(
                tail,
                new_tail,
                Ordering::SeqCst,
                Ordering::Acquire,
            ) {
                Ok(_) => unsafe {
                    // 如果我们已经到达块的末尾，请安装下一个。
                    if offset + 1 == BLOCK_CAP {
                        let next_block = Box::into_raw(next_block.unwrap());
                        self.tail.block.store(next_block, Ordering::Release);
                        self.tail.index.fetch_add(1 << SHIFT, Ordering::Release);
                        (*block).next.store(next_block, Ordering::Release);
                    }

                    token.list.block = block as *const u8;
                    token.list.offset = offset;
                    return true;
                },
                Err(_) => {
                    backoff.spin_light();
                    tail = self.tail.index.load(Ordering::Acquire);
                    block = self.tail.block.load(Ordering::Acquire);
                }
            }
        }
    }

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

        // 将消息写入插槽。
        let block = token.list.block as *mut Block<T>;
        let offset = token.list.offset;
        let slot = (*block).slots.get_unchecked(offset);
        slot.msg.get().write(MaybeUninit::new(msg));
        slot.state.fetch_or(WRITE, Ordering::Release);

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

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

        loop {
            // 计算索引在块中的偏移量。
            let offset = (head >> SHIFT) % LAP;

            // 如果我们到达了块的末尾，请等到安装下一个块。
            if offset == BLOCK_CAP {
                backoff.spin_heavy();
                head = self.head.index.load(Ordering::Acquire);
                block = self.head.block.load(Ordering::Acquire);
                continue;
            }

            let mut new_head = head + (1 << SHIFT);

            if new_head & MARK_BIT == 0 {
                atomic::fence(Ordering::SeqCst);
                let tail = self.tail.index.load(Ordering::Relaxed);

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

                // 如果头部和尾部不在同一个块中，则在头部设置 `MARK_BIT`。
                if (head >> SHIFT) / LAP != (tail >> SHIFT) / LAP {
                    new_head |= MARK_BIT;
                }
            }

            // 仅当第一条消息被发送到通道时，此处的块才可以为空。
            // 在这种情况下，只需等到它被初始化。
            if block.is_null() {
                backoff.spin_heavy();
                head = self.head.index.load(Ordering::Acquire);
                block = self.head.block.load(Ordering::Acquire);
                continue;
            }

            // 尝试向前移动头部索引。
            match self.head.index.compare_exchange_weak(
                head,
                new_head,
                Ordering::SeqCst,
                Ordering::Acquire,
            ) {
                Ok(_) => unsafe {
                    // 如果我们已经到达街区的尽头，请移动到下一个。
                    if offset + 1 == BLOCK_CAP {
                        let next = (*block).wait_next();
                        let mut next_index = (new_head & !MARK_BIT).wrapping_add(1 << SHIFT);
                        if !(*next).next.load(Ordering::Relaxed).is_null() {
                            next_index |= MARK_BIT;
                        }

                        self.head.block.store(next, Ordering::Release);
                        self.head.index.store(next_index, Ordering::Release);
                    }

                    token.list.block = block as *const u8;
                    token.list.offset = offset;
                    return true;
                },
                Err(_) => {
                    backoff.spin_light();
                    head = self.head.index.load(Ordering::Acquire);
                    block = self.head.block.load(Ordering::Acquire);
                }
            }
        }
    }

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

        // 阅读消息。
        let block = token.list.block as *mut Block<T>;
        let offset = token.list.offset;
        let slot = (*block).slots.get_unchecked(offset);
        slot.wait_write();
        let msg = slot.msg.get().read().assume_init();

        // 如果我们已经到达末尾，或者如果另一个线程想要销售但不能因为我们正忙于从槽读取而不能销售块。
        //
        if offset + 1 == BLOCK_CAP {
            Block::destroy(block, 0);
        } else if slot.state.fetch_or(READ, Ordering::AcqRel) & DESTROY != 0 {
            Block::destroy(block, offset + 1);
        }

        Ok(msg)
    }

    /// 尝试向通道发送消息。
    pub(crate) fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
        self.send(msg, None).map_err(|err| match err {
            SendTimeoutError::Disconnected(msg) => TrySendError::Disconnected(msg),
            SendTimeoutError::Timeout(_) => unreachable!(),
        })
    }

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

    /// 尝试在不阻塞的情况下接收消息。
    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) {
                unsafe {
                    return self.read(token).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 mut tail = self.tail.index.load(Ordering::SeqCst);
            let mut head = self.head.index.load(Ordering::SeqCst);

            // 如果尾部索引没有改变，我们就有了一致的索引可以使用。
            if self.tail.index.load(Ordering::SeqCst) == tail {
                // 擦除低位。
                tail &= !((1 << SHIFT) - 1);
                head &= !((1 << SHIFT) - 1);

                // 如果索引落在块末端，请修复它们。
                if (tail >> SHIFT) & (LAP - 1) == LAP - 1 {
                    tail = tail.wrapping_add(1 << SHIFT);
                }
                if (head >> SHIFT) & (LAP - 1) == LAP - 1 {
                    head = head.wrapping_add(1 << SHIFT);
                }

                // 旋转索引，使头部落入第一个块。
                let lap = (head >> SHIFT) / LAP;
                tail = tail.wrapping_sub((lap * LAP) << SHIFT);
                head = head.wrapping_sub((lap * LAP) << SHIFT);

                // 删除低位。
                tail >>= SHIFT;
                head >>= SHIFT;

                // 返回差值减去尾部和头部之间的块数。
                return tail - head - tail / LAP;
            }
        }
    }

    /// 返回通道的容量。
    pub(crate) fn capacity(&self) -> Option<usize> {
        None
    }

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

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

    /// 断开接收器。
    ///
    /// 如果此调用断开了通信，则返回 `true`。
    pub(crate) fn disconnect_receivers(&self) -> bool {
        let tail = self.tail.index.fetch_or(MARK_BIT, Ordering::SeqCst);

        if tail & MARK_BIT == 0 {
            // 如果接收者首先被丢弃，则丢弃所有消息以急切地释放内存。
            //
            self.discard_all_messages();
            true
        } else {
            false
        }
    }

    /// 丢弃所有消息。
    ///
    /// 仅当所有接收者都丢弃时才应调用此方法。
    fn discard_all_messages(&self) {
        let backoff = Backoff::new();
        let mut tail = self.tail.index.load(Ordering::Acquire);
        loop {
            let offset = (tail >> SHIFT) % LAP;
            if offset != BLOCK_CAP {
                break;
            }

            // 对 tail 的新更新将被 MARK_BIT 拒绝并中止，除非它处于边界。
            // 我们需要等待更新生效，否则可能会出现内存泄漏。
            //
            backoff.spin_heavy();
            tail = self.tail.index.load(Ordering::Acquire);
        }

        let mut head = self.head.index.load(Ordering::Acquire);
        let mut block = self.head.block.load(Ordering::Acquire);

        // 如果我们要丢弃消息，我们需要与初始化同步
        if head >> SHIFT != tail >> SHIFT {
            // 仅当发送方正在初始化通道而另一个发送方设法通过将消息插入半初始化通道并推进尾部来发送消息时，此处的块才可以为空。
            //
            // 在这种情况下，只需等到它被初始化。
            //
            while block.is_null() {
                backoff.spin_heavy();
                block = self.head.block.load(Ordering::Acquire);
            }
        }

        unsafe {
            // 抛弃头尾之间的所有消息并释放堆分配的块。
            while head >> SHIFT != tail >> SHIFT {
                let offset = (head >> SHIFT) % LAP;

                if offset < BLOCK_CAP {
                    // 丢弃插槽中的消息。
                    let slot = (*block).slots.get_unchecked(offset);
                    slot.wait_write();
                    let p = &mut *slot.msg.get();
                    p.as_mut_ptr().drop_in_place();
                } else {
                    (*block).wait_next();
                    // 释放块并移动到下一个。
                    let next = (*block).next.load(Ordering::Acquire);
                    drop(Box::from_raw(block));
                    block = next;
                }

                head = head.wrapping_add(1 << SHIFT);
            }

            // 释放最后一个剩余块。
            if !block.is_null() {
                drop(Box::from_raw(block));
            }
        }
        head &= !MARK_BIT;
        self.head.block.store(ptr::null_mut(), Ordering::Release);
        self.head.index.store(head, Ordering::Release);
    }

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

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

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

impl<T> Drop for Channel<T> {
    fn drop(&mut self) {
        let mut head = self.head.index.load(Ordering::Relaxed);
        let mut tail = self.tail.index.load(Ordering::Relaxed);
        let mut block = self.head.block.load(Ordering::Relaxed);

        // 擦除低位。
        head &= !((1 << SHIFT) - 1);
        tail &= !((1 << SHIFT) - 1);

        unsafe {
            // 抛弃头尾之间的所有消息并释放堆分配的块。
            while head != tail {
                let offset = (head >> SHIFT) % LAP;

                if offset < BLOCK_CAP {
                    // 丢弃插槽中的消息。
                    let slot = (*block).slots.get_unchecked(offset);
                    let p = &mut *slot.msg.get();
                    p.as_mut_ptr().drop_in_place();
                } else {
                    // 释放块并移动到下一个。
                    let next = (*block).next.load(Ordering::Relaxed);
                    drop(Box::from_raw(block));
                    block = next;
                }

                head = head.wrapping_add(1 << SHIFT);
            }

            // 释放最后一个剩余块。
            if !block.is_null() {
                drop(Box::from_raw(block));
            }
        }
    }
}