rust - 在 Rust 中跨 worker 分配工作

Question

use std::iter;

fn worker_sum(from: u64, to: u64) -> u64 {
    range(from, to).fold(0u64, |sum, x| sum + x)
}

fn main() {
    let max = 5u64;
    let step = 2u64;

    let (sender, receiver) = channel::<u64>();
    for x in iter::range_step_inclusive(0u64, max, step) {
        let end = if x + step > max { max } else { x + step };
        //println!("{} -> {} = {}", x, end, worker_sum(x, end));
        let local_sender = sender.clone();
        spawn(proc(){
            local_sender.send(worker_sum(x, end));
        });
    }
    loop {
        match receiver.try_recv() {
            Ok(x) => println!("{}", x),
            Err(_) => break,
        }
    }
}

我得到错误:

task '' failed at 'sending on a closed channel', /home/rustbuild/src/rust-buildbot/slave/nightly-linux/build/src/libsync/comm/mod.rs:573

我以某种方式理解了这个问题，但是如何从 channel 中正确地“选择”？文档真的很少，即使我使用的是夜间构建，据说它改进了文档(自版本 0.13 起)。

所以我的问题是:

1. 如何在代码结构改动尽可能少的情况下解决问题？
2. 如何使代码地道？

Answer 1

这里的问题是在发送所有数据之前 channel 被读取任务关闭。你的循环是:

loop {
    match receiver.try_recv() {
        Ok(x) => println!("{}", x),
        Err(_) => break,
    }
}

在此循环中，您的接收器在遇到错误时立即中断。一旦循环中断，您的函数将到达其范围的末尾并且接收器将被销毁。完成此操作后，任何发送更多数据的尝试都将失败。

这里的问题是，您的接收方得到一个Err(Empty)，因为发送方还没有发送任何东西。您必须等待它们并且只有在遇到 Err(Disconnected)

时才会中断

您需要将代码更改为如下内容(注释中的解释):

use std::iter;

fn worker_sum(from: u64, to: u64) -> u64 {
    range(from, to).fold(0u64, |sum, x| sum + x)
}

fn main() {
    let max = 5u64;
    let step = 2u64;

    let (sender, receiver) = channel::<u64>();

    for x in iter::range_step_inclusive(0u64, max, step) {
        let end = if x + step > max { max } else { x + step };
        // here, each thread will own its own sender, and the channel will
        // be closed once all senders are destroyed.
        let local_sender = sender.clone();
        spawn(proc(){
            local_sender.send(worker_sum(x, end));
            // Once we reach here, the sender of this task is destroyed.
        });
    }

    // We destroy the sender of the main task,
    // because we don't want to wait for it:
    // it would deadlock the program
    drop(sender);

    loop {
        match receiver.try_recv() {
            Ok(x) => println!("{}", x),
            // We break only if the channel is closed,
            // it means that all senders are finished.
            Err(e) if e == ::std::comm::Disconnected => { break; },
            _ => {}
        }
    }
}