Skip to main content

rc/
rc.rs

1use std::rc::Rc;
2use timely::dataflow::{InputHandle, ProbeHandle};
3use timely::dataflow::operators::{Input, Inspect, Probe};
4
5#[derive(Debug, Clone)]
6pub struct Test {
7    _field: Rc<usize>,
8}
9
10fn main() {
11    // initializes and runs a timely dataflow.
12    timely::execute_from_args(std::env::args(), |worker| {
13        // create a new input, exchange data, and inspect its output
14        let index = worker.index();
15        let mut input = InputHandle::new();
16        let probe = ProbeHandle::new();
17        worker.dataflow(|scope| {
18            scope.input_from(&mut input)
19                 .container::<Vec<_>>()
20                 //.exchange(|x| *x) // <-- cannot exchange this; Rc is not Send.
21                 .inspect(move |x| println!("worker {}:\thello {:?}", index, x))
22                 .probe_with(&probe);
23        });
24
25        // introduce data and watch!
26        for round in 0..10 {
27            input.send(Test { _field: Rc::new(round) } );
28            input.advance_to(round + 1);
29            worker.step_while(|| probe.less_than(input.time()));
30        }
31    }).unwrap();
32}