Tange

A Task-based parallelization framework.

What is it?

Tange is a framework that makes it easy to write defered, data parallel computations that are executed concurrently across a local machine. It can scale up to millions of tasks per Graph and can be useful for a number of different applications:

• Data processing.
• All-Reduce operations.
• Distributed machine learning algorithms.
• General parallel computing.

How to Use It?

Tange defines a Deferred struct which represents a computation. Deferred objects are accessed with three simple functions:

1. lift - Lift takes a concrete value and lifts it into a Deferred object
2. apply - Apply applies a function to a Deferred, producing a new Deferred object.
3. join - Join combines two Deferred objects with a joiner function, producing a new Deferred.

Example - Hello World

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

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

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

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

asserteq!(helloworld.run(&GreedyScheduler::new()), Some("Hello World!".into())); ```

Example

Let's count all the words across a directory.

```rust extern crate tange;

use tange::scheduler::GreedyScheduler; use tange::deferred::{Deferred,batchapply,treereduce};

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

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

fn readfiles(dir: &Path, buffer: &mut Vec>) -> io::Result<()> { if dir.isdir() { for entry in readdir(dir)? { let entry = entry?; let path = entry.path(); if path.isdir() { readfiles(&path, buffer)?; } else { let p = path.tostringlossy().intoowned(); 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);

} ```