Struct Handle 

pub struct Handle<T: Timestamp, CB: ContainerBuilder<Container: Clone>> { /* private fields */ }

A handle to an input Stream, used to introduce data to a timely dataflow computation.

Implementations

impl<T: Timestamp, C: Container + Clone> Handle<T, CapacityContainerBuilder<C>>

pub fn new() -> Self

Allocates a new input handle, from which one can create timely streams.

Examples

use timely::*;
use timely::dataflow::InputHandle;
use timely::dataflow::operators::{Input, Inspect};

// construct and execute a timely dataflow
timely::execute(Config::thread(), |worker| {

    // add an input and base computation off of it
    let mut input = InputHandle::new();
    worker.dataflow(|scope| {
        scope.input_from(&mut input)
             .container::<Vec<_>>()
             .inspect(|x| println!("hello {:?}", x));
    });

    // introduce input, advance computation
    for round in 0..10 {
        input.send(round);
        input.advance_to(round + 1);
        worker.step();
    }
});

Examples found in repository

examples/threadless.rs (line 12)

fn main() {

    // create a naked single-threaded worker.
    let allocator = timely::communication::allocator::Thread::default();
    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator, None);

    // create input and probe handles.
    let mut input = InputHandle::new();
    let probe = ProbeHandle::new();

    // directly build a dataflow.
    worker.dataflow(|scope| {
        input
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe_with(&probe);
    });

    // manage inputs.
    for i in 0 .. 10 {
        input.send(i);
        input.advance_to(i);
        while probe.less_than(input.time()) {
            worker.step();
        }
    }
}

examples/rc.rs (line 15)

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 (line 9)

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/exchange.rs (line 10)

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

        let batch = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();
        let rounds = std::env::args().nth(2).unwrap().parse::<usize>().unwrap();
        let mut input = InputHandle::new();

        // create a new input, exchange data, and inspect its output
        let probe = worker.dataflow(|scope|
            scope
                .input_from(&mut input)
                .container::<Vec<_>>()
                .exchange(|&x| x as u64)
                .probe()
                .0
        );


        let timer = std::time::Instant::now();

        for round in 0 .. rounds {

            for i in 0 .. batch {
                input.send(i);
            }
            input.advance_to(round);

            while probe.less_than(input.time()) {
                worker.step();
            }

        }

        let volume = (rounds * batch) as f64;
        let elapsed = timer.elapsed();
        let seconds = elapsed.as_secs() as f64 + (f64::from(elapsed.subsec_nanos())/1000000000.0);

        println!("{:?}\tworker {} complete; rate: {:?}", timer.elapsed(), worker.index(), volume / seconds);

    }).unwrap();
}

examples/unionfind.rs (line 20)

fn main() {

    // command-line args: numbers of nodes and edges in the random graph.
    let nodes: usize = std::env::args().nth(1).unwrap().parse().unwrap();
    let edges: usize = std::env::args().nth(2).unwrap().parse().unwrap();
    let batch: usize = std::env::args().nth(3).unwrap().parse().unwrap();

    timely::execute_from_args(std::env::args().skip(4), move |worker| {

        let index = worker.index();
        let peers = worker.peers();

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

        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                //  .exchange(move |x: &(usize, usize)| (x.0 % (peers - 1)) as u64 + 1)
                 .union_find()
                 .exchange(|_| 0)
                 .union_find()
                 .probe_with(&probe);
        });

        // Generate roughly random data.
        use std::hash::{BuildHasher, BuildHasherDefault, DefaultHasher};
        let hasher = BuildHasherDefault::<DefaultHasher>::new();
        let insert = (0..).map(move |i| (hasher.hash_one(&(i,index,0)) as usize % nodes,
                                         hasher.hash_one(&(i,index,1)) as usize % nodes));

        for (edge, arc) in insert.take(edges / peers).enumerate() {
            input.send(arc);
            if edge % batch == (batch - 1) {
                let next = input.epoch() + 1;
                input.advance_to(next);
                while probe.less_than(input.time()) {
                    worker.step();
                }
            }
        }

    }).unwrap(); // asserts error-free execution;
}

examples/distinct.rs (line 12)

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::<usize,_,_>(|scope| {
            let mut counts_by_time = HashMap::new();
            scope.input_from(&mut input)
                .unary(Exchange::new(|x| *x), "Distinct", move |_, _|
                    move |input, output| {
                        input.for_each_time(|time, data| {
                            let counts =
                            counts_by_time
                                .entry(*time.time())
                                .or_insert(HashMap::new());
                            let mut session = output.session(&time);
                            for data in data {
                                for &datum in data.iter() {
                                    let count = counts.entry(datum).or_insert(0);
                                    if *count == 0 {
                                        session.give(datum);
                                    }
                                    *count += 1;
                                }
                            }
                        })
                    })
                .container::<Vec<_>>()
                .inspect(move |x| println!("worker {}:\tvalue {}", index, x))
                .probe_with(&probe);
        });

        // introduce data and watch!
        for round in 0..1 {
            if index == 0 {
                [0, 1, 2, 2, 2, 3, 3, 4].iter().for_each(|x| input.send(*x));
            } else if index == 1 {
                [0, 0, 3, 4, 4, 5, 7, 7].iter().for_each(|x| input.send(*x));
            }
            input.advance_to(round + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }
    }).unwrap();
}

Additional examples can be found in:

• examples/wordcount.rs
• examples/hashjoin.rs
• examples/loopdemo.rs
• examples/logging-send.rs
• examples/openloop.rs
• examples/columnar.rs
• examples/pagerank.rs

impl<T: Timestamp, CB: ContainerBuilder<Container: Clone>> Handle<T, CB>

pub fn new_with_builder() -> Self

Allocates a new input handle, from which one can create timely streams.

Examples

use timely::*;
use timely::dataflow::InputHandle;
use timely::dataflow::operators::{Input, Inspect};
use timely_container::CapacityContainerBuilder;

// construct and execute a timely dataflow
timely::execute(Config::thread(), |worker| {

    // add an input and base computation off of it
    let mut input = InputHandle::<_, CapacityContainerBuilder<_>>::new_with_builder();
    worker.dataflow(|scope| {
        scope.input_from(&mut input)
             .container::<Vec<_>>()
             .inspect(|x| println!("hello {:?}", x));
    });

    // introduce input, advance computation
    for round in 0..10 {
        input.send(round);
        input.advance_to(round + 1);
        worker.step();
    }
});

pub fn to_stream<G>(&mut self, scope: &mut G) -> Stream<G, CB::Container>

where T: TotalOrder, G: Scope<Timestamp = T>,

Creates an input stream from the handle in the supplied scope.

Examples

use timely::*;
use timely::dataflow::InputHandle;
use timely::dataflow::operators::{Input, Inspect};

// construct and execute a timely dataflow
timely::execute(Config::thread(), |worker| {

    // add an input and base computation off of it
    let mut input = InputHandle::new();
    worker.dataflow(|scope| {
        input.to_stream(scope)
             .container::<Vec<_>>()
             .inspect(|x| println!("hello {:?}", x));
    });

    // introduce input, advance computation
    for round in 0..10 {
        input.send(round);
        input.advance_to(round + 1);
        worker.step();
    }
});

Examples found in repository

examples/threadless.rs (line 18)

fn main() {

    // create a naked single-threaded worker.
    let allocator = timely::communication::allocator::Thread::default();
    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator, None);

    // create input and probe handles.
    let mut input = InputHandle::new();
    let probe = ProbeHandle::new();

    // directly build a dataflow.
    worker.dataflow(|scope| {
        input
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe_with(&probe);
    });

    // manage inputs.
    for i in 0 .. 10 {
        input.send(i);
        input.advance_to(i);
        while probe.less_than(input.time()) {
            worker.step();
        }
    }
}

examples/wordcount.rs (line 20)

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

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

        // define a distribution function for strings.
        let exchange = Exchange::new(|x: &(String, i64)| (x.0).len() as u64);

        // create a new input, exchange data, and inspect its output
        worker.dataflow::<usize,_,_>(|scope| {
            input.to_stream(scope)
                 .flat_map(|(text, diff): (String, i64)|
                    text.split_whitespace()
                        .map(move |word| (word.to_owned(), diff))
                        .collect::<Vec<_>>()
                 )
                 .unary_frontier(exchange, "WordCount", |_capability, _info| {

                    let mut queues = HashMap::new();
                    let mut counts = HashMap::new();

                    move |(input, frontier), output| {
                        input.for_each_time(|time, data| {
                            queues.entry(time.retain(output.output_index()))
                                  .or_insert(Vec::new())
                                  .extend(data.map(std::mem::take));
                        });

                        for (key, val) in queues.iter_mut() {
                            if !frontier.less_equal(key.time()) {
                                let mut session = output.session(key);
                                for mut batch in val.drain(..) {
                                    for (word, diff) in batch.drain(..) {
                                        let entry = counts.entry(word.clone()).or_insert(0i64);
                                        *entry += diff;
                                        session.give((word, *entry));
                                    }
                                }
                            }
                        }

                        queues.retain(|_key, val| !val.is_empty());
                    }})
                 .container::<Vec<_>>()
                 .inspect(|x| println!("seen: {:?}", x))
                 .probe_with(&probe);
        });

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

examples/columnar.rs (line 41)

fn main() {

    type InnerContainer = <WordCount as columnar::Columnar>::Container;
    type Container = Column<InnerContainer>;

    use columnar::Len;

    let config = timely::Config {
        communication: timely::CommunicationConfig::ProcessBinary(3),
        worker: timely::WorkerConfig::default(),
    };

    // initializes and runs a timely dataflow.
    timely::execute(config, |worker| {
        let mut input = <InputHandle<_, CapacityContainerBuilder<Container>>>::new();
        let probe = ProbeHandle::new();

        // create a new input, exchange data, and inspect its output
        worker.dataflow::<usize, _, _>(|scope| {
            input
                .to_stream(scope)
                .unary(
                    Pipeline,
                    "Split",
                    |_cap, _info| {
                        move |input, output| {
                            input.for_each_time(|time, data| {
                                let mut session = output.session(&time);
                                for data in data {
                                    for wordcount in data.borrow().into_index_iter().flat_map(|wordcount| {
                                        wordcount.text.split_whitespace().map(move |text| WordCountReference { text, diff: wordcount.diff })
                                    }) {
                                        session.give(wordcount);
                                    }
                                }
                            });
                        }
                    },
                )
                .container::<Container>()
                .unary_frontier(
                    ExchangeCore::<ColumnBuilder<InnerContainer>,_>::new_core(|x: &WordCountReference<&str,&i64>| x.text.len() as u64),
                    "WordCount",
                    |_capability, _info| {
                        let mut queues = HashMap::new();
                        let mut counts = HashMap::new();

                        move |(input, frontier), output| {
                            input.for_each_time(|time, data| {
                                queues
                                    .entry(time.retain(output.output_index()))
                                    .or_insert(Vec::new())
                                    .extend(data.map(std::mem::take));

                            });

                            for (key, val) in queues.iter_mut() {
                                if !frontier.less_equal(key.time()) {
                                    let mut session = output.session(key);
                                    for batch in val.drain(..) {
                                        for wordcount in batch.borrow().into_index_iter() {
                                            let total =
                                            if let Some(count) = counts.get_mut(wordcount.text) {
                                                *count += wordcount.diff;
                                                *count
                                            }
                                            else {
                                                counts.insert(wordcount.text.to_string(), *wordcount.diff);
                                                *wordcount.diff
                                            };
                                            session.give(WordCountReference { text: wordcount.text, diff: total });
                                        }
                                    }
                                }
                            }

                            queues.retain(|_key, val| !val.is_empty());
                        }
                    },
                )
                .container::<Container>()
                .inspect_container(|x| {
                    match x {
                        Ok((time, data)) => {
                            println!("seen at: {:?}\t{:?} records", time, data.record_count());
                            for wc in data.borrow().into_index_iter() {
                                println!("  {}: {}", wc.text, wc.diff);
                            }
                        },
                        Err(frontier) => println!("frontier advanced to {:?}", frontier),
                    }
                })
                .probe_with(&probe);
        });

        // introduce data and watch!
        for round in 0..10 {
            input.send(WordCountReference { text: "flat container", diff: 1 });
            input.advance_to(round + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }
    })
    .unwrap();
}

examples/pagerank.rs (line 18)

fn main() {

    timely::execute_from_args(std::env::args().skip(3), move |worker| {

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

        worker.dataflow::<usize,_,_>(|scope| {

            // create a new input, into which we can push edge changes.
            let edge_stream = input.to_stream(scope);

            // create a new feedback stream, which will be changes to ranks.
            let (handle, rank_stream) = scope.feedback(1);

            // bring edges and ranks together!
            let changes = edge_stream.binary_frontier(
                rank_stream,
                Exchange::new(|x: &((usize, usize), i64)| (x.0).0 as u64),
                Exchange::new(|x: &(usize, i64)| x.0 as u64),
                "PageRank",
                |_capability, _info| {

                    // where we stash out-of-order data.
                    let mut edge_stash: HashMap<_, Vec<_>> = HashMap::new();
                    let mut rank_stash: HashMap<_, Vec<_>> = HashMap::new();

                    // lists of edges, ranks, and changes.
                    let mut edges = Vec::new();
                    let mut ranks = Vec::new();
                    let mut diffs = Vec::new(); // for received but un-acted upon deltas.
                    let mut delta = Vec::new();

                    let timer = ::std::time::Instant::now();

                    move |(input1, frontier1), (input2, frontier2), output| {

                        // hold on to edge changes until it is time.
                        input1.for_each_time(|time, data| {
                            let entry = edge_stash.entry(time.retain(output.output_index())).or_default();
                            data.for_each(|data| entry.append(data));
                        });

                        // hold on to rank changes until it is time.
                        input2.for_each_time(|time, data| {
                            let entry = rank_stash.entry(time.retain(output.output_index())).or_default();
                            data.for_each(|data| entry.append(data));
                        });

                        let frontiers = &[frontier1, frontier2];

                        for (time, edge_changes) in edge_stash.iter_mut() {
                            if frontiers.iter().all(|f| !f.less_equal(time)) {

                                let mut session = output.session(time);

                                compact(edge_changes);

                                for ((src, dst), diff) in edge_changes.drain(..) {

                                    // 0. ensure enough state allocated
                                    while edges.len() <= src { edges.push(Vec::new()); }
                                    while ranks.len() <= src { ranks.push(1_000); }
                                    while diffs.len() <= src { diffs.push(0); }

                                    // 1. subtract previous distribution.
                                    allocate(ranks[src], &edges[src][..], &mut delta);
                                    for x in delta.iter_mut() { x.1 *= -1; }

                                    // 2. update edges.
                                    edges[src].push((dst, diff));
                                    compact(&mut edges[src]);

                                    // 3. re-distribute allocations.
                                    allocate(ranks[src], &edges[src][..], &mut delta);

                                    // 4. compact down and send cumulative changes.
                                    compact(&mut delta);
                                    for (dst, diff) in delta.drain(..) {
                                        session.give((dst, diff));
                                    }
                                }
                            }
                        }

                        edge_stash.retain(|_key, val| !val.is_empty());

                        for (time, rank_changes) in rank_stash.iter_mut() {
                            if frontiers.iter().all(|f| !f.less_equal(time)) {

                                let mut session = output.session(time);

                                compact(rank_changes);

                                let mut cnt = 0;
                                let mut sum = 0;
                                let mut max = 0;

                                for (src, diff) in rank_changes.drain(..) {

                                    cnt += 1;
                                    sum += diff.abs();
                                    max = if max < diff.abs() { diff.abs() } else { max };

                                    // 0. ensure enough state allocated
                                    while edges.len() <= src { edges.push(Vec::new()); }
                                    while ranks.len() <= src { ranks.push(1_000); }
                                    while diffs.len() <= src { diffs.push(0); }

                                    // 1. subtract previous distribution.
                                    allocate(ranks[src], &edges[src][..], &mut delta);
                                    for x in delta.iter_mut() { x.1 *= -1; }

                                    // 2. update ranks.
                                    diffs[src] += diff;
                                    if diffs[src].abs() >= 6 {
                                        ranks[src] += diffs[src];
                                        diffs[src] = 0;
                                    }

                                    // 3. re-distribute allocations.
                                    allocate(ranks[src], &edges[src][..], &mut delta);

                                    // 4. compact down and send cumulative changes.
                                    compact(&mut delta);
                                    for (dst, diff) in delta.drain(..) {
                                        session.give((dst, diff));
                                    }
                                }

                                println!("{:?}:\t{:?}\t{}\t{}\t{}", timer.elapsed(), time.time(), cnt, sum, max);
                            }
                        }

                        rank_stash.retain(|_key, val| !val.is_empty());

                    }
                }
            );

            changes
                .probe_with(&probe)
                .connect_loop(handle);

        });

        let nodes: usize = std::env::args().nth(1).unwrap().parse().unwrap();
        let edges: usize = std::env::args().nth(2).unwrap().parse().unwrap();

        let index = worker.index();
        // Generate roughly random data.
        use std::hash::{BuildHasher, BuildHasherDefault, DefaultHasher};
        let hasher = BuildHasherDefault::<DefaultHasher>::new();
        let mut insert = (0..).map(move |i| (hasher.hash_one(&(i,index,0)) as usize % nodes,
                                             hasher.hash_one(&(i,index,1)) as usize % nodes));
        let remove = insert.clone();

        for ins in (&mut insert).take(edges / worker.peers()) {
            input.send((ins, 1));
        }

        input.advance_to(1);

        while probe.less_than(input.time()) {
            worker.step();
        }

        for (ins, del) in insert.zip(remove).take(1000) {
            input.send((ins, 1));
            input.send((del,-1));
            input.advance_to(input.time() + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }

    }).unwrap(); // asserts error-free execution;
}

pub fn flush(&mut self)

Flush all contents and distribute to downstream operators.

pub fn send_batch(&mut self, buffer: &mut CB::Container)

Sends a batch of records into the corresponding timely dataflow Stream, at the current epoch.

This method flushes single elements previously sent with send, to keep the insertion order.

Examples

use timely::*;
use timely::dataflow::InputHandle;
use timely::dataflow::operators::{Input, InspectCore};

// construct and execute a timely dataflow
timely::execute(Config::thread(), |worker| {

    // add an input and base computation off of it
    let mut input = InputHandle::new();
    worker.dataflow(|scope| {
        scope.input_from(&mut input)
             .inspect_container(|x| println!("hello {:?}", x));
    });

    // introduce input, advance computation
    for round in 0..10 {
        input.send_batch(&mut vec![format!("{}", round)]);
        input.advance_to(round + 1);
        worker.step();
    }
});

pub fn advance_to(&mut self, next: T)

Advances the current epoch to next.

This method allows timely dataflow to issue progress notifications as it can now determine that this input can no longer produce data at earlier timestamps.

Examples found in repository

examples/threadless.rs (line 27)

fn main() {

    // create a naked single-threaded worker.
    let allocator = timely::communication::allocator::Thread::default();
    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator, None);

    // create input and probe handles.
    let mut input = InputHandle::new();
    let probe = ProbeHandle::new();

    // directly build a dataflow.
    worker.dataflow(|scope| {
        input
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe_with(&probe);
    });

    // manage inputs.
    for i in 0 .. 10 {
        input.send(i);
        input.advance_to(i);
        while probe.less_than(input.time()) {
            worker.step();
        }
    }
}

examples/rc.rs (line 28)

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 (line 26)

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/exchange.rs (line 30)

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

        let batch = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();
        let rounds = std::env::args().nth(2).unwrap().parse::<usize>().unwrap();
        let mut input = InputHandle::new();

        // create a new input, exchange data, and inspect its output
        let probe = worker.dataflow(|scope|
            scope
                .input_from(&mut input)
                .container::<Vec<_>>()
                .exchange(|&x| x as u64)
                .probe()
                .0
        );


        let timer = std::time::Instant::now();

        for round in 0 .. rounds {

            for i in 0 .. batch {
                input.send(i);
            }
            input.advance_to(round);

            while probe.less_than(input.time()) {
                worker.step();
            }

        }

        let volume = (rounds * batch) as f64;
        let elapsed = timer.elapsed();
        let seconds = elapsed.as_secs() as f64 + (f64::from(elapsed.subsec_nanos())/1000000000.0);

        println!("{:?}\tworker {} complete; rate: {:?}", timer.elapsed(), worker.index(), volume / seconds);

    }).unwrap();
}

examples/event_driven.rs (line 40)

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

        let timer = std::time::Instant::now();

        let mut args = std::env::args();
        args.next();

        let dataflows = args.next().unwrap().parse::<usize>().unwrap();
        let length = args.next().unwrap().parse::<usize>().unwrap();
        let record = args.next() == Some("record".to_string());

        let mut inputs = Vec::new();
        let mut probes = Vec::new();

        // create a new input, exchange data, and inspect its output
        for _dataflow in 0 .. dataflows {
            worker.dataflow(|scope| {
                let (input, mut stream) = scope.new_input();
                for _step in 0 .. length {
                    stream = stream.map(|x: ()| x);
                }
                let (probe, _stream) = stream.probe();
                inputs.push(input);
                probes.push(probe);
            });
        }

        println!("{:?}\tdataflows built ({} x {})", timer.elapsed(), dataflows, length);

        for round in 0 .. {
            let dataflow = round % dataflows;
            if record {
                inputs[dataflow].send(());
            }
            inputs[dataflow].advance_to(round);
            let mut steps = 0;
            while probes[dataflow].less_than(&round) {
                worker.step();
                steps += 1;
            }
            println!("{:?}\tround {} complete in {} steps", timer.elapsed(), round, steps);
        }

    }).unwrap();
}

examples/unionfind.rs (line 42)

fn main() {

    // command-line args: numbers of nodes and edges in the random graph.
    let nodes: usize = std::env::args().nth(1).unwrap().parse().unwrap();
    let edges: usize = std::env::args().nth(2).unwrap().parse().unwrap();
    let batch: usize = std::env::args().nth(3).unwrap().parse().unwrap();

    timely::execute_from_args(std::env::args().skip(4), move |worker| {

        let index = worker.index();
        let peers = worker.peers();

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

        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                //  .exchange(move |x: &(usize, usize)| (x.0 % (peers - 1)) as u64 + 1)
                 .union_find()
                 .exchange(|_| 0)
                 .union_find()
                 .probe_with(&probe);
        });

        // Generate roughly random data.
        use std::hash::{BuildHasher, BuildHasherDefault, DefaultHasher};
        let hasher = BuildHasherDefault::<DefaultHasher>::new();
        let insert = (0..).map(move |i| (hasher.hash_one(&(i,index,0)) as usize % nodes,
                                         hasher.hash_one(&(i,index,1)) as usize % nodes));

        for (edge, arc) in insert.take(edges / peers).enumerate() {
            input.send(arc);
            if edge % batch == (batch - 1) {
                let next = input.epoch() + 1;
                input.advance_to(next);
                while probe.less_than(input.time()) {
                    worker.step();
                }
            }
        }

    }).unwrap(); // asserts error-free execution;
}

Additional examples can be found in:

• examples/distinct.rs
• examples/wordcount.rs
• examples/hashjoin.rs
• examples/loopdemo.rs
• examples/logging-send.rs
• examples/openloop.rs
• examples/columnar.rs
• examples/pagerank.rs

pub fn close(self)

Closes the input.

This method allows timely dataflow to issue all progress notifications blocked by this input and to begin to shut down operators, as this input can no longer produce data.

pub fn epoch(&self) -> &T

Reports the current epoch.

Examples found in repository

examples/unionfind.rs (line 41)

fn main() {

    // command-line args: numbers of nodes and edges in the random graph.
    let nodes: usize = std::env::args().nth(1).unwrap().parse().unwrap();
    let edges: usize = std::env::args().nth(2).unwrap().parse().unwrap();
    let batch: usize = std::env::args().nth(3).unwrap().parse().unwrap();

    timely::execute_from_args(std::env::args().skip(4), move |worker| {

        let index = worker.index();
        let peers = worker.peers();

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

        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                //  .exchange(move |x: &(usize, usize)| (x.0 % (peers - 1)) as u64 + 1)
                 .union_find()
                 .exchange(|_| 0)
                 .union_find()
                 .probe_with(&probe);
        });

        // Generate roughly random data.
        use std::hash::{BuildHasher, BuildHasherDefault, DefaultHasher};
        let hasher = BuildHasherDefault::<DefaultHasher>::new();
        let insert = (0..).map(move |i| (hasher.hash_one(&(i,index,0)) as usize % nodes,
                                         hasher.hash_one(&(i,index,1)) as usize % nodes));

        for (edge, arc) in insert.take(edges / peers).enumerate() {
            input.send(arc);
            if edge % batch == (batch - 1) {
                let next = input.epoch() + 1;
                input.advance_to(next);
                while probe.less_than(input.time()) {
                    worker.step();
                }
            }
        }

    }).unwrap(); // asserts error-free execution;
}

examples/hashjoin.rs (line 95)

fn main() {

    // command-line args: numbers of nodes and edges in the random graph.
    let keys: u64 = std::env::args().nth(1).unwrap().parse().unwrap();
    let vals: usize = std::env::args().nth(2).unwrap().parse().unwrap();
    let batch: usize = std::env::args().nth(3).unwrap().parse().unwrap();

    timely::execute_from_args(std::env::args().skip(4), move |worker| {

        let index = worker.index();
        let peers = worker.peers();

        let mut input1 = InputHandle::new();
        let mut input2 = InputHandle::new();
        let probe = ProbeHandle::new();

        worker.dataflow(|scope| {

            let stream1 = scope.input_from(&mut input1);
            let stream2 = scope.input_from(&mut input2);

            let exchange1 = Exchange::new(|x: &(u64, u64)| x.0);
            let exchange2 = Exchange::new(|x: &(u64, u64)| x.0);

            stream1
                .binary(stream2, exchange1, exchange2, "HashJoin", |_capability, _info| {

                    let mut map1 = HashMap::<u64, Vec<u64>>::new();
                    let mut map2 = HashMap::<u64, Vec<u64>>::new();

                    move |input1, input2, output| {

                        // Drain first input, check second map, update first map.
                        input1.for_each_time(|time, data| {
                            let mut session = output.session(&time);
                            for (key, val1) in data.flat_map(|d| d.drain(..)) {
                                if let Some(values) = map2.get(&key) {
                                    for val2 in values.iter() {
                                        session.give((val1, *val2));
                                    }
                                }

                                map1.entry(key).or_default().push(val1);
                            }
                        });

                        // Drain second input, check first map, update second map.
                        input2.for_each_time(|time, data| {
                            let mut session = output.session(&time);
                            for (key, val2) in data.flat_map(|d| d.drain(..)) {
                                if let Some(values) = map1.get(&key) {
                                    for val1 in values.iter() {
                                        session.give((*val1, val2));
                                    }
                                }

                                map2.entry(key).or_default().push(val2);
                            }
                        });
                    }
                })
                .container::<Vec<_>>()
                .probe_with(&probe);
        });

        // Generate roughly random data.
        use std::hash::{BuildHasher, BuildHasherDefault, DefaultHasher};
        let hasher = BuildHasherDefault::<DefaultHasher>::new();
        let mut insert = (0..).map(move |i| (hasher.hash_one(&(i,index,0)) % keys,
                                             hasher.hash_one(&(i,index,1)) % keys,
                                             hasher.hash_one(&(i,index,2)) % keys,
                                             hasher.hash_one(&(i,index,3)) % keys));

        let timer = std::time::Instant::now();

        let mut sent = 0;
        while sent < (vals / peers) {

            // Send some amount of data, no more than `batch`.
            let to_send = std::cmp::min(batch, vals/peers - sent);
            for (src0, dst0, src1, dst1) in (&mut insert).take(to_send) {
                input1.send((src0, dst0));
                input2.send((src1, dst1));
            }
            sent += to_send;

            // Advance input, iterate until data cleared.
            let next = input1.epoch() + 1;
            input1.advance_to(next);
            input2.advance_to(next);
            while probe.less_than(input1.time()) {
                worker.step();
            }

            println!("{:?}\tworker {} batch complete", timer.elapsed(), index)
        }

    }).unwrap(); // asserts error-free execution;
}

pub fn time(&self) -> &T

Reports the current timestamp.

Examples found in repository

examples/threadless.rs (line 28)

fn main() {

    // create a naked single-threaded worker.
    let allocator = timely::communication::allocator::Thread::default();
    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator, None);

    // create input and probe handles.
    let mut input = InputHandle::new();
    let probe = ProbeHandle::new();

    // directly build a dataflow.
    worker.dataflow(|scope| {
        input
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe_with(&probe);
    });

    // manage inputs.
    for i in 0 .. 10 {
        input.send(i);
        input.advance_to(i);
        while probe.less_than(input.time()) {
            worker.step();
        }
    }
}

examples/rc.rs (line 29)

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 (line 27)

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/exchange.rs (line 32)

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

        let batch = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();
        let rounds = std::env::args().nth(2).unwrap().parse::<usize>().unwrap();
        let mut input = InputHandle::new();

        // create a new input, exchange data, and inspect its output
        let probe = worker.dataflow(|scope|
            scope
                .input_from(&mut input)
                .container::<Vec<_>>()
                .exchange(|&x| x as u64)
                .probe()
                .0
        );


        let timer = std::time::Instant::now();

        for round in 0 .. rounds {

            for i in 0 .. batch {
                input.send(i);
            }
            input.advance_to(round);

            while probe.less_than(input.time()) {
                worker.step();
            }

        }

        let volume = (rounds * batch) as f64;
        let elapsed = timer.elapsed();
        let seconds = elapsed.as_secs() as f64 + (f64::from(elapsed.subsec_nanos())/1000000000.0);

        println!("{:?}\tworker {} complete; rate: {:?}", timer.elapsed(), worker.index(), volume / seconds);

    }).unwrap();
}

examples/unionfind.rs (line 43)

fn main() {

    // command-line args: numbers of nodes and edges in the random graph.
    let nodes: usize = std::env::args().nth(1).unwrap().parse().unwrap();
    let edges: usize = std::env::args().nth(2).unwrap().parse().unwrap();
    let batch: usize = std::env::args().nth(3).unwrap().parse().unwrap();

    timely::execute_from_args(std::env::args().skip(4), move |worker| {

        let index = worker.index();
        let peers = worker.peers();

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

        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                //  .exchange(move |x: &(usize, usize)| (x.0 % (peers - 1)) as u64 + 1)
                 .union_find()
                 .exchange(|_| 0)
                 .union_find()
                 .probe_with(&probe);
        });

        // Generate roughly random data.
        use std::hash::{BuildHasher, BuildHasherDefault, DefaultHasher};
        let hasher = BuildHasherDefault::<DefaultHasher>::new();
        let insert = (0..).map(move |i| (hasher.hash_one(&(i,index,0)) as usize % nodes,
                                         hasher.hash_one(&(i,index,1)) as usize % nodes));

        for (edge, arc) in insert.take(edges / peers).enumerate() {
            input.send(arc);
            if edge % batch == (batch - 1) {
                let next = input.epoch() + 1;
                input.advance_to(next);
                while probe.less_than(input.time()) {
                    worker.step();
                }
            }
        }

    }).unwrap(); // asserts error-free execution;
}

examples/distinct.rs (line 51)

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::<usize,_,_>(|scope| {
            let mut counts_by_time = HashMap::new();
            scope.input_from(&mut input)
                .unary(Exchange::new(|x| *x), "Distinct", move |_, _|
                    move |input, output| {
                        input.for_each_time(|time, data| {
                            let counts =
                            counts_by_time
                                .entry(*time.time())
                                .or_insert(HashMap::new());
                            let mut session = output.session(&time);
                            for data in data {
                                for &datum in data.iter() {
                                    let count = counts.entry(datum).or_insert(0);
                                    if *count == 0 {
                                        session.give(datum);
                                    }
                                    *count += 1;
                                }
                            }
                        })
                    })
                .container::<Vec<_>>()
                .inspect(move |x| println!("worker {}:\tvalue {}", index, x))
                .probe_with(&probe);
        });

        // introduce data and watch!
        for round in 0..1 {
            if index == 0 {
                [0, 1, 2, 2, 2, 3, 3, 4].iter().for_each(|x| input.send(*x));
            } else if index == 1 {
                [0, 0, 3, 4, 4, 5, 7, 7].iter().for_each(|x| input.send(*x));
            }
            input.advance_to(round + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }
    }).unwrap();
}

Additional examples can be found in:

• examples/wordcount.rs
• examples/hashjoin.rs
• examples/logging-send.rs
• examples/columnar.rs
• examples/pagerank.rs

impl<T: Timestamp, CB: ContainerBuilder<Container: Clone>> Handle<T, CB>

pub fn send<D>(&mut self, data: D)

where CB: PushInto<D>,

Sends one record into the corresponding timely dataflow Stream, at the current epoch.

Examples

use timely::*;
use timely::dataflow::InputHandle;
use timely::dataflow::operators::{Input, Inspect};

// construct and execute a timely dataflow
timely::execute(Config::thread(), |worker| {

    // add an input and base computation off of it
    let mut input = InputHandle::new();
    worker.dataflow(|scope| {
        scope.input_from(&mut input)
             .container::<Vec<_>>()
             .inspect(|x| println!("hello {:?}", x));
    });

    // introduce input, advance computation
    for round in 0..10 {
        input.send(round);
        input.advance_to(round + 1);
        worker.step();
    }
});

Examples found in repository

examples/threadless.rs (line 26)

fn main() {

    // create a naked single-threaded worker.
    let allocator = timely::communication::allocator::Thread::default();
    let mut worker = timely::worker::Worker::new(WorkerConfig::default(), allocator, None);

    // create input and probe handles.
    let mut input = InputHandle::new();
    let probe = ProbeHandle::new();

    // directly build a dataflow.
    worker.dataflow(|scope| {
        input
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe_with(&probe);
    });

    // manage inputs.
    for i in 0 .. 10 {
        input.send(i);
        input.advance_to(i);
        while probe.less_than(input.time()) {
            worker.step();
        }
    }
}

examples/rc.rs (line 27)

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 (line 24)

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/exchange.rs (line 28)

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

        let batch = std::env::args().nth(1).unwrap().parse::<usize>().unwrap();
        let rounds = std::env::args().nth(2).unwrap().parse::<usize>().unwrap();
        let mut input = InputHandle::new();

        // create a new input, exchange data, and inspect its output
        let probe = worker.dataflow(|scope|
            scope
                .input_from(&mut input)
                .container::<Vec<_>>()
                .exchange(|&x| x as u64)
                .probe()
                .0
        );


        let timer = std::time::Instant::now();

        for round in 0 .. rounds {

            for i in 0 .. batch {
                input.send(i);
            }
            input.advance_to(round);

            while probe.less_than(input.time()) {
                worker.step();
            }

        }

        let volume = (rounds * batch) as f64;
        let elapsed = timer.elapsed();
        let seconds = elapsed.as_secs() as f64 + (f64::from(elapsed.subsec_nanos())/1000000000.0);

        println!("{:?}\tworker {} complete; rate: {:?}", timer.elapsed(), worker.index(), volume / seconds);

    }).unwrap();
}

examples/event_driven.rs (line 38)

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

        let timer = std::time::Instant::now();

        let mut args = std::env::args();
        args.next();

        let dataflows = args.next().unwrap().parse::<usize>().unwrap();
        let length = args.next().unwrap().parse::<usize>().unwrap();
        let record = args.next() == Some("record".to_string());

        let mut inputs = Vec::new();
        let mut probes = Vec::new();

        // create a new input, exchange data, and inspect its output
        for _dataflow in 0 .. dataflows {
            worker.dataflow(|scope| {
                let (input, mut stream) = scope.new_input();
                for _step in 0 .. length {
                    stream = stream.map(|x: ()| x);
                }
                let (probe, _stream) = stream.probe();
                inputs.push(input);
                probes.push(probe);
            });
        }

        println!("{:?}\tdataflows built ({} x {})", timer.elapsed(), dataflows, length);

        for round in 0 .. {
            let dataflow = round % dataflows;
            if record {
                inputs[dataflow].send(());
            }
            inputs[dataflow].advance_to(round);
            let mut steps = 0;
            while probes[dataflow].less_than(&round) {
                worker.step();
                steps += 1;
            }
            println!("{:?}\tround {} complete in {} steps", timer.elapsed(), round, steps);
        }

    }).unwrap();
}

examples/unionfind.rs (line 39)

fn main() {

    // command-line args: numbers of nodes and edges in the random graph.
    let nodes: usize = std::env::args().nth(1).unwrap().parse().unwrap();
    let edges: usize = std::env::args().nth(2).unwrap().parse().unwrap();
    let batch: usize = std::env::args().nth(3).unwrap().parse().unwrap();

    timely::execute_from_args(std::env::args().skip(4), move |worker| {

        let index = worker.index();
        let peers = worker.peers();

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

        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                //  .exchange(move |x: &(usize, usize)| (x.0 % (peers - 1)) as u64 + 1)
                 .union_find()
                 .exchange(|_| 0)
                 .union_find()
                 .probe_with(&probe);
        });

        // Generate roughly random data.
        use std::hash::{BuildHasher, BuildHasherDefault, DefaultHasher};
        let hasher = BuildHasherDefault::<DefaultHasher>::new();
        let insert = (0..).map(move |i| (hasher.hash_one(&(i,index,0)) as usize % nodes,
                                         hasher.hash_one(&(i,index,1)) as usize % nodes));

        for (edge, arc) in insert.take(edges / peers).enumerate() {
            input.send(arc);
            if edge % batch == (batch - 1) {
                let next = input.epoch() + 1;
                input.advance_to(next);
                while probe.less_than(input.time()) {
                    worker.step();
                }
            }
        }

    }).unwrap(); // asserts error-free execution;
}

Additional examples can be found in:

• examples/distinct.rs
• examples/wordcount.rs
• examples/hashjoin.rs
• examples/loopdemo.rs
• examples/logging-send.rs
• examples/openloop.rs
• examples/columnar.rs
• examples/pagerank.rs

Trait Implementations

impl<T: Debug + Timestamp, CB: Debug + ContainerBuilder<Container: Clone>> Debug for Handle<T, CB>

where CB::Container: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T: Timestamp, CB: ContainerBuilder<Container: Clone>> Default for Handle<T, CB>

fn default() -> Self

Returns the “default value” for a type.

impl<T: Timestamp, CB: ContainerBuilder<Container: Clone>> Drop for Handle<T, CB>

fn drop(&mut self)

Executes the destructor for this type.

impl<T, CB, D> PushInto<D> for Handle<T, CB>

where T: Timestamp, CB: ContainerBuilder<Container: Clone> + PushInto<D>,

fn push_into(&mut self, item: D)

Push item into self.