及时数据流

及时数据流是一种低延迟的循环数据流计算模型，由论文 Naiad: a timely dataflow system 提出。本项目是及时数据流在 Rust 中的扩展和更模块化的实现。

本项目类似于一个分布式数据并行计算引擎，可以将相同的程序从您笔记本电脑上的单个线程扩展到跨计算机集群的分布式执行。主要目标是表达能力和高性能。如果您还没有使用及时数据流，那么它可能比您目前使用的任何东西都更加表达性强和运行速度快。

请务必阅读 及时数据流的文档。它仍在进行中，但大部分都在改进。在 mdbook 格式的 长文本 中有更多内容，包括针对当前构建测试的示例。还有一个介绍及时数据流的博客文章系列（第 1 部分，第 2 部分，第 3 部分），尽管警告，那里的示例可能需要调整才能与当前代码兼容。

示例

要使用及时数据流，请将以下内容添加到项目 Cargo.toml 文件中的依赖项部分

[dependencies]
timely="*"

这将从 crates.io 引入 timely crate，这应该允许您开始编写类似这样的及时数据流程序（也在 timely/examples/simple.rs 中提供）

extern crate timely;

use timely::dataflow::operators::*;

fn main() {
    timely::example(|scope| {
        (0..10).to_stream(scope)
               .inspect(|x| println!("seen: {:?}", x));
    });
}

您可以通过在 timely-dataflow 仓库的根目录中键入以下内容来运行此示例

% cargo run --example simple
Running `target/debug/examples/simple`
seen: 0
seen: 1
seen: 2
seen: 3
seen: 4
seen: 5
seen: 6
seen: 7
seen: 8
seen: 9

这是一个非常简单的示例（这已经在名称中说明了），它只是表明您可以如何编写数据流程序。

做更多的事情

对于更复杂的示例，可以考虑非常相似但更明确的 examples/hello.rs，它分别创建和驱动数据流

extern crate timely;

use timely::dataflow::{InputHandle, ProbeHandle};
use timely::dataflow::operators::{Input, Exchange, Inspect, Probe};

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {

        let index = worker.index();
        let mut input = InputHandle::new();
        let mut probe = ProbeHandle::new();

        // create a new input, exchange data, and inspect its output
        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                 .exchange(|x| *x)
                 .inspect(move |x| println!("worker {}:\thello {}", index, x))
                 .probe_with(&mut probe);
        });

        // introduce data and watch!
        for round in 0..10 {
            if index == 0 {
                input.send(round);
            }
            input.advance_to(round + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }
    }).unwrap();
}

此示例做了很多，以展示及时可以为您做些什么。

我们首先构建一个数据流图，创建一个输入流（使用 input_from），然后将输出交换（使用 exchange）来驱动记录在工人之间（使用数据本身来指示要路由到哪个工人）。我们 inspect 数据并打印工人索引，以指示哪个工人接收了哪些数据，然后 probe 结果，以便每个工人都可以看到给定一轮数据是否已经处理完毕。

然后，通过反复引入数据轮次来驱动计算，其中 round 本身用作数据。在每一轮中，每个工人引入相同的数据，然后反复执行数据流步骤，直到 probe 显示所有工人已经处理了该时代的所有工作，此时计算继续。

对于两个工人，输出看起来像

% cargo run --example hello -- -w2
Running `target/debug/examples/hello -w2`
worker 0:   hello 0
worker 1:   hello 1
worker 0:   hello 2
worker 1:   hello 3
worker 0:   hello 4
worker 1:   hello 5
worker 0:   hello 6
worker 1:   hello 7
worker 0:   hello 8
worker 1:   hello 9

请注意，尽管工人类 0 引入了数据 (0..10)，但每个元素都路由到特定的工人，正如我们预期的。

% cat hostfile.txt
host0:port
host1:port
host2:port
host3:port
...

host0% cargo run -- -w 2 -h hostfile.txt -n 4 -p 0
host1% cargo run -- -w 2 -h hostfile.txt -n 4 -p 1
host2% cargo run -- -w 2 -h hostfile.txt -n 4 -p 2
host3% cargo run -- -w 2 -h hostfile.txt -n 4 -p 3