如何使用 mpsc 通道在线程之间创建环形通信?

Question

我想生成 n 个能够与环形拓扑中的其他线程通信的线程，例如线程 0 可以将消息发送到线程 1，线程 1 发送到线程 2，依此类推，线程 n 发送到线程 0。

这是我想要在 n=3 的情况下实现的目标的一个例子：

use std::sync::mpsc::{self, Receiver, Sender};
use std::thread;
let (tx0, rx0): (Sender<i32>, Receiver<i32>) = mpsc::channel();
let (tx1, rx1): (Sender<i32>, Receiver<i32>) = mpsc::channel();
let (tx2, rx2): (Sender<i32>, Receiver<i32>) = mpsc::channel();
let child0 = thread::spawn(move || {
tx0.send(0).unwrap();
println!("thread 0 sent: 0");
println!("thread 0 recv: {:?}", rx2.recv().unwrap());
});
let child1 = thread::spawn(move || {
tx1.send(1).unwrap();
println!("thread 1 sent: 1");
println!("thread 1 recv: {:?}", rx0.recv().unwrap());
});
let child2 = thread::spawn(move || {
tx2.send(2).unwrap();
println!("thread 2 sent: 2");
println!("thread 2 recv: {:?}", rx1.recv().unwrap());
});
child0.join();
child1.join();
child2.join();

在这里，我在循环中创建通道，将它们存储在向量中，对发送方重新排序，将它们存储在新的向量中，然后生成每个线程都有自己的发送方-接收方(tx1/rx0、tx2/rx1 等(对。

const NTHREADS: usize = 8;
// create n channels
let channels: Vec<(Sender<i32>, Receiver<i32>)> =
(0..NTHREADS).into_iter().map(|_| mpsc::channel()).collect();
// switch tupel entries for the senders to create ring topology
let mut channels_ring: Vec<(Sender<i32>, Receiver<i32>)> = (0..NTHREADS)
.into_iter()
.map(|i| {
(
channels[if i < channels.len() - 1 { i + 1 } else { 0 }].0,
channels[i].1,
)
})
.collect();
let mut children = Vec::new();
for i in 0..NTHREADS {
let (tx, rx) = channels_ring.remove(i);
let child = thread::spawn(move || {
tx.send(i).unwrap();
println!("thread {} sent: {}", i, i);
println!("thread {} recv: {:?}", i, rx.recv().unwrap());
});
children.push(child);
}
for child in children {
let _ = child.join();
}

这不起作用，因为无法复制发送方以创建新向量。 但是，如果我使用 refs(和发件人(：

let mut channels_ring: Vec<(&Sender<i32>, Receiver<i32>)> = (0..NTHREADS)
.into_iter()
.map(|i| {
(
&channels[if i < channels.len() - 1 { i + 1 } else { 0 }].0,
channels[i].1,
)
})
.collect();

我无法生成线程，因为std::sync::mpsc::Sender<i32>线程之间无法安全地共享。

Answer 1

Sender

s和Receivers 无法共享，因此您需要将它们移动到各自的线程中。这意味着将它们从Vec中删除，或者在迭代时消耗Vec- 不允许向量处于无效状态(有孔(，即使是作为中间步骤。使用into_iter迭代向量将通过使用它们来实现这一点。

你可以用一个小技巧让发送方和接收方在一个循环中配对，就是创建两个向量;一个是发送方，一个是接收方;然后旋转一个，以便每个向量的相同索引将给你你想要的对。

use std::sync::mpsc::{self, Receiver, Sender};
use std::thread;
fn main() {
const NTHREADS: usize = 8;
// create n channels
let (mut senders, receivers): (Vec<Sender<i32>>, Vec<Receiver<i32>>) =
(0..NTHREADS).into_iter().map(|_| mpsc::channel()).unzip();
// move the first sender to the back
senders.rotate_left(1);
let children: Vec<_> = senders
.into_iter()
.zip(receivers.into_iter())
.enumerate()
.map(|(i, (tx, rx))| {
thread::spawn(move || {
tx.send(i as i32).unwrap();
println!("thread {} sent: {}", i, i);
println!("thread {} recv: {:?}", i, rx.recv().unwrap());
})
})
.collect();
for child in children {
let _ = child.join();
}
}

Answer 2

这不起作用，因为无法复制 Sender 来创建新向量。但是，如果我使用 refs(和发件人(：

虽然Sender确实无法复制，但它确实实现了Clone，因此您始终可以手动克隆它。但是这种方法不适用于Receiver，这不是Clone，您还需要从向量中提取。

第一个代码的问题在于，不能使用let foo = vec[i]从非Copy值向量中仅移动一个值。这将使向量处于无效状态，其中一个元素无效，随后访问该元素将导致未定义的行为。要做到这一点，Vec需要跟踪哪些元素被移动，哪些没有移动，这将给所有Vec带来成本。因此，Vec不允许将元素移出，而是让用户跟踪移动。

将值移出Vec的一种简单方法是将Vec<T>替换为Vec<Option<T>>并使用Option::take。foo = vec[i]被替换为foo = vec[i].take().unwrap()，这会将T值从vec[i]中的选项中移出(同时断言它不是None(，并在向量中保留None，Option<T>的有效变体。这是您第一次尝试以这种方式修改(游乐场(：

const NTHREADS: usize = 8;
let channels_ring: Vec<_> = {
let mut channels: Vec<_> = (0..NTHREADS)
.into_iter()
.map(|_| {
let (tx, rx) = mpsc::channel();
(Some(tx), Some(rx))
})
.collect();
(0..NTHREADS)
.into_iter()
.map(|rxpos| {
let txpos = if rxpos < NTHREADS - 1 { rxpos + 1 } else { 0 };
(
channels[txpos].0.take().unwrap(),
channels[rxpos].1.take().unwrap(),
)
})
.collect()
};
let children: Vec<_> = channels_ring
.into_iter()
.enumerate()
.map(|(i, (tx, rx))| {
thread::spawn(move || {
tx.send(i as i32).unwrap();
println!("thread {} sent: {}", i, i);
println!("thread {} recv: {:?}", i, rx.recv().unwrap());
})
})
.collect();
for child in children {
child.join().unwrap();
}