Struct Worker 

pub struct Worker<A: Allocate> { /* private fields */ }

A Worker is the entry point to a timely dataflow computation. It wraps a Allocate, and has a list of dataflows that it manages.

Implementations

impl<A: Allocate> Worker<A>

pub fn new(config: Config, c: A, now: Option<Instant>) -> Worker<A>

Allocates a new Worker bound to a channel allocator.

Examples found in repository

examples/threadless.rs (line 9)

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();
        }
    }
}

pub fn step(&mut self) -> bool

Performs one step of the computation.

A step gives each dataflow operator a chance to run, and is the main way to ensure that a computation proceeds.

Examples

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

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

    worker.dataflow::<usize,_,_>(|scope| {
        (0 .. 10)
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x));
    });

    worker.step();
});

Examples found in repository

examples/unordered_input.rs (line 15)

fn main() {
    timely::execute(Config::thread(), |worker| {
        let (mut input, mut cap) = worker.dataflow::<usize,_,_>(|scope| {
            let (input, stream) = scope.new_unordered_input();
            stream.container::<Vec<_>>().inspect_batch(|t, x| println!("{:?} -> {:?}", t, x));
            input
        });

        for round in 0..10 {
            input.activate().session(&cap).give(round);
            cap = cap.delayed(&(round + 1));
            worker.step();
        }
    }).unwrap();
}

examples/sequence.rs (line 20)

fn main() {
    timely::execute(Config::process(4), |worker| {

        let timer = Instant::now();
        let mut sequencer = Sequencer::new(worker, Instant::now());

        for round in 0 .. {
            // if worker.index() < 3 {
                std::thread::sleep(Duration::from_secs(1 + worker.index() as u64));
                sequencer.push(format!("worker {:?}, round {:?}", worker.index(), round));
            // }
            for element in &mut sequencer {
                println!("{:?}:\tWorker {:?}:\t recv'd: {:?}", timer.elapsed(), worker.index(), element);
            }
            worker.step();
        }

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

examples/threadless.rs (line 29)

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

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

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

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();
}

Additional examples can be found in:

• examples/unionfind.rs
• 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 step_or_park(&mut self, duration: Option<Duration>) -> bool

Performs one step of the computation.

A step gives each dataflow operator a chance to run, and is the main way to ensure that a computation proceeds.

This method takes an optional timeout and may park the thread until there is work to perform or until this timeout expires. A value of None allows the worker to park indefinitely, whereas a value of Some(Duration::new(0, 0)) will return without parking the thread.

Examples

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

    use std::time::Duration;
    use timely::dataflow::operators::{ToStream, Inspect};

    worker.dataflow::<usize,_,_>(|scope| {
        (0 .. 10)
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x));
    });

    worker.step_or_park(Some(Duration::from_secs(1)));
});

pub fn step_while<F: FnMut() -> bool>(&mut self, func: F)

Calls self.step() as long as func evaluates to true.

This method will continually execute even if there is not work for the worker to perform. Consider using the similar method Self::step_or_park_while(duration) to allow the worker to yield control if that is appropriate.

Examples

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

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

    let probe =
    worker.dataflow::<usize,_,_>(|scope| {
        (0 .. 10)
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe()
            .0
    });

    worker.step_while(|| probe.less_than(&0));
});

Examples found in repository

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();
}

pub fn step_or_park_while<F: FnMut() -> bool>( &mut self, duration: Option<Duration>, func: F, )

Calls self.step_or_park(duration) as long as func evaluates to true.

This method may yield whenever there is no work to perform, as performed by Self::step_or_park(). Please consult the documentation for further information about that method and its behavior. In particular, the method can park the worker indefinitely, if no new work re-awakens the worker.

Examples

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

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

    let probe =
    worker.dataflow::<usize,_,_>(|scope| {
        (0 .. 10)
            .to_stream(scope)
            .container::<Vec<_>>()
            .inspect(|x| println!("{:?}", x))
            .probe()
            .0
    });

    worker.step_or_park_while(None, || probe.less_than(&0));
});

pub fn index(&self) -> usize

The index of the worker out of its peers.

Examples

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

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

    println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);

});

Examples found in repository

examples/capture_send.rs (line 8)

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/sequence.rs (line 14)

fn main() {
    timely::execute(Config::process(4), |worker| {

        let timer = Instant::now();
        let mut sequencer = Sequencer::new(worker, Instant::now());

        for round in 0 .. {
            // if worker.index() < 3 {
                std::thread::sleep(Duration::from_secs(1 + worker.index() as u64));
                sequencer.push(format!("worker {:?}, round {:?}", worker.index(), round));
            // }
            for element in &mut sequencer {
                println!("{:?}:\tWorker {:?}:\t recv'd: {:?}", timer.elapsed(), worker.index(), element);
            }
            worker.step();
        }

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

examples/pingpong.rs (line 10)

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

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

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

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();
}

Additional examples can be found in:

• examples/logging-recv.rs
• examples/exchange.rs
• examples/unionfind.rs
• examples/distinct.rs
• examples/hashjoin.rs
• examples/loopdemo.rs
• examples/logging-send.rs
• examples/openloop.rs
• examples/bfs.rs
• examples/pagerank.rs

pub fn peers(&self) -> usize

The total number of peer workers.

Examples

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

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

    println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);

});

Examples found in repository

examples/pingpong.rs (line 11)

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

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/logging-recv.rs (line 16)

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
}

examples/unionfind.rs (line 18)

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

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;
}

examples/loopdemo.rs (line 15)

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();
}

Additional examples can be found in:

• examples/openloop.rs
• examples/bfs.rs
• examples/pagerank.rs

pub fn timer(&self) -> Option<Instant>

A timer started at the initiation of the timely computation.

Examples

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

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

    println!("{:?}\tWorker {} of {}", timer.elapsed(), index, peers);

});

pub fn new_identifier(&mut self) -> usize

Allocate a new worker-unique identifier.

This method is public, though it is not expected to be widely used outside of the timely dataflow system.

pub fn peek_identifier(&self) -> usize

The next worker-unique identifier to be allocated.

pub fn log_register(&self) -> Option<RefMut<'_, Registry>>

Access to named loggers.

Examples

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

    worker.log_register()
          .unwrap()
          .insert::<timely::logging::TimelyEventBuilder,_>("timely", |time, data|
              println!("{:?}\t{:?}", time, data)
          );
});

Examples found in repository

examples/logging-send.rs (line 19)

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

        // Register timely worker logging.
        worker.log_register().unwrap().insert::<TimelyEventBuilder,_>("timely", |time, data|
            if let Some(data) = data {
                data.iter().for_each(|x| println!("LOG1: {:?}", x))
            }
            else {
                println!("LOG1: Flush {time:?}");
            }
        );

        // Register timely progress logging.
        // Less generally useful: intended for debugging advanced custom operators or timely
        // internals.
        worker.log_register().unwrap().insert::<TimelyProgressEventBuilder<usize>,_>("timely/progress/usize", |time, data|
            if let Some(data) = data {
                data.iter().for_each(|x| {
                    println!("PROGRESS: {:?}", x);
                    let (_, ev) = x;
                    print!("PROGRESS: TYPED MESSAGES: ");
                    for (n, p, t, d) in ev.messages.iter() {
                        print!("{:?}, ", (n, p, t, d));
                    }
                    println!();
                    print!("PROGRESS: TYPED INTERNAL: ");
                    for (n, p, t, d) in ev.internal.iter() {
                        print!("{:?}, ", (n, p, t, d));
                    }
                    println!();
                })
            }
            else {
                println!("PROGRESS: Flush {time:?}");
            }
        );

        worker.log_register().unwrap().insert::<TrackerEventBuilder<usize>,_>("timely/reachability/usize", |time, data|
            if let Some(data) = data {
                data.iter().for_each(|x| {
                    println!("REACHABILITY: {:?}", x);
                })
            }
            else {
                println!("REACHABILITY: Flush {time:?}");
            }
        );

        worker.log_register().unwrap().insert::<TimelySummaryEventBuilder<usize>,_>("timely/summary/usize", |time, data|
            if let Some(data) = data {
                data.iter().for_each(|(_, x)| {
                    println!("SUMMARY: {:?}", x);
                })
            }
            else {
                println!("SUMMARY: Flush {time:?}");
            }
        );

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

        // Register timely worker logging.
        worker.log_register().unwrap().insert::<TimelyEventBuilder,_>("timely", |time, data|
            if let Some(data) = data {
                data.iter().for_each(|x| println!("LOG2: {:?}", x))
            }
            else {
                println!("LOG2: Flush {time:?}");
            }
        );

        // create a new input, exchange data, and inspect its output
        worker.dataflow(|scope| {
            scope
                .input_from(&mut input)
                .exchange(|&x| x as u64)
                .probe_with(&probe);
        });

        // Register user-level logging.
        type MyBuilder = CapacityContainerBuilder<Vec<(Duration, ())>>;
        worker.log_register().unwrap().insert::<MyBuilder,_>("input", |time, data|
            if let Some(data) = data {
                for element in data.iter() {
                    println!("Round tick at: {:?}", element.0);
                }
            }
            else {
                println!("Round flush at: {time:?}");
            }
        );

        let input_logger = worker.log_register().unwrap().get::<MyBuilder>("input").expect("Input logger absent");

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

        for round in 0 .. rounds {

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

            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();
}

pub fn dataflow<T, R, F>(&mut self, func: F) -> R

where T: Refines<()>, F: FnOnce(&mut Child<'_, Self, T>) -> R,

Construct a new dataflow.

Examples

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

    // We must supply the timestamp type here, although
    // it would generally be determined by type inference.
    worker.dataflow::<usize,_,_>(|scope| {

        // uses of `scope` to build dataflow

    });
});

Examples found in repository

examples/capture_send.rs (lines 11-16)

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/unordered_input.rs (lines 6-10)

fn main() {
    timely::execute(Config::thread(), |worker| {
        let (mut input, mut cap) = worker.dataflow::<usize,_,_>(|scope| {
            let (input, stream) = scope.new_unordered_input();
            stream.container::<Vec<_>>().inspect_batch(|t, x| println!("{:?} -> {:?}", t, x));
            input
        });

        for round in 0..10 {
            input.activate().session(&cap).give(round);
            cap = cap.delayed(&(round + 1));
            worker.step();
        }
    }).unwrap();
}

examples/threadless.rs (lines 16-22)

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/pingpong.rs (lines 12-22)

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 21-25)

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 17-23)

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();
}

Additional examples can be found in:

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

pub fn dataflow_named<T, R, F>(&mut self, name: &str, func: F) -> R

where T: Refines<()>, F: FnOnce(&mut Child<'_, Self, T>) -> R,

Construct a new dataflow with a (purely cosmetic) name.

Examples

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

    // We must supply the timestamp type here, although
    // it would generally be determined by type inference.
    worker.dataflow_named::<usize,_,_>("Some Dataflow", |scope| {

        // uses of `scope` to build dataflow

    });
});

pub fn dataflow_core<T, R, F, V>( &mut self, name: &str, logging: Option<TimelyLogger>, resources: V, func: F, ) -> R

where T: Refines<()>, F: FnOnce(&mut V, &mut Child<'_, Self, T>) -> R, V: Any + 'static,

Construct a new dataflow with specific configurations.

This method constructs a new dataflow, using a name, logger, and additional resources specified as argument. The name is cosmetic, the logger is used to handle events generated by the dataflow, and the additional resources are kept alive for as long as the dataflow is alive (use case: shared library bindings).

Examples

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

    // We must supply the timestamp type here, although
    // it would generally be determined by type inference.
    worker.dataflow_core::<usize,_,_,_>(
        "dataflow X",           // Dataflow name
        None,                   // Optional logger
        37,                     // Any resources
        |resources, scope| {    // Closure

            // uses of `resources`, `scope`to build dataflow

        }
    );
});

pub fn drop_dataflow(&mut self, dataflow_identifier: usize)

Drops an identified dataflow.

This method removes the identified dataflow, which will no longer be scheduled. Various other resources will be cleaned up, though the method is currently in public beta rather than expected to work. Please report all crashes and unmet expectations!

pub fn next_dataflow_index(&self) -> usize

Returns the next index to be used for dataflow construction.

This identifier will appear in the address of contained operators, and can be used to drop the dataflow using self.drop_dataflow().

pub fn installed_dataflows(&self) -> Vec<usize>

List the current dataflow indices.

pub fn has_dataflows(&self) -> bool

Returns true if there is at least one dataflow under management.

Trait Implementations

impl<A: Allocate> AsWorker for Worker<A>

fn config(&self) -> &Config

Returns the worker configuration parameters.

fn index(&self) -> usize

Index of the worker among its peers.

fn peers(&self) -> usize

Number of peer workers.

fn allocate<D: Exchangeable>( &mut self, identifier: usize, address: Rc<[usize]>, ) -> (Vec<Box<dyn Push<D>>>, Box<dyn Pull<D>>)

Allocates a new channel from a supplied identifier and address.

fn pipeline<T: 'static>( &mut self, identifier: usize, address: Rc<[usize]>, ) -> (ThreadPusher<T>, ThreadPuller<T>)

Constructs a pipeline channel from the worker to itself.

fn broadcast<T: Exchangeable + Clone>( &mut self, identifier: usize, address: Rc<[usize]>, ) -> (Box<dyn Push<T>>, Box<dyn Pull<T>>)

Allocates a broadcast channel, where each pushed message is received by all.

fn new_identifier(&mut self) -> usize

Allocates a new worker-unique identifier.

fn peek_identifier(&self) -> usize

The next worker-unique identifier to be allocated.

fn log_register(&self) -> Option<RefMut<'_, Registry>>

Provides access to named logging streams.

fn logger_for<CB: ContainerBuilder>(&self, name: &str) -> Option<Logger<CB>>

Acquires a logger by name, if the log register exists and the name is registered.

fn logging(&self) -> Option<TimelyLogger>

Provides access to the timely logging stream.

impl<A: Allocate> Clone for Worker<A>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<A: Allocate> Scheduler for Worker<A>

fn activations(&self) -> Rc<RefCell<Activations>>

Provides a shared handle to the activation scheduler.

fn activator_for(&self, path: Rc<[usize]>) -> Activator

Constructs an Activator tied to the specified operator address.

fn sync_activator_for(&self, path: Vec<usize>) -> SyncActivator

Constructs a SyncActivator tied to the specified operator address.

impl<A: Allocate> ScopeParent for Worker<A>

type Timestamp = ()

The timestamp associated with data in this scope.