异步（五）

上一讲的代码都相对比较简单，这一讲将会有极大的突破，代码会复杂很多，也算进入了 Future 最关键的地方——如何正确手动实现自己想要的 Future。

channel 是什么

我们得从调度单元开始讲起，更准确地说，异步的基本单元是 Future 这个 task，而同步，它的基本单元是线程，这又回到了第二讲时候提出的绿色线程的概念。channel 它在同步环境中是用来跨线程通信用的，而在异步环境中，它是用来跨 Future（task）通信用的，合并起来，channel 是跨调度单元的通信工具（同理其实可以推理到 lock/semaphore）。

接着，channel 和 queue 的核心区别：是否存在唤醒（调度单元的）机制。channel 实际上是由 queue + 唤醒机制组合而成的。

在同步环境中，唤醒机制是

在接收数据时能唤醒 sender 端被阻塞住的线程
在发送数据时能唤醒 receiver 端因无数据消费而陷入等待状态的线程

在异步环境中，唤醒机制是：

在接收数据时能唤醒 sender 端陷入 pending 状态的 Future
在发送数据时能唤醒 receiver 端因无数据消费而陷入 pending 状态的 Future

不能混用的原因也很简单，因为如果同步 channel 在异步环境中将线程 block 住了，就意味着这个线程的异步执行器被卡死了，这样的操作会导致异步环境陷入异常状态，我们之前说过，异步需要让线程一直工作，并频繁将执行到 pending 状态的 task（Future）切换到后台继续执行可以执行的 task（Future）。

mpsc channel

我们直接跳过了 oneshot channel，因为它是一个一次性操作，在塞数据的时候，不需要考虑如果塞不进去怎么办的问题，它可以简单认为塞不进去就是 channel 被关闭了。

ok，我们现在开始考虑怎么构建一个异步版本的 mpsc channel，需要一些什么东西：

一个 lock free 的 Queue 实现
记录 channel 状态的 field
唤醒 receiver 开始工作的 waker
唤醒 sender 继续工作的 waker list（unbound channel 不需要）

unbound

先从简单的 unbound channel 开始，unbound 意味着 sender 端不需要考虑数据满的问题，也就省略了唤醒 sender 因无法塞数据而导致的 pending 状态。

struct Inner<T> {
    // High level marker channel status
	// The lower bit expresses the number of items in the Queue
    state: AtomicUsize,
    message_queue: Queue<T>,
    recv_task: AtomicWaker,
}

发送的时候：

fn send(&self, msg: T) -> Result {
    let mut curr = self.state.load(SeqCst)
    let state = decode(curr);
    loop {
        if state.is_open {
            return Err
        }

        state.num + 1
        let next = encode(state);
        // CAS
        match self.state.compare_exchange(curr, next, SeqCst, SeqCst) {
            Ok(_) => break,
            Err(actual) => curr = actual,
        }
    }
    self.queue.push(msg);
    self.recv_task.wake();
    Ok(())
}

接收的时候：

fn next_message(&self) -> Poll<Option<T>> {
    match self.queue.pop() {
        Some(msg) => {
            // same as sender add num with cas
            self.dec_num_on_state();
            Poll::Ready(Some(msg))
        }
        None => {
            if self.state.is_open {
                Poll::Read(None)
            } else {
                Poll::Pending
            }
        }
    }
}

impl<T> Stream for Inner<T> {
    type Item = T;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // Try to read a message off of the message queue.
        match self.next_message() {
            Poll::Ready(msg) => {
                Poll::Ready(msg)
            }
            Poll::Pending => {
                // There are no messages to read, in this case, park.
                self.recv_task.register(cx.waker());
                // Check queue again after parking to prevent race condition:
                // a message could be added to the queue after previous `next_message`
                // before `register` call.
                self.next_message()
            }
        }
    }
}

这里引入了几个东西：

AtomicXX 与 CAS 过程，这两个都是与 lock free 相关的并发元语，用来替代锁的实现。
Stream trait：流式任务，Future 是一次性任务，Stream 是多次任务

在上述代码中，我们可以看到：

在发送端，如果数据塞入成功，调用 recv_task 去唤醒 receiver 开始工作
recv_task 的注册是由 receiver 端发现自己处于 pending 状态时塞入的

这就是最基本的自唤醒式 Future 了，在不考虑清理的难度下，上述结构就可以正常工作了，但 mpsc channel 需要有正常的 drop 过程，还需要记录一下 sender 的数量，这就是另外的话题了，略过不表。

bound

相对与 unbound channel 来说，bound channel 更复杂：

需要根据 Item 的数量给 Sender 端发 Pending 状态
需要双向唤醒，并且还需要指定唤醒对象

相应的，代码需要做一些调整：

struct SenderTask {
    task: Option<Waker>,
    is_parked: bool,
}

struct Inner<T> {
    buffer: usize,
    state: AtomicUsize,
    message_queue: Queue<T>,
    parked_queue: Queue<Arc<Mutex<SenderTask>>>,
    recv_task: AtomicWaker,
}

struct Sender<T> {
    inner: Arc<Inner<T>>,
    sender_task: Arc<Mutex<SenderTask>>,
    maybe_parked: AtomicBool,
}