timely/progress/frontier.rs
1//! Tracks minimal sets of mutually incomparable elements of a partial order.
2
3use serde::{Deserialize, Serialize};
4use smallvec::SmallVec;
5
6use crate::progress::ChangeBatch;
7use crate::order::{PartialOrder, TotalOrder};
8
9/// A set of mutually incomparable elements.
10///
11/// An antichain is a set of partially ordered elements, each of which is incomparable to the others.
12/// This antichain implementation allows you to repeatedly introduce elements to the antichain, and
13/// which will evict larger elements to maintain the *minimal* antichain, those incomparable elements
14/// no greater than any other element.
15///
16/// Two antichains are equal if they contain the same set of elements, even if in different orders.
17/// This can make equality testing quadratic, though linear in the common case that the sequences
18/// are identical.
19#[derive(Debug, Serialize, Deserialize)]
20pub struct Antichain<T> {
21 elements: SmallVec<[T; 1]>
22}
23
24impl<T: PartialOrder> Antichain<T> {
25 /// Updates the `Antichain` if the element is not greater than or equal to some present element.
26 ///
27 /// Returns `true` if element is added to the set
28 ///
29 /// # Examples
30 ///
31 ///```
32 /// use timely::progress::frontier::Antichain;
33 ///
34 /// let mut frontier = Antichain::new();
35 /// assert!(frontier.insert(2));
36 /// assert!(!frontier.insert(3));
37 ///```
38 pub fn insert(&mut self, element: T) -> bool {
39 if !self.elements.iter().any(|x| x.less_equal(&element)) {
40 self.elements.retain(|x| !element.less_equal(x));
41 self.elements.push(element);
42 true
43 }
44 else {
45 false
46 }
47 }
48
49 /// Updates the `Antichain` if the element is not greater than or equal to some present element.
50 ///
51 /// Returns `true` if element is added to the set
52 ///
53 /// Accepts a reference to an element, which is cloned when inserting.
54 ///
55 /// # Examples
56 ///
57 ///```
58 /// use timely::progress::frontier::Antichain;
59 ///
60 /// let mut frontier = Antichain::new();
61 /// assert!(frontier.insert_ref(&2));
62 /// assert!(!frontier.insert(3));
63 ///```
64 pub fn insert_ref(&mut self, element: &T) -> bool where T: Clone {
65 if !self.elements.iter().any(|x| x.less_equal(element)) {
66 self.elements.retain(|x| !element.less_equal(x));
67 self.elements.push(element.clone());
68 true
69 }
70 else {
71 false
72 }
73 }
74
75 /// Updates the `Antichain` if the element is not greater than or equal to some present element.
76 /// If the antichain needs updating, it uses the `to_owned` closure to convert the element into
77 /// a `T`.
78 ///
79 /// Returns `true` if element is added to the set
80 ///
81 /// # Examples
82 ///
83 ///```
84 /// use timely::progress::frontier::Antichain;
85 ///
86 /// let mut frontier = Antichain::new();
87 /// assert!(frontier.insert_with(&2, |x| *x));
88 /// assert!(!frontier.insert(3));
89 ///```
90 pub fn insert_with<O: PartialOrder<T>, F: FnOnce(&O) -> T>(&mut self, element: &O, to_owned: F) -> bool where T: PartialOrder<O> {
91 if !self.elements.iter().any(|x| x.less_equal(element)) {
92 self.elements.retain(|x| !element.less_equal(x));
93 self.elements.push(to_owned(element));
94 true
95 }
96 else {
97 false
98 }
99 }
100
101 /// Reserves capacity for at least additional more elements to be inserted in the given `Antichain`
102 pub fn reserve(&mut self, additional: usize) {
103 self.elements.reserve(additional);
104 }
105
106 /// Performs a sequence of insertion and returns `true` iff any insertion does.
107 ///
108 /// # Examples
109 ///
110 ///```
111 /// use timely::progress::frontier::Antichain;
112 ///
113 /// let mut frontier = Antichain::new();
114 /// assert!(frontier.extend(Some(3)));
115 /// assert!(frontier.extend(vec![2, 5]));
116 /// assert!(!frontier.extend(vec![3, 4]));
117 ///```
118 pub fn extend<I: IntoIterator<Item=T>>(&mut self, iterator: I) -> bool {
119 let mut added = false;
120 for element in iterator {
121 added = self.insert(element) || added;
122 }
123 added
124 }
125
126 /// Returns `true` if any item in the antichain is strictly less than the argument.
127 ///
128 /// # Examples
129 ///
130 ///```
131 /// use timely::progress::frontier::Antichain;
132 ///
133 /// let mut frontier = Antichain::from_elem(2);
134 /// assert!(frontier.less_than(&3));
135 /// assert!(!frontier.less_than(&2));
136 /// assert!(!frontier.less_than(&1));
137 ///
138 /// frontier.clear();
139 /// assert!(!frontier.less_than(&3));
140 ///```
141 #[inline]
142 pub fn less_than(&self, time: &T) -> bool {
143 self.elements.iter().any(|x| x.less_than(time))
144 }
145
146 /// Returns `true` if any item in the antichain is less than or equal to the argument.
147 ///
148 /// # Examples
149 ///
150 ///```
151 /// use timely::progress::frontier::Antichain;
152 ///
153 /// let mut frontier = Antichain::from_elem(2);
154 /// assert!(frontier.less_equal(&3));
155 /// assert!(frontier.less_equal(&2));
156 /// assert!(!frontier.less_equal(&1));
157 ///
158 /// frontier.clear();
159 /// assert!(!frontier.less_equal(&3));
160 ///```
161 #[inline]
162 pub fn less_equal(&self, time: &T) -> bool {
163 self.elements.iter().any(|x| x.less_equal(time))
164 }
165
166 /// Returns `true` if every element of `other` is greater or equal to some element of `self`.
167 #[deprecated(since="0.12.0", note="please use `PartialOrder::less_equal` instead")]
168 #[inline]
169 pub fn dominates(&self, other: &Antichain<T>) -> bool {
170 <Self as PartialOrder>::less_equal(self, other)
171 }
172}
173
174impl<T: PartialOrder> std::iter::FromIterator<T> for Antichain<T> {
175 fn from_iter<I>(iterator: I) -> Self
176 where
177 I: IntoIterator<Item=T>
178 {
179 let mut result = Self::new();
180 result.extend(iterator);
181 result
182 }
183}
184
185impl<T> Antichain<T> {
186
187 /// Creates a new empty `Antichain`.
188 ///
189 /// # Examples
190 ///
191 ///```
192 /// use timely::progress::frontier::Antichain;
193 ///
194 /// let mut frontier = Antichain::<u32>::new();
195 ///```
196 pub fn new() -> Antichain<T> { Antichain { elements: SmallVec::new() } }
197
198 /// Creates a new empty `Antichain` with space for `capacity` elements.
199 ///
200 /// # Examples
201 ///
202 ///```
203 /// use timely::progress::frontier::Antichain;
204 ///
205 /// let mut frontier = Antichain::<u32>::with_capacity(10);
206 ///```
207 pub fn with_capacity(capacity: usize) -> Self {
208 Self {
209 elements: SmallVec::with_capacity(capacity),
210 }
211 }
212
213 /// Creates a new singleton `Antichain`.
214 ///
215 /// # Examples
216 ///
217 ///```
218 /// use timely::progress::frontier::Antichain;
219 ///
220 /// let mut frontier = Antichain::from_elem(2);
221 ///```
222 pub fn from_elem(element: T) -> Antichain<T> {
223 let mut elements = SmallVec::with_capacity(1);
224 elements.push(element);
225 Antichain { elements }
226 }
227
228 /// Clears the contents of the antichain.
229 ///
230 /// # Examples
231 ///
232 ///```
233 /// use timely::progress::frontier::Antichain;
234 ///
235 /// let mut frontier = Antichain::from_elem(2);
236 /// frontier.clear();
237 /// assert!(frontier.elements().is_empty());
238 ///```
239 pub fn clear(&mut self) { self.elements.clear() }
240
241 /// Drains the elements, leaving the allocation for reuse.
242 pub fn drain(&mut self) -> smallvec::Drain<'_, [T; 1]> { self.elements.drain(..) }
243
244 /// Sorts the elements so that comparisons between antichains can be made.
245 pub fn sort(&mut self) where T: Ord { self.elements.sort() }
246
247 /// Reveals the elements in the antichain.
248 ///
249 /// This method is redundant with `<Antichain<T> as Deref>`, but the method
250 /// is in such broad use that we probably don't want to deprecate it without
251 /// some time to fix all things.
252 ///
253 /// # Examples
254 ///
255 ///```
256 /// use timely::progress::frontier::Antichain;
257 ///
258 /// let mut frontier = Antichain::from_elem(2);
259 /// assert_eq!(frontier.elements(), &[2]);
260 ///```
261 #[inline] pub fn elements(&self) -> &[T] { &self[..] }
262
263 /// Reveals the elements in the antichain.
264 ///
265 /// # Examples
266 ///
267 ///```
268 /// use timely::progress::frontier::Antichain;
269 ///
270 /// let mut frontier = Antichain::from_elem(2);
271 /// assert_eq!(&*frontier.borrow(), &[2]);
272 ///```
273 #[inline] pub fn borrow(&self) -> AntichainRef<'_, T> { AntichainRef::new(&self.elements) }}
274
275impl<T: PartialEq> PartialEq for Antichain<T> {
276 fn eq(&self, other: &Self) -> bool {
277 // Lengths should be the same, with the option for fast acceptance if identical.
278 self.elements().len() == other.elements().len() &&
279 (
280 self.elements().iter().zip(other.elements().iter()).all(|(t1,t2)| t1 == t2) ||
281 self.elements().iter().all(|t1| other.elements().iter().any(|t2| t1.eq(t2)))
282 )
283 }
284}
285
286impl<T: Eq> Eq for Antichain<T> { }
287
288impl<T: PartialOrder> PartialOrder for Antichain<T> {
289 fn less_equal(&self, other: &Self) -> bool {
290 other.elements().iter().all(|t2| self.elements().iter().any(|t1| t1.less_equal(t2)))
291 }
292}
293
294impl<T: Clone> Clone for Antichain<T> {
295 fn clone(&self) -> Self {
296 Antichain { elements: self.elements.clone() }
297 }
298 fn clone_from(&mut self, source: &Self) {
299 self.elements.clone_from(&source.elements)
300 }
301}
302
303impl<T> Default for Antichain<T> {
304 fn default() -> Self {
305 Self::new()
306 }
307}
308
309impl<T: TotalOrder> TotalOrder for Antichain<T> { }
310
311impl<T: TotalOrder> Antichain<T> {
312 /// Convert to the at most one element the antichain contains.
313 pub fn into_option(mut self) -> Option<T> {
314 debug_assert!(self.len() <= 1);
315 self.elements.pop()
316 }
317 /// Return a reference to the at most one element the antichain contains.
318 pub fn as_option(&self) -> Option<&T> {
319 debug_assert!(self.len() <= 1);
320 self.elements.last()
321 }
322}
323
324impl<T: Ord+std::hash::Hash> std::hash::Hash for Antichain<T> {
325 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
326 let mut temp = self.elements.iter().collect::<Vec<_>>();
327 temp.sort();
328 for element in temp {
329 element.hash(state);
330 }
331 }
332}
333
334impl<T: PartialOrder> From<Vec<T>> for Antichain<T> {
335 fn from(vec: Vec<T>) -> Self {
336 // TODO: We could reuse `vec` with some care.
337 let mut temp = Antichain::new();
338 for elem in vec.into_iter() { temp.insert(elem); }
339 temp
340 }
341}
342
343impl<T> From<Antichain<T>> for SmallVec<[T; 1]> {
344 fn from(val: Antichain<T>) -> Self {
345 val.elements
346 }
347}
348
349impl<T> ::std::ops::Deref for Antichain<T> {
350 type Target = [T];
351 fn deref(&self) -> &Self::Target {
352 &self.elements
353 }
354}
355
356impl<T> ::std::iter::IntoIterator for Antichain<T> {
357 type Item = T;
358 type IntoIter = smallvec::IntoIter<[T; 1]>;
359 fn into_iter(self) -> Self::IntoIter {
360 self.elements.into_iter()
361 }
362}
363
364/// An antichain based on a multiset whose elements frequencies can be updated.
365///
366/// The `MutableAntichain` maintains frequencies for many elements of type `T`, and exposes the set
367/// of elements with positive count not greater than any other elements with positive count. The
368/// antichain may both advance and retreat; the changes do not all need to be to elements greater or
369/// equal to some elements of the frontier.
370///
371/// The type `T` must implement `PartialOrder` as well as `Ord`. The implementation of the `Ord` trait
372/// is used to efficiently organize the updates for cancellation, and to efficiently determine the lower
373/// bounds, and only needs to not contradict the `PartialOrder` implementation (that is, if `PartialOrder`
374/// orders two elements, then so does the `Ord` implementation).
375///
376/// The `MutableAntichain` implementation is done with the intent that updates to it are done in batches,
377/// and it is acceptable to rebuild the frontier from scratch when a batch of updates change it. This means
378/// that it can be expensive to maintain a large number of counts and change few elements near the frontier.
379#[derive(Clone, Debug, Serialize, Deserialize)]
380pub struct MutableAntichain<T> {
381 updates: ChangeBatch<T>,
382 frontier: Vec<T>,
383 changes: ChangeBatch<T>,
384}
385
386impl<T> MutableAntichain<T> {
387 /// Creates a new empty `MutableAntichain`.
388 ///
389 /// # Examples
390 ///
391 ///```
392 /// use timely::progress::frontier::MutableAntichain;
393 ///
394 /// let frontier = MutableAntichain::<usize>::new();
395 /// assert!(frontier.is_empty());
396 ///```
397 #[inline]
398 pub fn new() -> MutableAntichain<T> {
399 MutableAntichain {
400 updates: ChangeBatch::new(),
401 frontier: Vec::new(),
402 changes: ChangeBatch::new(),
403 }
404 }
405
406 /// Removes all elements.
407 ///
408 /// # Examples
409 ///
410 ///```
411 /// use timely::progress::frontier::MutableAntichain;
412 ///
413 /// let mut frontier = MutableAntichain::<usize>::new();
414 /// frontier.clear();
415 /// assert!(frontier.is_empty());
416 ///```
417 #[inline]
418 pub fn clear(&mut self) {
419 self.updates.clear();
420 self.frontier.clear();
421 self.changes.clear();
422 }
423
424 /// Reveals the minimal elements with positive count.
425 ///
426 /// # Examples
427 ///
428 ///```
429 /// use timely::progress::frontier::MutableAntichain;
430 ///
431 /// let mut frontier = MutableAntichain::<usize>::new();
432 /// assert!(frontier.frontier().len() == 0);
433 ///```
434 #[inline]
435 pub fn frontier(&self) -> AntichainRef<'_, T> {
436 AntichainRef::new(&self.frontier)
437 }
438
439 /// Creates a new singleton `MutableAntichain`.
440 ///
441 /// # Examples
442 ///
443 ///```
444 /// use timely::progress::frontier::{AntichainRef, MutableAntichain};
445 ///
446 /// let mut frontier = MutableAntichain::from_elem(0u64);
447 /// assert!(frontier.frontier() == AntichainRef::new(&[0u64]));
448 ///```
449 #[inline]
450 pub fn from_elem(bottom: T) -> MutableAntichain<T>
451 where
452 T: Ord+Clone,
453 {
454 MutableAntichain {
455 updates: ChangeBatch::new_from(bottom.clone(), 1),
456 frontier: vec![bottom],
457 changes: ChangeBatch::new(),
458 }
459 }
460
461 /// Returns `true` if there are no elements in the `MutableAntichain`.
462 ///
463 /// # Examples
464 ///
465 ///```
466 /// use timely::progress::frontier::MutableAntichain;
467 ///
468 /// let mut frontier = MutableAntichain::<usize>::new();
469 /// assert!(frontier.is_empty());
470 ///```
471 #[inline]
472 pub fn is_empty(&self) -> bool {
473 self.frontier.is_empty()
474 }
475
476 /// Returns `true` if any item in the `MutableAntichain` is strictly less than the argument.
477 ///
478 /// # Examples
479 ///
480 ///```
481 /// use timely::progress::frontier::MutableAntichain;
482 ///
483 /// let mut frontier = MutableAntichain::from_elem(1u64);
484 /// assert!(!frontier.less_than(&0));
485 /// assert!(!frontier.less_than(&1));
486 /// assert!(frontier.less_than(&2));
487 ///```
488 #[inline]
489 pub fn less_than<O>(&self, time: &O) -> bool
490 where
491 T: PartialOrder<O>,
492 {
493 self.frontier().less_than(time)
494 }
495
496 /// Returns `true` if any item in the `MutableAntichain` is less than or equal to the argument.
497 ///
498 /// # Examples
499 ///
500 ///```
501 /// use timely::progress::frontier::MutableAntichain;
502 ///
503 /// let mut frontier = MutableAntichain::from_elem(1u64);
504 /// assert!(!frontier.less_equal(&0));
505 /// assert!(frontier.less_equal(&1));
506 /// assert!(frontier.less_equal(&2));
507 ///```
508 #[inline]
509 pub fn less_equal<O>(&self, time: &O) -> bool
510 where
511 T: PartialOrder<O>,
512 {
513 self.frontier().less_equal(time)
514 }
515
516 /// Applies updates to the antichain and enumerates any changes.
517 ///
518 /// # Examples
519 ///
520 ///```
521 /// use timely::progress::frontier::{AntichainRef, MutableAntichain};
522 ///
523 /// let mut frontier = MutableAntichain::from_elem(1u64);
524 /// let changes =
525 /// frontier
526 /// .update_iter(vec![(1, -1), (2, 7)])
527 /// .collect::<Vec<_>>();
528 ///
529 /// assert!(frontier.frontier() == AntichainRef::new(&[2]));
530 /// assert!(changes == vec![(1, -1), (2, 1)]);
531 ///```
532 #[inline]
533 pub fn update_iter<I>(&mut self, updates: I) -> smallvec::Drain<'_, [(T, i64); 2]>
534 where
535 T: Clone + PartialOrder + Ord,
536 I: IntoIterator<Item = (T, i64)>,
537 {
538 // track whether a rebuild is needed.
539 let mut rebuild_required = false;
540 for (time, delta) in updates {
541 // If we do not yet require a rebuild, test whether we might require one
542 // and set the flag in that case.
543 if !rebuild_required {
544 rebuild_required = self.requires_rebuild(&time, delta);
545 }
546
547 self.updates.update(time, delta);
548 }
549
550 if rebuild_required {
551 self.rebuild()
552 }
553 self.changes.drain()
554 }
555
556 /// Tests whether applying `(time, delta)` will require a frontier rebuild.
557 ///
558 /// Factored out of [`Self::update_iter`] so it is generic only over `T` and not
559 /// the iterator type, deduplicating the inlined `frontier.iter().any(...)` bodies
560 /// across `update_iter` monomorphizations.
561 fn requires_rebuild(&self, time: &T, delta: i64) -> bool
562 where
563 T: PartialOrder,
564 {
565 // Single-pass `for` loop (instead of two `Iterator::any` calls) avoids
566 // monomorphizing `slice::Iter::any` over per-call-site closure types and
567 // traverses `self.frontier` at most once.
568 let mut beyond_frontier = false;
569 let mut before_frontier = true;
570 for f in &self.frontier {
571 if !beyond_frontier && f.less_than(time) {
572 beyond_frontier = true;
573 }
574 if before_frontier && f.less_equal(time) {
575 before_frontier = false;
576 }
577 if beyond_frontier && !before_frontier {
578 break;
579 }
580 }
581 !(beyond_frontier || (delta < 0 && before_frontier))
582 }
583
584 /// Rebuilds `self.frontier` from `self.updates`.
585 ///
586 /// This method is meant to be used for bulk updates to the frontier, and does more work than one might do
587 /// for single updates, but is meant to be an efficient way to process multiple updates together. This is
588 /// especially true when we want to apply very large numbers of updates.
589 fn rebuild(&mut self)
590 where
591 T: Clone + PartialOrder + Ord,
592 {
593 for time in self.frontier.drain(..) {
594 self.changes.update(time, -1);
595 }
596
597 // build new frontier using strictly positive times.
598 // as the times are sorted, we don't need to worry that we might displace frontier elements.
599 for time in self.updates.iter().filter(|x| x.1 > 0) {
600 if !self.frontier.iter().any(|f| f.less_equal(&time.0)) {
601 self.frontier.push(time.0.clone());
602 }
603 }
604
605 for time in self.frontier.iter() {
606 self.changes.update(time.clone(), 1);
607 }
608 }
609
610 /// Reports the count for a queried time.
611 pub fn count_for<O>(&self, query_time: &O) -> i64
612 where
613 T: PartialEq<O>,
614 {
615 self.updates
616 .unstable_internal_updates()
617 .iter()
618 .filter(|td| td.0.eq(query_time))
619 .map(|td| td.1)
620 .sum()
621 }
622
623 /// Reports the updates that form the frontier. Returns an iterator of timestamps and their frequency.
624 ///
625 /// Rebuilds the internal representation before revealing times and frequencies.
626 pub fn updates(&mut self) -> impl Iterator<Item=&(T, i64)>
627 where
628 T: Clone + PartialOrder + Ord,
629 {
630 self.rebuild();
631 self.updates.iter()
632 }
633}
634
635impl<T> Default for MutableAntichain<T> {
636 fn default() -> Self {
637 Self::new()
638 }
639}
640
641/// Extension trait for filtering time changes through antichains.
642pub trait MutableAntichainFilter<T: PartialOrder+Ord+Clone> {
643 /// Filters time changes through an antichain.
644 ///
645 /// # Examples
646 ///
647 /// ```
648 /// use timely::progress::frontier::{MutableAntichain, MutableAntichainFilter};
649 ///
650 /// let mut frontier = MutableAntichain::from_elem(1u64);
651 /// let changes =
652 /// vec![(1, -1), (2, 7)]
653 /// .filter_through(&mut frontier)
654 /// .collect::<Vec<_>>();
655 ///
656 /// assert!(changes == vec![(1, -1), (2, 1)]);
657 /// ```
658 fn filter_through(self, antichain: &mut MutableAntichain<T>) -> smallvec::Drain<'_, [(T,i64); 2]>;
659}
660
661impl<T: PartialOrder+Ord+Clone, I: IntoIterator<Item=(T,i64)>> MutableAntichainFilter<T> for I {
662 fn filter_through(self, antichain: &mut MutableAntichain<T>) -> smallvec::Drain<'_, [(T,i64); 2]> {
663 antichain.update_iter(self)
664 }
665}
666
667impl<T: PartialOrder+Ord+Clone> From<Antichain<T>> for MutableAntichain<T> {
668 fn from(antichain: Antichain<T>) -> Self {
669 let mut result = MutableAntichain::new();
670 result.update_iter(antichain.into_iter().map(|time| (time, 1)));
671 result
672 }
673}
674impl<'a, T: PartialOrder+Ord+Clone> From<AntichainRef<'a, T>> for MutableAntichain<T> {
675 fn from(antichain: AntichainRef<'a, T>) -> Self {
676 let mut result = MutableAntichain::new();
677 result.update_iter(antichain.into_iter().map(|time| (time.clone(), 1)));
678 result
679 }
680}
681
682impl<T> std::iter::FromIterator<(T, i64)> for MutableAntichain<T>
683where
684 T: Clone + PartialOrder + Ord,
685{
686 fn from_iter<I>(iterator: I) -> Self
687 where
688 I: IntoIterator<Item=(T, i64)>,
689 {
690 let mut result = Self::new();
691 result.update_iter(iterator);
692 result
693 }
694}
695
696/// A wrapper for elements of an antichain.
697#[derive(Debug)]
698pub struct AntichainRef<'a, T: 'a> {
699 /// Elements contained in the antichain.
700 frontier: &'a [T],
701}
702
703impl<'a, T: 'a> Clone for AntichainRef<'a, T> {
704 fn clone(&self) -> Self { *self }
705}
706
707impl<'a, T: 'a> Copy for AntichainRef<'a, T> { }
708
709impl<'a, T: 'a> AntichainRef<'a, T> {
710 /// Create a new `AntichainRef` from a reference to a slice of elements forming the frontier.
711 ///
712 /// This method does not check that this antichain has any particular properties, for example
713 /// that there are no elements strictly less than other elements.
714 pub fn new(frontier: &'a [T]) -> Self {
715 Self {
716 frontier,
717 }
718 }
719
720 /// Constructs an owned antichain from the antichain reference.
721 ///
722 /// # Examples
723 ///
724 ///```
725 /// use timely::progress::{Antichain, frontier::AntichainRef};
726 ///
727 /// let frontier = AntichainRef::new(&[1u64]);
728 /// assert_eq!(frontier.to_owned(), Antichain::from_elem(1u64));
729 ///```
730 pub fn to_owned(&self) -> Antichain<T> where T: Clone {
731 Antichain {
732 elements: self.frontier.into()
733 }
734 }
735}
736
737impl<T> AntichainRef<'_, T> {
738
739 /// Returns `true` if any item in the `AntichainRef` is strictly less than the argument.
740 ///
741 /// # Examples
742 ///
743 ///```
744 /// use timely::progress::frontier::AntichainRef;
745 ///
746 /// let frontier = AntichainRef::new(&[1u64]);
747 /// assert!(!frontier.less_than(&0));
748 /// assert!(!frontier.less_than(&1));
749 /// assert!(frontier.less_than(&2));
750 ///```
751 #[inline]
752 pub fn less_than<O>(&self, time: &O) -> bool where T: PartialOrder<O> {
753 self.iter().any(|x| x.less_than(time))
754 }
755
756 /// Returns `true` if any item in the `AntichainRef` is less than or equal to the argument.
757 #[inline]
758 ///
759 /// # Examples
760 ///
761 ///```
762 /// use timely::progress::frontier::AntichainRef;
763 ///
764 /// let frontier = AntichainRef::new(&[1u64]);
765 /// assert!(!frontier.less_equal(&0));
766 /// assert!(frontier.less_equal(&1));
767 /// assert!(frontier.less_equal(&2));
768 ///```
769 pub fn less_equal<O>(&self, time: &O) -> bool where T: PartialOrder<O> {
770 self.iter().any(|x| x.less_equal(time))
771 }
772}
773
774impl<T: PartialEq> PartialEq for AntichainRef<'_, T> {
775 fn eq(&self, other: &Self) -> bool {
776 // Lengths should be the same, with the option for fast acceptance if identical.
777 self.len() == other.len() &&
778 (
779 self.iter().zip(other.iter()).all(|(t1,t2)| t1 == t2) ||
780 self.iter().all(|t1| other.iter().any(|t2| t1.eq(t2)))
781 )
782 }
783}
784
785impl<T: Eq> Eq for AntichainRef<'_, T> { }
786
787impl<T: PartialOrder> PartialOrder for AntichainRef<'_, T> {
788 fn less_equal(&self, other: &Self) -> bool {
789 other.iter().all(|t2| self.iter().any(|t1| t1.less_equal(t2)))
790 }
791}
792
793impl<T: TotalOrder> TotalOrder for AntichainRef<'_, T> { }
794
795impl<T: TotalOrder> AntichainRef<'_, T> {
796 /// Return a reference to the at most one element the antichain contains.
797 pub fn as_option(&self) -> Option<&T> {
798 debug_assert!(self.len() <= 1);
799 self.frontier.last()
800 }
801}
802
803impl<T> ::std::ops::Deref for AntichainRef<'_, T> {
804 type Target = [T];
805 fn deref(&self) -> &Self::Target {
806 self.frontier
807 }
808}
809
810impl<'a, T: 'a> ::std::iter::IntoIterator for &'a AntichainRef<'a, T> {
811 type Item = &'a T;
812 type IntoIter = ::std::slice::Iter<'a, T>;
813 fn into_iter(self) -> Self::IntoIter {
814 self.iter()
815 }
816}
817
818#[cfg(test)]
819mod tests {
820 use std::collections::HashSet;
821
822 use super::*;
823
824 #[derive(PartialEq, Eq, PartialOrd, Ord, Hash)]
825 struct Elem(char, usize);
826
827 impl PartialOrder for Elem {
828 fn less_equal(&self, other: &Self) -> bool {
829 self.0 <= other.0 && self.1 <= other.1
830 }
831 }
832
833 #[test]
834 fn antichain_hash() {
835 let mut hashed = HashSet::new();
836 hashed.insert(Antichain::from(vec![Elem('a', 2), Elem('b', 1)]));
837
838 assert!(hashed.contains(&Antichain::from(vec![Elem('a', 2), Elem('b', 1)])));
839 assert!(hashed.contains(&Antichain::from(vec![Elem('b', 1), Elem('a', 2)])));
840
841 assert!(!hashed.contains(&Antichain::from(vec![Elem('a', 2)])));
842 assert!(!hashed.contains(&Antichain::from(vec![Elem('a', 1)])));
843 assert!(!hashed.contains(&Antichain::from(vec![Elem('b', 2)])));
844 assert!(!hashed.contains(&Antichain::from(vec![Elem('a', 1), Elem('b', 2)])));
845 assert!(!hashed.contains(&Antichain::from(vec![Elem('c', 3)])));
846 assert!(!hashed.contains(&Antichain::from(vec![])));
847 }
848
849 #[test]
850 fn mutable_compaction() {
851 let mut mutable = MutableAntichain::new();
852 mutable.update_iter(Some((7, 1)));
853 mutable.update_iter(Some((7, 1)));
854 mutable.update_iter(Some((7, 1)));
855 mutable.update_iter(Some((7, 1)));
856 mutable.update_iter(Some((7, 1)));
857 mutable.update_iter(Some((7, 1)));
858 mutable.update_iter(Some((8, 1)));
859 mutable.update_iter(Some((8, 1)));
860 mutable.update_iter(Some((8, 1)));
861 mutable.update_iter(Some((8, 1)));
862 mutable.update_iter(Some((8, 1)));
863 for _ in 0 .. 1000 {
864 mutable.update_iter(Some((9, 1)));
865 mutable.update_iter(Some((9, -1)));
866 }
867 assert!(mutable.updates.unstable_internal_updates().len() <= 32);
868 }
869}