https://zhuanlan.zhihu.com/p/50101525
本文分析Rust标准库中的channel，channel(通道)作为线程间通信的一种方式被广泛使用。
Rust提供了多生产者单消费者的channel。我们重点关注多个生产者的情况。
它的实现方式非常有趣。我把它分为通道升级跟并发队列两部分。
本文描述通道升级
对于一个channel()调用，我们得到的(sender, receiver)是oneshot的，这一点从源码可以得到暗示：
#[stable(feature = "rust1", since = "1.0.0")]
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
let a = Arc::new(oneshot::Packet::new());
(Sender::new(Flavor::Oneshot(a.clone())), Receiver::new(Flavor::Oneshot(a)))
}这里至少有四个结构：
- oneshot::Packet：Packet，真正存放数据的地方。此处是单个数据(其他类型可能使用队列)
- Flavor::Oneshot。
- Sender/Receiver。
我们分别看下他们的数据结构源码，首先是oneshot::Packet，它位于mpsc/oneshot.rs：
pub struct Packet<T> {
// Internal state of the chan/port pair (stores the blocked thread as well)
state: AtomicUsize,
// One-shot data slot location
data: UnsafeCell<Option<T>>,
// when used for the second time, a oneshot channel must be upgraded, and
// this contains the slot for the upgrade
upgrade: UnsafeCell<MyUpgrade<T>>,
}可以看出data是为一个数据准备的。upgrade字段用于通道升级。
另外还有其他类型的Packet，查看同一文件夹发现有shared::Packet/stream::Packet/sync::Packet，他们分别位于shared.rs/stream.rs/sync.rs中。我们重点关注shared::Packet：
pub struct Packet<T> {
queue: mpsc::Queue<T>,
cnt: AtomicIsize, // How many items are on this channel
steals: UnsafeCell<isize>, // How many times has a port received without blocking?
to_wake: AtomicUsize, // SignalToken for wake up
// The number of channels which are currently using this packet.
channels: AtomicUsize,
// See the discussion in Port::drop and the channel send methods for what
// these are used for
port_dropped: AtomicBool,
sender_drain: AtomicIsize,
// this lock protects various portions of this implementation during
// select()
select_lock: Mutex<()>,
}清楚地看到queue字段，它用于存放数据。我们先不关注数据字段。
对于这四个类型的Packet，标准库提供了enun Flavor<T>来做区分：
enum Flavor<T> {
Oneshot(Arc<oneshot::Packet<T>>),
Stream(Arc<stream::Packet<T>>),
Shared(Arc<shared::Packet<T>>),
Sync(Arc<sync::Packet<T>>),
}而我们的Sender/Receiver对象则非常简单地通过存储Flavor<T>来关联到Packet：
pub struct Sender<T> {
inner: UnsafeCell<Flavor<T>>,
}
pub struct Receiver<T> {
inner: UnsafeCell<Flavor<T>>,
}我们再看一下fn channel：
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
let a = Arc::new(oneshot::Packet::new());
(Sender::new(Flavor::Oneshot(a.clone())), Receiver::new(Flavor::Oneshot(a)))
}就可以了解到Sender/Receiver里面都存了Flavor，根据Flavor的类型区分Packet的类型，同时Packet作为共享数据被安全地共享。
这就是我们调用channel得到的结果。因为我们重点关注多生产者的情况，所以我们再看一下Clone for Sender的实现：
impl<T> Clone for Sender<T> {
fn clone(&self) -> Sender<T> {
let packet = match *unsafe { self.inner() } {
Flavor::Oneshot(ref p) => {
let a = Arc::new(shared::Packet::new());
{
let guard = a.postinit_lock();
let rx = Receiver::new(Flavor::Shared(a.clone()));
let sleeper = match p.upgrade(rx) {
oneshot::UpSuccess |
oneshot::UpDisconnected => None,
oneshot::UpWoke(task) => Some(task),
};
a.inherit_blocker(sleeper, guard);
}
a
}
Flavor::Stream(ref p) => {
let a = Arc::new(shared::Packet::new());
{
let guard = a.postinit_lock();
let rx = Receiver::new(Flavor::Shared(a.clone