/// 共享通道。
///
/// 这是通道的风格，不一定针对任何特定用例进行优化，但在使用方式上是最通用的。
/// 共享通道是可克隆的，允许多个发送者。
///
/// 可以在父模块的注释中找到高级实现细节。
/// 您还将注意到，共享通道和流通道的实现非常相似，这绝非偶然！
///
///
pub use self::Failure::*;
use self::StartResult::*;

use core::cmp;
use core::intrinsics::abort;

use crate::cell::UnsafeCell;
use crate::ptr;
use crate::sync::atomic::{AtomicBool, AtomicIsize, AtomicUsize, Ordering};
use crate::sync::mpsc::blocking::{self, SignalToken};
use crate::sync::mpsc::mpsc_queue as mpsc;
use crate::sync::{Mutex, MutexGuard};
use crate::thread;
use crate::time::Instant;

const DISCONNECTED: isize = isize::MIN;
const FUDGE: isize = 1024;
const MAX_REFCOUNT: usize = (isize::MAX) as usize;
#[cfg(test)]
const MAX_STEALS: isize = 5;
#[cfg(not(test))]
const MAX_STEALS: isize = 1 << 20;

pub struct Packet<T> {
    queue: mpsc::Queue<T>,
    cnt: AtomicIsize,          // 该通道上有多少项
    steals: UnsafeCell<isize>, // 一个端口在没有阻塞的情况下接收了多少次？
    to_wake: AtomicUsize,      // SignalToken 唤醒

    // 当前正在使用此数据包的通道数。
    channels: AtomicUsize,

    // 有关这些功能的用途，请参见 Port::drop 中的讨论和通道发送方法。
    //
    port_dropped: AtomicBool,
    sender_drain: AtomicIsize,

    // 此锁在执行过程中保护此实现的各个部分
    // select()
    select_lock: Mutex<()>,
}

pub enum Failure {
    Empty,
    Disconnected,
}

#[derive(PartialEq, Eq)]
enum StartResult {
    Installed,
    Abort,
}

impl<T> Packet<T> {
    // 数据包的创建 (必须) 之后必须先调用 postinit_lock，然后再调用 Inherited_blocker
    //
    pub fn new() -> Packet<T> {
        Packet {
            queue: mpsc::Queue::new(),
            cnt: AtomicIsize::new(0),
            steals: UnsafeCell::new(0),
            to_wake: AtomicUsize::new(0),
            channels: AtomicUsize::new(2),
            port_dropped: AtomicBool::new(false),
            sender_drain: AtomicIsize::new(0),
            select_lock: Mutex::new(()),
        }
    }

    // 这个函数应该在新创建的 Packet 用 Arc 包装后使用，否则在克隆时互斥锁数据将被复制，这可能会导致在它由不透明数据结构表示的平台上出现问题
    //
    //
    //
    //
    pub fn postinit_lock(&self) -> MutexGuard<'_, ()> {
        self.select_lock.lock().unwrap()
    }

    // 在创建共享数据包时，可以使用此函数来继承以前阻塞的线程。
    // 这样做是为了防止 select() 中的线程被虚假唤醒。
    //
    // 这只能在通道创建时调用
    //
    pub fn inherit_blocker(&self, token: Option<SignalToken>, guard: MutexGuard<'_, ()>) {
        if let Some(token) = token {
            assert_eq!(self.cnt.load(Ordering::SeqCst), 0);
            assert_eq!(self.to_wake.load(Ordering::SeqCst), 0);
            self.to_wake.store(unsafe { token.cast_to_usize() }, Ordering::SeqCst);
            self.cnt.store(-1, Ordering::SeqCst);

            // 这家商店有点简陋。这里发生的是我们正在将阻止程序从 oneshot 或流通道传输到此共享通道。这样做时，我们绝不会虚假地唤醒它们，而只是在适当的时间唤醒它们。共享通道的此实现假定，一旦接收完成，任何阻塞的 recv() 都将撤消 try_recv() 中执行的窃取增量。
            //
            // 但是，我们正在继承的该线程不在 recv 的中间。因此，我们第一次唤醒它们时，它们将从原来的端口中唤醒，移至升级的端口，然后调用块 recv() 函数。
            //
            // 调用此函数时，他们会发现立即有可用数据，将其算作窃取。实际上，这并不是偷窃，因为我们适当地阻止了他们等待数据。
            //
            // 为了弥补这一不良增长，我们最初将窃取计数设置为
            // -1。您还会在 abort_selection() 中找到一些特殊代码，以确保此 -1 窃取次数不会逃逸太多。
            //
            //
            //
            //
            //
            //
            //
            //
            //
            //
            unsafe {
                *self.steals.get() = -1;
            }
        }

        // 构建共享数据包后，我们抓住了这个锁。
        // 此锁定的目的是确保 abort_selection() 不会干扰此方法。
        // 解锁此锁后，表示完成 self.cnt 和 self.to_wake 的修改，并且该端口已准备就绪，世界各地均可继续使用它。
        //
        //
        drop(guard);
    }

    pub fn send(&self, t: T) -> Result<(), T> {
        // 请参见 Port::drop，了解发生了什么情况
        if self.port_dropped.load(Ordering::SeqCst) {
            return Err(t);
        }

        // 请注意，多发送者案例在语义上要比单发送者案例稍微复杂一些。递增的逻辑为 "添加，如果断开连接，则存储断开连接"。
        // 这可能最终导致某些发送者认为，即使实际上存在断开连接，也不会断开连接。
        // 这意味着，当一个线程试图恢复断开连接状态时，其他线程可能会通过愉快地增加此非常负的断开连接计数来进行遍历。
        //
        // 为了防止发送方在实际断开通道时虚假地尝试发送，此处对计数进行了范围检查。
        //
        // 这样做也是出于另一个原因。请记住，此函数的返回值为：
        //
        //  `true` == 数据*可能*被接收，这基本上没有意义 `false` == 数据将永远不会被接收，这有很多意义
        //
        // 在 SPSC 情况下，我们检查 'queue.is_empty()' 以查看是否实际接收到数据，但是在多生产者环境中，这种相同的条件没有任何意义。
        // 结果，此预检检查将用作确定的 "永远不会收到"。
        // 一旦超出此检查范围，我们将永久进入 "这可能会收到" 的领域
        //
        //
        //
        //
        //
        //
        //
        //
        //
        //
        //
        //
        if self.cnt.load(Ordering::SeqCst) < DISCONNECTED + FUDGE {
            return Err(t);
        }

        self.queue.push(t);
        match self.cnt.fetch_add(1, Ordering::SeqCst) {
            -1 => {
                self.take_to_wake().signal();
            }

            // 在这种情况下，我们可能无法发送数据，我们需要考虑重新弹出数据以完全销毁它。
            // 但是，我们必须在多个发送者之间进行仲裁，因为我们使用的队列是单消费者队列。
            // 为了做到这一点，所有退出的推动器都将使用原子计数，以便对流经的推动器进行计数。
            // 看到 0 的推动器必须尽可能多地进入 drain，然后只有在它们是唯一推动器时才能退出 (否则必须再次尝试)。
            //
            //
            //
            //
            //
            n if n < DISCONNECTED + FUDGE => {
                // 请参见 'try' 中有关共享通道的文档，以了解 "未断开连接" 窗口正常的原因。
                //
                self.cnt.store(DISCONNECTED, Ordering::SeqCst);

                if self.sender_drain.fetch_add(1, Ordering::SeqCst) == 0 {
                    loop {
                        // drain 队列，有关线程产量的信息，请参见 try_recv 中的讨论
                        //
                        loop {
                            match self.queue.pop() {
                                mpsc::Data(..) => {}
                                mpsc::Empty => break,
                                mpsc::Inconsistent => thread::yield_now(),
                            }
                        }
                        // 也许我们已经完成了，如果不是最后一个，那么我们需要再试一次。
                        //
                        if self.sender_drain.fetch_sub(1, Ordering::SeqCst) == 1 {
                            break;
                        }
                    }

                    // 此时，队列上可能仍然有数据，但前提是计数没有增加，并且其他一些发送方尚未完成推送数据。
                    // 该发送者将 drain 自己的数据。
                    //
                    //
                }
            }

            // 不能像 SPSC 案例那样对此案例做出任何假设。
            _ => {}
        }

        Ok(())
    }

    pub fn recv(&self, deadline: Option<Instant>) -> Result<T, Failure> {
        // 该代码本质上与流情况下的代码完全相同 (请参见 stream.rs)
        //
        match self.try_recv() {
            Err(Empty) => {}
            data => return data,
        }

        let (wait_token, signal_token) = blocking::tokens();
        if self.decrement(signal_token) == Installed {
            if let Some(deadline) = deadline {
                let timed_out = !wait_token.wait_max_until(deadline);
                if timed_out {
                    self.abort_selection(false);
                }
            } else {
                wait_token.wait();
            }
        }

        match self.try_recv() {
            data @ Ok(..) => unsafe {
                *self.steals.get() -= 1;
                data
            },
            data => data,
        }
    }

    // 本质上与流递减函数完全相同。
    // 如果应继续执行阻塞，则返回 true。
    fn decrement(&self, token: SignalToken) -> StartResult {
        unsafe {
            assert_eq!(
                self.to_wake.load(Ordering::SeqCst),
                0,
                "This is a known bug in the Rust standard library. See https://github.com/rust-lang/rust/issues/39364"
            );
            let ptr = token.cast_to_usize();
            self.to_wake.store(ptr, Ordering::SeqCst);

            let steals = ptr::replace(self.steals.get(), 0);

            match self.cnt.fetch_sub(1 + steals, Ordering::SeqCst) {
                DISCONNECTED => {
                    self.cnt.store(DISCONNECTED, Ordering::SeqCst);
                }
                // 如果我们将盗窃因素考虑在内，并且注意到该通道没有数据，那么我们就可以成功入睡
                //
                n => {
                    assert!(n >= 0);
                    if n - steals <= 0 {
                        return Installed;
                    }
                }
            }

            self.to_wake.store(0, Ordering::SeqCst);
            drop(SignalToken::cast_from_usize(ptr));
            Abort
        }
    }

    pub fn try_recv(&self) -> Result<T, Failure> {
        let ret = match self.queue.pop() {
            mpsc::Data(t) => Some(t),
            mpsc::Empty => None,

            // 这是一个有趣的案例。通道被报告为有可用数据，但是由于队列处于不一致状态，我们的 pop() 失败了。
            // 这意味着某个地方还需要完成一些推动器，但是我们可以保证弹出操作最终会成功。
            // 在这种情况下，我们在一个 yield 循环中旋转，因为远程发送方应该 "非常快" 地完成他们的入队操作。
            //
            // 避免这种良率循环将需要不同的队列抽象，这可以保证在成功完成 M 次推送之后，至少 M 次弹出会成功。
            // 当前队列可确保如果有 N 次活动推送，则所有 N 完成后，您可以弹出 N 次。
            //
            //
            //
            //
            //
            //
            //
            mpsc::Inconsistent => {
                let data;
                loop {
                    thread::yield_now();
                    match self.queue.pop() {
                        mpsc::Data(t) => {
                            data = t;
                            break;
                        }
                        mpsc::Empty => panic!("inconsistent => empty"),
                        mpsc::Inconsistent => {}
                    }
                }
                Some(data)
            }
        };
        match ret {
            // 有关为什么我们可能减少盗窃的信息，请参见流实现中的讨论。
            //
            Some(data) => unsafe {
                if *self.steals.get() > MAX_STEALS {
                    match self.cnt.swap(0, Ordering::SeqCst) {
                        DISCONNECTED => {
                            self.cnt.store(DISCONNECTED, Ordering::SeqCst);
                        }
                        n => {
                            let m = cmp::min(n, *self.steals.get());
                            *self.steals.get() -= m;
                            self.bump(n - m);
                        }
                    }
                    assert!(*self.steals.get() >= 0);
                }
                *self.steals.get() += 1;
                Ok(data)
            },

            // 有关为什么再次尝试的信息，请参见流实现中的讨论。
            //
            None => {
                match self.cnt.load(Ordering::SeqCst) {
                    n if n != DISCONNECTED => Err(Empty),
                    _ => {
                        match self.queue.pop() {
                            mpsc::Data(t) => Ok(t),
                            mpsc::Empty => Err(Disconnected),
                            // 没有发送者，不一致是不可能的。
                            mpsc::Inconsistent => unreachable!(),
                        }
                    }
                }
            }
        }
    }

    // 为通道克隆准备此共享数据包，实际上只是增加引用计数。
    //
    pub fn clone_chan(&self) {
        let old_count = self.channels.fetch_add(1, Ordering::SeqCst);

        // 请参见 Arc::clone () 上的注释，了解为什么要这样做 (对于 `mem::forget`)。
        if old_count > MAX_REFCOUNT {
            abort();
        }
    }

    // 减少通道上的引用计数。
    // 每当 Chan 丢弃并可能最终唤醒接收者时，就会调用此方法。
    // 另一端，接收者的责任是确定我们已断开连接。
    pub fn drop_chan(&self) {
        match self.channels.fetch_sub(1, Ordering::SeqCst) {
            1 => {}
            n if n > 1 => return,
            n => panic!("bad number of channels left {}", n),
        }

        match self.cnt.swap(DISCONNECTED, Ordering::SeqCst) {
            -1 => {
                self.take_to_wake().signal();
            }
            DISCONNECTED => {}
            n => {
                assert!(n >= 0);
            }
        }
    }

    // 请参见 stream.rs 内部的冗长讨论，以了解为何耗尽队列以及为何以这种方式完成队列。
    //
    pub fn drop_port(&self) {
        self.port_dropped.store(true, Ordering::SeqCst);
        let mut steals = unsafe { *self.steals.get() };
        while {
            match self.cnt.compare_exchange(
                steals,
                DISCONNECTED,
                Ordering::SeqCst,
                Ordering::SeqCst,
            ) {
                Ok(_) => false,
                Err(old) => old != DISCONNECTED,
            }
        } {
            // 请参见 'try_recv' 中的讨论，以了解为什么我们要对此线程进行控制。
            //
            loop {
                match self.queue.pop() {
                    mpsc::Data(..) => {
                        steals += 1;
                    }
                    mpsc::Empty | mpsc::Inconsistent => break,
                }
            }
        }
    }

    // 占用 'to_wake' 字段的所有权。
    fn take_to_wake(&self) -> SignalToken {
        let ptr = self.to_wake.load(Ordering::SeqCst);
        self.to_wake.store(0, Ordering::SeqCst);
        assert!(ptr != 0);
        unsafe { SignalToken::cast_from_usize(ptr) }
    }

    ////////////////////////////////////////////////////////////////////////////
    // 选择实现
    ////////////////////////////////////////////////////////////////////////////

    // 增加通道上的计数 (用于选择)
    fn bump(&self, amt: isize) -> isize {
        match self.cnt.fetch_add(amt, Ordering::SeqCst) {
            DISCONNECTED => {
                self.cnt.store(DISCONNECTED, Ordering::SeqCst);
                DISCONNECTED
            }
            n => n,
        }
    }

    // 取消先前在该端口上等待的线程，并返回该端口上是否有数据。
    //
    // 这与流实现类似 (因此，注释较少)，但对 "steals" 变量使用不同的值。
    //
    //
    pub fn abort_selection(&self, _was_upgrade: bool) -> bool {
        // 在我们做其他事情之前，我们先弹开这个锁。
        // 这样做的原因是为了确保进行中的所有升级都不会进行。
        // 没有这种反弹，我们就可以与 Inherited_blocker 竞争有关查看和处理 to_wake 的问题。
        //
        // 一旦获得了锁，我们就可以保证 Inherited_blocker 已完成。
        {
            let _guard = self.select_lock.lock().unwrap();
        }

        // 像流实现一样，我们要确保通道上的计数变为非负数。
        // 我们不知道当前流的负值如何，因此，我们不使用窃取值 1，而是加载通道计数并弄清楚应该如何使它变为正值。
        //
        //
        //
        let steals = {
            let cnt = self.cnt.load(Ordering::SeqCst);
            if cnt < 0 && cnt != DISCONNECTED { -cnt } else { 0 }
        };
        let prev = self.bump(steals + 1);

        if prev == DISCONNECTED {
            assert_eq!(self.to_wake.load(Ordering::SeqCst), 0);
            true
        } else {
            let cur = prev + steals + 1;
            assert!(cur >= 0);
            if prev < 0 {
                drop(self.take_to_wake());
            } else {
                while self.to_wake.load(Ordering::SeqCst) != 0 {
                    thread::yield_now();
                }
            }
            unsafe {
                // 如果窃取次数为 -1，则它是从我们继承阻止程序起的先发制人的 -1 窃取次数。
                // 这很好，因为我们将使用实际值覆盖它。
                //
                let old = self.steals.get();
                assert!(*old == 0 || *old == -1);
                *old = steals;
                prev >= 0
            }
        }
    }
}

impl<T> Drop for Packet<T> {
    fn drop(&mut self) {
        // 请注意，此负载不仅是断开连接正确性的断言，而且还是读取 `to_wake` 之前的适当栅栏，因此无法在删除 `to_wake` 断言的同时删除此断言。
        //
        //
        //
        assert_eq!(self.cnt.load(Ordering::SeqCst), DISCONNECTED);
        assert_eq!(self.to_wake.load(Ordering::SeqCst), 0);
        assert_eq!(self.channels.load(Ordering::SeqCst), 0);
    }
}