Function execute_from_args 

pub fn execute_from_args<I, T, F>(
    iter: I,
    func: F,
) -> Result<WorkerGuards<T>, String>
where
    I: Iterator<Item = String>,
    T: Send + 'static,
    F: Fn(&mut Worker<Allocator>) -> T + Send + Sync + 'static,

Executes a timely dataflow from supplied arguments and per-communicator logic.

The execute method takes arguments (typically std::env::args()) and spins up some number of workers threads, each of which execute the supplied closure to construct and run a timely dataflow computation.

The closure may return a T: Send+'static. The execute_from_args method returns immediately after initializing the timely computation with a result containing a WorkerGuards<T> (or error information), which can be joined to recover the result T values from the local workers.

Note: if the caller drops the result of execute_from_args, the drop code will block awaiting the completion of the timely computation.

The arguments execute_from_args currently understands are:

-w, --threads: number of per-process worker threads.

-n, --processes: number of processes involved in the computation.

-p, --process: identity of this process; from 0 to n-1.

-h, --hostfile: a text file whose lines are “hostname:port” in order of process identity. If not specified, localhost will be used, with port numbers increasing from 2101 (chosen arbitrarily).

This method is only available if the getopts feature is enabled, which it is by default.

Examples

use timely::dataflow::operators::{ToStream, Inspect};

// execute a timely dataflow using command line parameters
timely::execute_from_args(std::env::args(), |worker| {
    worker.dataflow::<(),_,_>(|scope| {
        (0..10).to_stream(scope)
               .container::<Vec<_>>()
               .inspect(|x| println!("seen: {:?}", x));
    })
}).unwrap();

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

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

Examples found in repository

examples/capture_send.rs (lines 6-17)

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {

        let addr = format!("127.0.0.1:{}", 8000 + worker.index());
        let send = TcpStream::connect(addr).unwrap();

        worker.dataflow::<u64,_,_>(|scope|
            (0..10u64)
                .to_stream(scope)
                .container::<Vec<_>>()
                .capture_into(EventWriter::new(send))
        );
    }).unwrap();
}

examples/pingpong.rs (lines 9-23)

fn main() {

    let iterations = std::env::args().nth(1).unwrap().parse::<u64>().unwrap();
    let elements = std::env::args().nth(2).unwrap().parse::<u64>().unwrap();

    // initializes and runs a timely dataflow
    timely::execute_from_args(std::env::args().skip(3), move |worker| {
        let index = worker.index();
        let peers = worker.peers();
        worker.dataflow::<u64,_,_>(move |scope| {
            let (helper, cycle) = scope.feedback(1);
            (0 .. elements)
                  .filter(move |&x| (x as usize) % peers == index)
                  .to_stream(scope)
                  .concat(cycle)
                  .exchange(|&x| x)
                  .map_in_place(|x| *x += 1)
                  .branch_when(move |t| t < &iterations).1
                  .connect_loop(helper);
        });
    }).unwrap();
}

examples/capture_recv.rs (lines 6-26)

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {

        let source_peers = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();

        // create replayers from disjoint partition of source worker identifiers.
        let replayers =
        (0 .. source_peers)
            .filter(|i| i % worker.peers() == worker.index())
            .map(|i| TcpListener::bind(format!("127.0.0.1:{}", 8000 + i)).unwrap())
            .collect::<Vec<_>>()
            .into_iter()
            .map(|l| l.incoming().next().unwrap().unwrap())
            .map(EventReader::<_,Vec<u64>,_>::new)
            .collect::<Vec<_>>();

        worker.dataflow::<u64,_,_>(|scope| {
            replayers
                .replay_into(scope)
                .inspect(|x| println!("replayed: {:?}", x));
        })
    }).unwrap(); // asserts error-free execution
}

examples/rc.rs (lines 12-31)

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {
        // create a new input, exchange data, and inspect its output
        let index = worker.index();
        let mut input = InputHandle::new();
        let probe = ProbeHandle::new();
        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                 .container::<Vec<_>>()
                 //.exchange(|x| *x) // <-- cannot exchange this; Rc is not Send.
                 .inspect(move |x| println!("worker {}:\thello {:?}", index, x))
                 .probe_with(&probe);
        });

        // introduce data and watch!
        for round in 0..10 {
            input.send(Test { _field: Rc::new(round) } );
            input.advance_to(round + 1);
            worker.step_while(|| probe.less_than(input.time()));
        }
    }).unwrap();
}

examples/hello.rs (lines 6-31)

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 probe = ProbeHandle::new();

        // create a new input, exchange data, and inspect its output
        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                 .container::<Vec<_>>()
                 .exchange(|x| *x)
                 .inspect(move |x| println!("worker {}:\thello {}", index, x))
                 .probe_with(&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();
}

examples/logging-recv.rs (lines 9-29)

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {

        let source_peers = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();

        // create replayers from disjoint partition of source worker identifiers.
        let replayers =
        (0 .. source_peers)
            .filter(|i| i % worker.peers() == worker.index())
            .map(|i| TcpListener::bind(format!("127.0.0.1:{}", 8000 + i)).unwrap())
            .collect::<Vec<_>>()
            .into_iter()
            .map(|l| l.incoming().next().unwrap().unwrap())
            .map(EventReader::<Duration,Vec<(Duration,TimelySetup,TimelyEvent)>,_>::new)
            .collect::<Vec<_>>();

        worker.dataflow(|scope| {
            replayers
                .replay_into(scope)
                .inspect(|x| println!("replayed: {:?}", x));
        })
    }).unwrap(); // asserts error-free execution
}

Additional examples can be found in:

• examples/barrier.rs
• examples/flow_controlled.rs
• examples/exchange.rs
• examples/unionfind.rs
• examples/event_driven.rs
• examples/distinct.rs
• examples/wordcount.rs
• examples/hashjoin.rs
• examples/loopdemo.rs
• examples/logging-send.rs
• examples/openloop.rs
• examples/bfs.rs
• examples/pagerank.rs