Struct MutableAntichain 

pub struct MutableAntichain<T> { /* private fields */ }

An antichain based on a multiset whose elements frequencies can be updated.

The MutableAntichain maintains frequencies for many elements of type T, and exposes the set of elements with positive count not greater than any other elements with positive count. The antichain may both advance and retreat; the changes do not all need to be to elements greater or equal to some elements of the frontier.

The type T must implement PartialOrder as well as Ord. The implementation of the Ord trait is used to efficiently organize the updates for cancellation, and to efficiently determine the lower bounds, and only needs to not contradict the PartialOrder implementation (that is, if PartialOrder orders two elements, then so does the Ord implementation).

The MutableAntichain implementation is done with the intent that updates to it are done in batches, and it is acceptable to rebuild the frontier from scratch when a batch of updates change it. This means that it can be expensive to maintain a large number of counts and change few elements near the frontier.

Implementations

impl<T> MutableAntichain<T>

pub fn new() -> MutableAntichain<T>

Creates a new empty MutableAntichain.

Examples

 use timely::progress::frontier::MutableAntichain;

 let frontier = MutableAntichain::<usize>::new();
 assert!(frontier.is_empty());

pub fn clear(&mut self)

Removes all elements.

Examples

 use timely::progress::frontier::MutableAntichain;

 let mut frontier = MutableAntichain::<usize>::new();
 frontier.clear();
 assert!(frontier.is_empty());

pub fn frontier(&self) -> AntichainRef<'_, T>

Reveals the minimal elements with positive count.

Examples

 use timely::progress::frontier::MutableAntichain;

 let mut frontier = MutableAntichain::<usize>::new();
 assert!(frontier.frontier().len() == 0);

pub fn new_bottom(bottom: T) -> MutableAntichain<T>

where T: Ord + Clone,

Creates a new singleton MutableAntichain.

Examples

 use timely::progress::frontier::{AntichainRef, MutableAntichain};

 let mut frontier = MutableAntichain::new_bottom(0u64);
 assert!(frontier.frontier() == AntichainRef::new(&[0u64]));

pub fn is_empty(&self) -> bool

Returns true if there are no elements in the MutableAntichain.

Examples

 use timely::progress::frontier::MutableAntichain;

 let mut frontier = MutableAntichain::<usize>::new();
 assert!(frontier.is_empty());

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

where T: PartialOrder<O>,

Returns true if any item in the MutableAntichain is strictly less than the argument.

Examples

 use timely::progress::frontier::MutableAntichain;

 let mut frontier = MutableAntichain::new_bottom(1u64);
 assert!(!frontier.less_than(&0));
 assert!(!frontier.less_than(&1));
 assert!(frontier.less_than(&2));

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

where T: PartialOrder<O>,

Returns true if any item in the MutableAntichain is less than or equal to the argument.

Examples

 use timely::progress::frontier::MutableAntichain;

 let mut frontier = MutableAntichain::new_bottom(1u64);
 assert!(!frontier.less_equal(&0));
 assert!(frontier.less_equal(&1));
 assert!(frontier.less_equal(&2));

Examples found in repository

examples/wordcount.rs (line 39)

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

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

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 update_iter<I>(&mut self, updates: I) -> Drain<'_, [(T, i64); 2]>

where T: Clone + PartialOrder + Ord, I: IntoIterator<Item = (T, i64)>,

Applies updates to the antichain and enumerates any changes.

Examples

 use timely::progress::frontier::{AntichainRef, MutableAntichain};

 let mut frontier = MutableAntichain::new_bottom(1u64);
 let changes =
 frontier
     .update_iter(vec![(1, -1), (2, 7)])
     .collect::<Vec<_>>();

 assert!(frontier.frontier() == AntichainRef::new(&[2]));
 assert!(changes == vec![(1, -1), (2, 1)]);

pub fn count_for<O>(&self, query_time: &O) -> i64

where T: PartialEq<O>,

Reports the count for a queried time.

pub fn updates(&mut self) -> impl Iterator<Item = &(T, i64)>

where T: Clone + PartialOrder + Ord,

Reports the updates that form the frontier. Returns an iterator of timestamps and their frequency.

Rebuilds the internal representation before revealing times and frequencies.

Trait Implementations

impl<T: Clone> Clone for MutableAntichain<T>

fn clone(&self) -> MutableAntichain<T>

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<T: Debug> Debug for MutableAntichain<T>

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

Formats the value using the given formatter.

impl<T> Default for MutableAntichain<T>

fn default() -> Self

Returns the “default value” for a type.

impl<'de, T> Deserialize<'de> for MutableAntichain<T>

where T: Deserialize<'de>,

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl<T: PartialOrder + Ord + Clone> From<Antichain<T>> for MutableAntichain<T>

fn from(antichain: Antichain<T>) -> Self

Converts to this type from the input type.

impl<'a, T: PartialOrder + Ord + Clone> From<AntichainRef<'a, T>> for MutableAntichain<T>

fn from(antichain: AntichainRef<'a, T>) -> Self

Converts to this type from the input type.

impl<T> FromIterator<(T, i64)> for MutableAntichain<T>

where T: Clone + PartialOrder + Ord,

fn from_iter<I>(iterator: I) -> Self

where I: IntoIterator<Item = (T, i64)>,

Creates a value from an iterator.

impl<T> Serialize for MutableAntichain<T>

where T: Serialize,

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.