Struct Session 

pub struct Session<'a: 'b, 'b, T: Timestamp, CB: ContainerBuilder, CT: CapabilityTrait<T>> { /* private fields */ }

An active output building session, which accepts items and builds containers.

Implementations

impl<'a: 'b, 'b, T: Timestamp, CB: ContainerBuilder, CT: CapabilityTrait<T>> Session<'a, 'b, T, CB, CT>

pub fn builder(&mut self) -> &mut CB

Provides access to the underlying container builder.

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

where CB: PushInto<D>,

Provides one record at the time specified by the Session.

Examples found in repository

examples/unordered_input.rs (line 13)

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/unionfind.rs (line 82)

    fn union_find(self) -> StreamVec<G, (usize, usize)> {

        self.unary(Pipeline, "UnionFind", |_,_| {

            let mut roots = vec![];  // u32 works, and is smaller than uint/u64
            let mut ranks = vec![];  // u8 should be large enough (n < 2^256)

            move |input, output| {

                input.for_each_time(|time, data| {
                    let mut session = output.session(&time);
                    for &mut (mut x, mut y) in data.flatten() {

                        // grow arrays if required.
                        let m = ::std::cmp::max(x, y);
                        for i in roots.len() .. (m + 1) {
                            roots.push(i);
                            ranks.push(0);
                        }

                        // look up roots for `x` and `y`.
                        while x != roots[x] { x = roots[x]; }
                        while y != roots[y] { y = roots[y]; }

                        if x != y {
                            session.give((x, y));
                            match ranks[x].cmp(&ranks[y]) {
                                Ordering::Less    => { roots[x] = y },
                                Ordering::Greater => { roots[y] = x },
                                Ordering::Equal   => { roots[y] = x; ranks[x] += 1 },
                            }
                        }
                    }
                });
            }
        })
    }

examples/distinct.rs (line 31)

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {
        let index = worker.index();
        let mut input = InputHandle::new();
        let probe = ProbeHandle::new();

        // create a new input, exchange data, and inspect its output
        worker.dataflow::<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();
}

examples/wordcount.rs (line 45)

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/hashjoin.rs (line 46)

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.time() + 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/columnar.rs (line 53)

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(|b| b.is_ascii_whitespace()).filter(|s| !s.is_empty()).map(move |text| WordCountReference { text, diff: wordcount.diff })
                                    }) {
                                        session.give(wordcount);
                                    }
                                }
                            });
                        }
                    },
                )
                .container::<Container>()
                .unary_frontier(
                    ExchangeCore::<ColumnBuilder<InnerContainer>,_>::new_core(|x: &WordCountReference<&[u8],&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_vec(), *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!("  {}: {}", std::str::from_utf8(wc.text).unwrap_or("<invalid utf8>"), 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();
}

Additional examples can be found in:

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

pub fn give_iterator<I>(&mut self, iter: I)

where I: Iterator, CB: PushInto<I::Item>,

Provides an iterator of records at the time specified by the Session.

pub fn give_container(&mut self, container: &mut CB::Container)

Provide a container at the time specified by the Session.

pub fn give_containers<'c>( &mut self, containers: impl Iterator<Item = &'c mut CB::Container>, )

Provide multiple containers at the time specifid by the Session.

pub fn extract_and_send(&mut self)

Extracts built containers and sends them.

pub fn flush(&mut self)

Finalizes containers and sends them.

Trait Implementations

impl<'a: 'b, 'b, T: Timestamp, CB: ContainerBuilder, CT: CapabilityTrait<T>> Drop for Session<'a, 'b, T, CB, CT>

fn drop(&mut self)

Executes the destructor for this type.