<commit_before>use progress::frontier::MutableAntichain;
use progress::Timestamp;
use progress::count_map::CountMap;
use dataflow::operators::Capability;

/// Tracks requests for notification and delivers available notifications.
///
/// Notificator is meant to manage the delivery of requested notifications in the presence of
/// inputs that may have outstanding messages to deliver. The notificator tracks the frontiers,
/// as presented from the outside, for each input. Requested notifications can be served only
/// once there are no frontier elements less-or-equal to them, and there are no other pending
/// notification requests less than them. Each with be less-or-equal to itself, so we want to
/// dodge that corner case.
pub struct Notificator<T: Timestamp> {
    pending: Vec<(Capability<T>, u64)>,
    frontier: Vec<MutableAntichain<T>>,
    available: Vec<(Capability<T>, u64)>,
    candidates: Vec<(Capability<T>, u64)>,
}

impl<T: Timestamp> Notificator<T> {
    /// Allocates a new `Notificator`.
    pub fn new() -> Notificator<T> {
        Notificator {
            pending: Vec::new(),
            frontier: Vec::new(),
            available: Vec::new(),
            candidates: Vec::new(),
        }
    }

    /// Updates the `Notificator`'s frontiers from a `CountMap` per input.
    pub fn update_frontier_from_cm(&mut self, count_map: &mut [CountMap<T>]) {
        while self.frontier.len() < count_map.len() {
            self.frontier.push(MutableAntichain::new());
        }

        for (index, counts) in count_map.iter_mut().enumerate() {
            self.frontier[index].update_iter_and(counts.drain(), |_,_| { });
            // for (time, delta) in counts.drain() {
            //     self.frontier[index].update(&time, delta);
            // }
        }
    }

    /// Reveals the elements in the frontier of the indicated input.
    pub fn frontier(&self, input: usize) -> &[T] {
        self.frontier[input].frontier()
    }

    /// Requests a notification at the time associated with capability `cap`. Takes ownership of
    /// the capability.
    ///
    /// In order to request a notification at future timestamp, obtain a capability for the new
    /// timestamp first, as show in the example.
    ///
    /// #Examples
    /// ```
    /// use timely::dataflow::operators::ToStream;
    /// use timely::dataflow::operators::generic::unary::Unary;
    /// use timely::dataflow::channels::pact::Pipeline;
    ///
    /// timely::example(|scope| {
    ///     (0..10).to_stream(scope)
    ///            .unary_notify(Pipeline, "example", Vec::new(), |input, output, notificator| {
    ///                input.for_each(|cap, data| {
    ///                    output.session(&cap).give_content(data);
    ///                    let mut time = cap.time().clone();
    ///                    time.inner += 1;
    ///                    notificator.notify_at(cap.delayed(&time));
    ///                });
    ///                notificator.for_each(|cap,_,_| {
    ///                    println!("done with time: {:?}", cap.time());
    ///                });
    ///            });
    /// });
    /// ```
    #[inline]
    pub fn notify_at(&mut self, cap: Capability<T>) {
        let push = if let Some(&mut (_, ref mut count)) =
            self.pending.iter_mut().find(|&&mut (ref c, _)| c.time().eq(&cap.time())) {
                *count += 1;
                None
            } else {
                Some((cap, 1))
            };
        if let Some(p) = push {
            self.pending.push(p);
        }
    }

    /// Repeatedly calls `logic` till exhaustion of the available notifications.
    ///
    /// `logic` receives a capability for `t`, the timestamp being notified and a `count`
    /// representing how many capabilities were requested for that specific timestamp.
    #[inline(never)]
    pub fn for_each<F: FnMut(Capability<T>, u64, &mut Notificator<T>)>(&mut self, mut logic: F) {
        while let Some((cap, count)) = self.next() {
        // for (cap, count) in self {
            ::logging::log(&::logging::GUARDED_PROGRESS, true);
            logic(cap, count, self);
            ::logging::log(&::logging::GUARDED_PROGRESS, false);
        }
    }
}

impl<T: Timestamp> Iterator for Notificator<T> {
    type Item = (Capability<T>, u64);

    /// Retrieve the next available notification.
    ///
    /// Returns `None` if no notification is available. Returns `Some(cap, count)` otherwise:
    /// `cap` is a a capability for `t`, the timestamp being notified and, `count` represents
    /// how many notifications (out of those requested) are being delivered for that specific
    /// timestamp.
    #[inline(never)]
    fn next(&mut self) -> Option<(Capability<T>, u64)> {
        if self.available.is_empty() {
            let mut available = &mut self.available; // available is empty
            let mut pending = &mut self.pending;
            let mut candidates = &mut self.candidates;
            assert!(candidates.len() == 0);
            let frontier = &self.frontier;

            pending.drain_into_if(&mut candidates, |&(ref cap, _)| {
                !frontier.iter().any(|x| x.less_equal(&cap.time()))
            });

            while let Some((cap, count)) = candidates.pop() {
                let mut cap_in_minimal_antichain = available.is_empty();
                available.drain_into_if(&mut pending, |&(ref avail_cap, _)| {
                    let cap_lt_available = cap.time().less_than(&avail_cap.time());
                    cap_in_minimal_antichain |= cap_lt_available ||
                        (!cap.time().less_equal(&avail_cap.time()) && !avail_cap.time().less_equal(&cap.time()));
                    cap_lt_available
                });
                if cap_in_minimal_antichain {
                    available.push((cap, count));
                } else {
                    pending.push((cap, count));
                }
            }
        }

        self.available.pop()
    }<commit_msg>notify_at operates in bulk<commit_after>use std::collections::VecDeque;

use progress::frontier::MutableAntichain;
use progress::Timestamp;
use progress::count_map::CountMap;
use dataflow::operators::Capability;

/// Tracks requests for notification and delivers available notifications.
///
/// Notificator is meant to manage the delivery of requested notifications in the presence of
/// inputs that may have outstanding messages to deliver. The notificator tracks the frontiers,
/// as presented from the outside, for each input. Requested notifications can be served only
/// once there are no frontier elements less-or-equal to them, and there are no other pending
/// notification requests less than them. Each with be less-or-equal to itself, so we want to
/// dodge that corner case.
pub struct Notificator<T: Timestamp> {
    pending: Vec<(Capability<T>, u64)>,
    frontier: Vec<MutableAntichain<T>>,
    available: VecDeque<(Capability<T>, u64)>,
    candidates: Vec<(Capability<T>, u64)>,
}

impl<T: Timestamp> Notificator<T> {
    /// Allocates a new `Notificator`.
    pub fn new() -> Notificator<T> {
        Notificator {
            pending: Vec::new(),
            frontier: Vec::new(),
            available: VecDeque::new(),
            candidates: Vec::new(),
        }
    }

    /// Updates the `Notificator`'s frontiers from a `CountMap` per input.
    pub fn update_frontier_from_cm(&mut self, count_map: &mut [CountMap<T>]) {
        while self.frontier.len() < count_map.len() {
            self.frontier.push(MutableAntichain::new());
        }

        for (index, counts) in count_map.iter_mut().enumerate() {
            self.frontier[index].update_iter(counts.drain());
        }
    }

    /// Reveals the elements in the frontier of the indicated input.
    pub fn frontier(&self, input: usize) -> &[T] {
        self.frontier[input].frontier()
    }

    /// Requests a notification at the time associated with capability `cap`. Takes ownership of
    /// the capability.
    ///
    /// In order to request a notification at future timestamp, obtain a capability for the new
    /// timestamp first, as show in the example.
    ///
    /// #Examples
    /// ```
    /// use timely::dataflow::operators::ToStream;
    /// use timely::dataflow::operators::generic::unary::Unary;
    /// use timely::dataflow::channels::pact::Pipeline;
    ///
    /// timely::example(|scope| {
    ///     (0..10).to_stream(scope)
    ///            .unary_notify(Pipeline, "example", Vec::new(), |input, output, notificator| {
    ///                input.for_each(|cap, data| {
    ///                    output.session(&cap).give_content(data);
    ///                    let mut time = cap.time().clone();
    ///                    time.inner += 1;
    ///                    notificator.notify_at(cap.delayed(&time));
    ///                });
    ///                notificator.for_each(|cap,_,_| {
    ///                    println!("done with time: {:?}", cap.time());
    ///                });
    ///            });
    /// });
    /// ```
    #[inline]
    pub fn notify_at(&mut self, cap: Capability<T>) {
        self.pending.push((cap, 1));
        // let push = if let Some(&mut (_, ref mut count)) =
        //     self.pending.iter_mut().find(|&&mut (ref c, _)| c.time().eq(&cap.time())) {
        //         *count += 1;
        //         None
        //     } else {
        //         Some((cap, 1))
        //     };
        // if let Some(p) = push {
        //     self.pending.push(p);
        // }
    }

    /// Repeatedly calls `logic` till exhaustion of the available notifications.
    ///
    /// `logic` receives a capability for `t`, the timestamp being notified and a `count`
    /// representing how many capabilities were requested for that specific timestamp.
    #[inline(never)]
    pub fn for_each<F: FnMut(Capability<T>, u64, &mut Notificator<T>)>(&mut self, mut logic: F) {
        while let Some((cap, count)) = self.next() {
        // for (cap, count) in self {
            ::logging::log(&::logging::GUARDED_PROGRESS, true);
            logic(cap, count, self);
            ::logging::log(&::logging::GUARDED_PROGRESS, false);
        }
    }
}

impl<T: Timestamp> Iterator for Notificator<T> {
    type Item = (Capability<T>, u64);

    /// Retrieve the next available notification.
    ///
    /// Returns `None` if no notification is available. Returns `Some(cap, count)` otherwise:
    /// `cap` is a a capability for `t`, the timestamp being notified and, `count` represents
    /// how many notifications (out of those requested) are being delivered for that specific
    /// timestamp.
    #[inline(never)]
    fn next(&mut self) -> Option<(Capability<T>, u64)> {

        if self.available.is_empty() {
            self.make_available();
        }

        self.available.pop_front()

        //     let mut available = &mut self.available; // available is empty
        //     let mut pending = &mut self.pending;
        //     let mut candidates = &mut self.candidates;
        //     assert!(candidates.len() == 0);
        //     let frontier = &self.frontier;

        //     pending.drain_into_if(&mut candidates, |&(ref cap, _)| {
        //         !frontier.iter().any(|x| x.less_equal(&cap.time()))
        //     });

        //     while let Some((cap, count)) = candidates.pop() {
        //         let mut cap_in_minimal_antichain = available.is_empty();
        //         available.drain_into_if(&mut pending, |&(ref avail_cap, _)| {
        //             let cap_lt_available = cap.time().less_than(&avail_cap.time());
        //             cap_in_minimal_antichain |= cap_lt_available ||
        //                 (!cap.time().less_equal(&avail_cap.time()) && !avail_cap.time().less_equal(&cap.time()));
        //             cap_lt_available
        //         });
        //         if cap_in_minimal_antichain {
        //             available.push((cap, count));
        //         } else {
        //             pending.push((cap, count));
        //         }
        //     }
        // }

        // self.available.pop()
    }
}

impl<T: Timestamp> Notificator<T> {

    // appends elements of `self.pending` not `greater_equal` to `self.frontier` into `self.available`.
    fn make_available(&mut self) {

        // By invariant, nothing in self.available is greater_equal anything in self.pending.
        // It should be safe to append any ordered subset of self.pending to self.available,
        // in that the sequence of capabilities in self.available will remain non-decreasing.

        if self.pending.len() > 0 {
            self.pending.sort_by(|x,y| x.0.time().cmp(y.0.time()));
            for i in 0 .. self.pending.len() - 1 {
                if self.pending[i].0.time() == self.pending[i+1].0.time() {
                    self.pending[i+1].1 += self.pending[i].1;
                    self.pending[i].1 = 0;
                }
            }
            self.pending.retain(|x| x.1 > 0);

            for i in 0 .. self.pending.len() {
                if self.frontier.iter().all(|f| !f.less_equal(&self.pending[i].0)) {
                    // TODO : This clones a capability, whereas we could move it instead.
                    self.available.push_back((self.pending[i].0.clone(), self.pending[i].1));
                    self.pending[i].1 = 0;
                }
            }
            self.pending.retain(|x| x.1 > 0);
        }
    }
{'fraction_non_alphanumeric': 0.14722696562476797, 'fraction_numerical': 0.004603162818323557, 'mean_word_length': 1.9487740805604203, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 58, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1206572', 'n_tokens_mistral': 3970, 'n_tokens_neox': 3819, 'n_words': 1005}