Timely Dataflow

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

本项目类似于一个分布式数据并行计算引擎，可以将相同的程序从笔记本电脑上的单个线程扩展到集群的分布式执行。主要目标是表达能力和高性能。在没有使用 Timely Dataflow 的情况下，它可能比您目前使用的任何东西都更加表达丰富且速度更快。

请务必阅读 Timely Dataflow 文档。它仍在进行中，但大部分都在改进。还有更多 长篇文本 以 mdbook 格式提供，其中包含针对当前构建测试的示例。还有一系列博客文章（第 1 部分，第 2 部分，第 3 部分），以不同的方式介绍 Timely Dataflow，但请注意，那里的示例可能需要调整才能与当前代码一起构建。

示例

要使用 Timely Dataflow，请在项目的 Cargo.toml 文件的依赖项部分添加以下内容

[dependencies]
timely="*"

这将从 crates.io 添加 timely 包，这应该允许您开始编写像这样（也在 timely/examples/simple.rs 中可用）的 Timely Dataflow 程序。

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();
}

此示例做了很多工作，以展示timely能为您做更多的事情。

我们首先构建一个数据流图，创建一个输入流（使用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..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

• 如果你想用及时数据流编写程序，这对我们来说非常有趣。理想情况下，及时数据流旨在以一种符合人体工程学的非平凡数据流计算方法。随着人们使用它并报告他们的体验，我们了解了他们发现的类别、人体工程学的痛点以及其他我们事先甚至没有想象到的事情。了解及时数据流、尝试使用它并报告回来是有帮助的！

• 如果你喜欢编写小的示例程序或文档测试，及时数据流中有很多地方示例相对稀疏，或者实际上并没有测试演示的功能。这些通常很容易上手、完善并推进，而无需大量的前期义务。这可能也是一个很好的方法，让我们其中之一详细地解释一些内容，如果你需要的话。

• 如果你喜欢在及时数据流中亲自动手，问题跟踪器 有各种问题，涉及堆栈的不同级别。例如

  • 及时目前 比必须的多复制数据，这是为了直接满足 Rust 的所有权纪律。其中一些复制可以通过在资源管理中更加注意（例如，使用一个 Vec<u8> 的共享区域，就像 bytes crate 所做的那样）来省略。这里不是所有的事情都那么明显，因此这里也有设计工作的机会。

  • 我们最近实施了一系列日志更改，但仍有一些 期望的功能 还没有实现。如果你对通过探索记录其行为的底层基础设施来了解及时的工作方式感兴趣，这可能是一个不错的选择！它有一个额外的优势，即日志本身就是及时流，因此你甚至可以在及时上做一些日志处理。哇...

  • 关于 将 Rust 所有权习惯融入到及时数据流中 的一个开放问题。目前，及时流是可复制的对象，当一个流被重新使用时，项将被复制。我们可以使其更明确，并要求调用 cloned() 方法以获得所有权的对象，就像迭代器需要它一样。同时，不获取所有权而使用流引用应该给你一个机会查看经过的记录（而不需要复制，就像目前所做的那样）。对于可能需要序列化数据但不能充分利用所有权的交换通道来说，这通常已经足够了。

  • 在调度及时数据流算子方面，有很多有趣的研究工作。当我们有机会调度许多算子时，我们可能会稍微思考一下，意识到其中一些算子没有任何工作要做，可以跳过。更好的是，我们可以维护一个有工作要做算子的列表，对于没有工作要做的算子，则不进行任何操作。