Блог

У меня есть многопоточный код, который предполагает использование вложенного цикла, где внутренний выполняется параллельно. «Общий доступ» в каждом потоке Sender -это функция, которая будет возвращать результаты. В частности Sender , реализует Send , так что не должно быть никаких проблем с его клонированием и отправкой с использованием <url for_each_with() > . Однако компиляция этого кода:

 use std::sync::mpsc::channel;
use rayon::prelude::*;

fn main(){
    let (sender, receiver) = channel();

    (0..5).for_each(|i|{
        (0..5).into_par_iter().for_each_with(amp;sender, |sender, j|{
            sender.send(i   j).unwrap();   
        });
    });
}
 

Дает мне:

 8 |         (0..5).into_par_iter().for_each_with(amp;sender, |sender, j|{
  |                                              ^^^^^^^ `Sender<_>` cannot be shared between threads safely
  |
  = help: the trait `Sync` is not implemented for `Sender<_>`
  = note: required because of the requirements on the impl of `Send` for `amp;Sender<_>`
 

(Детская площадка)

Теперь я подумал, что это может быть связано с тем, что я пытаюсь клонировать ссылку, но если я перенесу фактическое sender в for_each_with() (т. Е. for_each_with(sender, ...) ), оно будет использовано первой итерацией внешнего цикла:

 error[E0507]: cannot move out of `sender`, a captured variable in an `FnMut` closure
 

(Детская площадка).

Как я могу реализовать этот шаблон таким образом, чтобы он удовлетворял компилятору Rust?

AFAIK, район использует пул потоков, это означает, что району требуется реализовать элемент Send , потому что он отправит его в поток. вискоза клонируется не для каждого элемента, а для каждой нити:

 fn for_each_with<OP, T>(self, init: T, op: OP) where
    OP: Fn(amp;mut T, Self::Item)   Sync   Send,
    T: Send   Clone,
 

Мы видим OP , что взятие а amp;mut T не а T . Это означает for_each_with() , что клон для каждого числа потоков, используемых не для количества продуктов изделия.

Ссылку нужно внедрить Sync , чтобы внедрить Send . Sender это определить, чтобы не реализовывать Sync . Я все равно не знаю подробностей этого выбора, это означает amp;Sender , что не может быть разделено между потоками. Я не думаю, что есть решение для устранения этого ограничения.

Но если вы хотите, вы можете использовать поперечную балку-канал, который реализует Sync :

 use crossbeam_channel::unbounded; // 0.5.1
use rayon::prelude::*; // 1.5.1

fn main() {
    let (sender, _receiver) = unbounded();

    for i in 0..5 {
        (0..5).into_par_iter().for_each_with(amp;sender, |sender, j| {
            sender.send(i   j).unwrap();
        });
    }
}
 

было бы хорошо скомпилировать. Бонус в том, что поперечный канал утверждает, что он быстрее. Тем не менее, клонирование полностью подходит для ЗППП Sender :

 use rayon::prelude::*;
use std::sync::mpsc::channel;

fn main() {
    let (sender, _receiver) = channel();

    for i in 0..5 {
        let sender = sender.clone();
        (0..5).into_par_iter().for_each_with(sender, |sender, j| {
            sender.send(i   j).unwrap();
        });
    }
}
 

Это действительно O(n) увеличило бы время клонирования, но Sender из-за зппп клонировать много значит. (На самом деле это, вероятно, просто добавить один клон, потому что вы решили сделать вложенный цикл, код, вероятно, не клонирует последний, а просто дает ему последний поток, и поэтому вы просто слишком сильно клонируете его, насколько мы можем проверить)

В любом случае, все ваши проблемы возникают из-за странной ситуации, следует сгладить итерацию, например:

 use rayon::prelude::*;
use std::sync::mpsc::channel;

fn main() {
    let (sender, _recever) = channel();

    (0..5)
        .into_par_iter()
        .flat_map_iter(|i| (0..5).map(move |j| (i, j)))
        .for_each_with(sender, |sender, (i, j)| {
            sender.send(i   j).unwrap();
        });
}
 

Вы также могли бы подумать о том, чтобы не использовать Sender вискозу, вероятно, для использования, collect() и вместо отправки товара вы могли бы просто забрать их в конце.

Sender сделано для клонирования, это буквально в документах. У каждого дочернего потока должен быть свой собственный отправитель.