Struct Handle 

pub struct Handle<T: Timestamp> { /* private fields */ }

Reports information about progress at the probe.

Implementations

impl<T: Timestamp> Handle<T>

pub fn less_than(&self, time: &T) -> bool

Returns true iff the frontier is strictly less than time.

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.time() + 1;
                input.advance_to(next);
                while probe.less_than(input.time()) {
                    worker.step();
                }
            }
        }

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

examples/event_driven.rs (line 47)

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, stream) = scope.new_input();
                let stream = scope.region(|inner| {
                    let mut stream = stream.enter(inner);
                    for _step in 0 .. length {
                        stream = stream.map(|x: ()| x);
                    }
                    stream.leave()
                });
                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;
            }
            if round % 1000 == 0 { println!("{:?}\tround {} complete in {} steps", timer.elapsed(), round, steps); }
        }

    }).unwrap();
}

Additional examples can be found in:

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

pub fn less_equal(&self, time: &T) -> bool

Returns true iff the frontier is less than or equal to time.

pub fn done(&self) -> bool

Returns true iff the frontier is empty.

pub fn new() -> Self

Allocates a new handle.

Examples found in repository

examples/threadless.rs (line 13)

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 16)

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 10)

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/flow_controlled.rs (line 8)

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {
        let mut input = (0u64..100000).peekable();
        worker.dataflow(|scope| {
            let probe_handle = probe::Handle::new();
            let probe_handle_2 = probe_handle.clone();

            iterator_source(
                scope,
                "Source",
                move |prev_t| {
                    if let Some(first_x) = input.peek().cloned() {
                        let next_t = first_x / 100 * 100;
                        Some(IteratorSourceInput {
                            lower_bound: Default::default(),
                            data: vec![
                                (next_t,
                                 input.by_ref().take(10).map(|x| (/* "timestamp" */ x, x)).collect::<Vec<_>>())],
                            target: *prev_t,
                        })
                    } else {
                        None
                    }
                },
                probe_handle_2)
            .inspect_time(|t, d| eprintln!("@ {:?}: {:?}", t, d))
            .probe_with(&probe_handle);
        });
    }).unwrap();
}

examples/unionfind.rs (line 21)

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.time() + 1;
                input.advance_to(next);
                while probe.less_than(input.time()) {
                    worker.step();
                }
            }
        }

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

examples/distinct.rs (line 13)

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

pub fn with_frontier<R, F: FnMut(AntichainRef<'_, T>) -> R>( &self, function: F, ) -> R

Invokes a method on the frontier, returning its result.

This method allows inspection of the frontier, which cannot be returned by reference as it is on the other side of a RefCell.

Examples

use timely::dataflow::operators::probe::Handle;

let handle = Handle::<usize>::new();
let frontier = handle.with_frontier(|frontier| frontier.to_vec());

Examples found in repository

examples/loopdemo.rs (line 58)

fn main() {

    let mut args = std::env::args();
    args.next();
    let rate: usize = args.next().expect("Must specify rate").parse().expect("Rate must be an usize");
    let duration_s: usize = args.next().expect("Must specify duration_s").parse().expect("duration_s must be an usize");

    timely::execute_from_args(args, move |worker| {

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

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

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

        // Create a dataflow that discards input data (just synchronizes).
        worker.dataflow(|scope| {

            let stream = scope.input_from(&mut input);

            let (loop_handle, loop_stream) = scope.feedback(1);

            let step =
            stream
                .concat(loop_stream)
                .map(|x| if x % 2 == 0 { x / 2 } else { 3 * x + 1 })
                .filter(|x| x > &1);

            step.probe_with(&probe)
                .connect_loop(loop_handle);
        });

        let ns_per_request = 1_000_000_000 / rate;
        let mut insert_counter = index;           // counts up as we insert records.
        let mut retire_counter = index;           // counts up as we retire records.

        let mut inserted_ns = 0;

        // We repeatedly consult the elapsed time, and introduce any data now considered available.
        // At the same time, we observe the output and record which inputs are considered retired.

        let mut counts = vec![[0u64; 16]; 64];

        let counter_limit = rate * duration_s;
        while retire_counter < counter_limit {

            // Open-loop latency-throughput test, parameterized by offered rate `ns_per_request`.
            let elapsed = timer.elapsed();
            let elapsed_ns = elapsed.as_secs() * 1_000_000_000 + (elapsed.subsec_nanos() as u64);

            // Determine completed ns.
            let acknowledged_ns: u64 = probe.with_frontier(|frontier| frontier[0]);

            // Notice any newly-retired records.
            while ((retire_counter * ns_per_request) as u64) < acknowledged_ns && retire_counter < counter_limit {
                let requested_at = (retire_counter * ns_per_request) as u64;
                let latency_ns = elapsed_ns - requested_at;

                let count_index = latency_ns.next_power_of_two().trailing_zeros() as usize;
                let low_bits = ((elapsed_ns - requested_at) >> (count_index - 5)) & 0xF;
                counts[count_index][low_bits as usize] += 1;

                retire_counter += peers;
            }

            // Now, should we introduce more records before stepping the worker?
            // Three choices here:
            //
            //   1. Wait until previous batch acknowledged.
            //   2. Tick at most once every millisecond-ish.
            //   3. Geometrically increasing outstanding batches.

            // Technique 1:
            // let target_ns = if acknowledged_ns >= inserted_ns { elapsed_ns } else { inserted_ns };

            // Technique 2:
            // let target_ns = elapsed_ns & !((1 << 20) - 1);

            // Technique 3:
            let scale = (inserted_ns - acknowledged_ns).next_power_of_two();
            let target_ns = elapsed_ns & !(scale - 1);

            if inserted_ns < target_ns {

                while ((insert_counter * ns_per_request) as u64) < target_ns {
                    input.send(insert_counter);
                    insert_counter += peers;
                }
                input.advance_to(target_ns);
                inserted_ns = target_ns;
            }

            worker.step();
        }

        // Report observed latency measurements.
        if index == 0 {

            let mut results = Vec::new();
            let total = counts.iter().map(|x| x.iter().sum::<u64>()).sum();
            let mut sum = 0;
            for index in (10 .. counts.len()).rev() {
                for sub in (0 .. 16).rev() {
                    if sum > 0 && sum < total {
                        let latency = (1 << (index-1)) + (sub << (index-5));
                        let fraction = (sum as f64) / (total as f64);
                        results.push((latency, fraction));
                    }
                    sum += counts[index][sub];
                }
            }
            for (latency, fraction) in results.drain(..).rev() {
                println!("{}\t{}", latency, fraction);
            }
        }

    }).unwrap();
}

examples/openloop.rs (line 52)

fn main() {

    let mut args = std::env::args();
    args.next();
    let rate: usize = args.next().expect("Must specify rate").parse().expect("Rate must be an usize");
    let duration_s: usize = args.next().expect("Must specify duration_s").parse().expect("duration_s must be an usize");

    timely::execute_from_args(args, move |worker| {

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

        // re-synchronize all workers (account for start-up).
        timely::synchronization::Barrier::new(worker).wait();

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

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

        // Create a dataflow that discards input data (just synchronizes).
        worker.dataflow(|scope| {
            scope
                .input_from(&mut input)     // read input.
                .filter(|_| false)          // do nothing.
                .probe_with(&probe);    // observe output.
        });

        let ns_per_request = 1_000_000_000 / rate;
        let mut insert_counter = index;           // counts up as we insert records.
        let mut retire_counter = index;           // counts up as we retire records.

        let mut inserted_ns = 0;

        // We repeatedly consult the elapsed time, and introduce any data now considered available.
        // At the same time, we observe the output and record which inputs are considered retired.

        let mut counts = vec![[0u64; 16]; 64];

        let counter_limit = rate * duration_s;
        while retire_counter < counter_limit {

            // Open-loop latency-throughput test, parameterized by offered rate `ns_per_request`.
            let elapsed = timer.elapsed();
            let elapsed_ns = elapsed.as_secs() * 1_000_000_000 + (elapsed.subsec_nanos() as u64);

            // Determine completed ns.
            let acknowledged_ns: u64 = probe.with_frontier(|frontier| frontier[0]);

            // Notice any newly-retired records.
            while ((retire_counter * ns_per_request) as u64) < acknowledged_ns && retire_counter < counter_limit {
                let requested_at = (retire_counter * ns_per_request) as u64;
                let latency_ns = elapsed_ns - requested_at;

                let count_index = latency_ns.next_power_of_two().trailing_zeros() as usize;
                let low_bits = ((elapsed_ns - requested_at) >> (count_index - 5)) & 0xF;
                counts[count_index][low_bits as usize] += 1;

                retire_counter += peers;
            }

            // Now, should we introduce more records before stepping the worker?
            // Three choices here:
            //
            //   1. Wait until previous batch acknowledged.
            //   2. Tick at most once every millisecond-ish.
            //   3. Geometrically increase outstanding batches.

            // Technique 1:
            // let target_ns = if acknowledged_ns >= inserted_ns { elapsed_ns } else { inserted_ns };

            // Technique 2:
            // let target_ns = elapsed_ns & !((1 << 20) - 1);

            // Technique 3:
            let target_ns = {
                let delta: u64 = inserted_ns - acknowledged_ns;
                let bits = ::std::mem::size_of::<u64>() * 8 - delta.leading_zeros() as usize;
                let scale = ::std::cmp::max((1 << bits) / 4, 1024);
                elapsed_ns & !(scale - 1)
            };

            // Common for each technique.
            if inserted_ns < target_ns {

                while ((insert_counter * ns_per_request) as u64) < target_ns {
                    input.send(insert_counter);
                    insert_counter += peers;
                }
                input.advance_to(target_ns);
                inserted_ns = target_ns;
            }

            worker.step();
        }

        // Report observed latency measurements.
        if index == 0 {

            let mut results = Vec::new();
            let total = counts.iter().map(|x| x.iter().sum::<u64>()).sum();
            let mut sum = 0;
            for index in (10 .. counts.len()).rev() {
                for sub in (0 .. 16).rev() {
                    if sum > 0 && sum < total {
                        let latency = (1 << (index-1)) + (sub << (index-5));
                        let fraction = (sum as f64) / (total as f64);
                        results.push((latency, fraction));
                    }
                    sum += counts[index][sub];
                }
            }
            for (latency, fraction) in results.drain(..).rev() {
                println!("{}\t{}", latency, fraction);
            }
        }

    }).unwrap();
}

Trait Implementations

impl<T: Timestamp> Clone for Handle<T>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T: Debug + Timestamp> Debug for Handle<T>

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

Formats the value using the given formatter.

impl<T> Default for Handle<T>

where T: Timestamp,

fn default() -> Self

Returns the “default value” for a type.