唐吉诃德

一个基于任务并行化的框架。

它是什么？

Tange 是一个框架，使得编写在本地机器上并发执行的数据并行计算变得简单。它可以扩展到每个图数百万个任务，可用于许多不同的应用。

• 数据处理。
• 所有-Reduce 操作。
• 分布式机器学习算法。
• 通用并行计算。

如何使用？

Tange 定义了一个 Deferred 结构，它表示一个计算。使用三个简单的函数来访问 Deferred 对象

1. lift - Lift 将一个具体值提升到 Deferred 对象
2. apply - Apply 将函数应用于 Deferred，生成一个新的 Deferred 对象。
3. join - Join 使用连接函数将两个 Deferred 对象合并，生成一个新的 Deferred。

示例 - Hello World

use tange::deferred::Deferred;
use tange::scheduler::GreedyScheduler;

// Create two Deferred object
let hello = Deferred::lift("Hello".to_owned(), None);
let world = Deferred::lift("World".to_owned(), None);

// Add an exclamation mark to "World"
let world_exclaim = world.apply(|w| format!("{}!", w));

// Join the words!
let hello_world = hello.join(&world_exclaim, |h, w| format!("{} {}", h, w));

assert_eq!(hello_world.run(&GreedyScheduler::new()), Some("Hello World!".into()));

示例

让我们计算一个目录中的所有单词。

extern crate tange;

use tange::scheduler::GreedyScheduler;
use tange::deferred::{Deferred,batch_apply,tree_reduce};

use std::io::{BufReader,BufRead};
use std::env::args;

use std::io;
use std::fs::{File, read_dir};
use std::path::Path;

fn read_files(dir: &Path, buffer: &mut Vec<Deferred<String>>) -> io::Result<()> {
    if dir.is_dir() {
        for entry in read_dir(dir)? {
            let entry = entry?;
            let path = entry.path();
            if path.is_dir() {
                read_files(&path, buffer)?;
            } else {
                let p = path.to_string_lossy().into_owned();
                buffer.push(Deferred::lift(p, None));
            }
        }
    }
    Ok(())
}

fn main() {
    let mut defs = Vec::new();
    for path in args().skip(1) {
        read_files(&Path::new(&path), &mut defs).expect("Error reading directory!");
    }

    if defs.len() == 0 {
        panic!("No files to count!");
    }

    // Read a file and count the number of words, split by white space
    let counts = batch_apply(&defs, |_idx, fname| {
        let mut count = 0usize;
        if let Ok(f) = File::open(&fname) {
            let mut br = BufReader::new(f);
            for maybe_line in br.lines() {
                if let Ok(line) = maybe_line {
                    for p in line.split_whitespace() {
                        if p.len() > 0 {
                            count += 1;
                        }
                    }
                } else {
                    eprintln!("Error reading {}, skipping rest of file...", fname);
                    break
                }
            }
        };
        count
    });

    // Sum the counts
    let total = tree_reduce(&counts, |left, right| left + right)
        .expect("Can't reduce if there are no files in the directory!");

    let count = total.run(&GreedyScheduler::new()).unwrap();
    println!("Found {} words", count);
}