1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151
//! A partially ordered measure of progress at each timely dataflow location.
use std::fmt::Debug;
use std::any::Any;
use std::default::Default;
use std::hash::Hash;
use crate::communication::Data;
use crate::order::PartialOrder;
/// A composite trait for types that serve as timestamps in timely dataflow.
pub trait Timestamp: Clone+Eq+PartialOrder+Debug+Send+Any+Data+Hash+Ord {
/// A type summarizing action on a timestamp along a dataflow path.
type Summary : PathSummary<Self> + 'static;
/// A minimum value suitable as a default.
fn minimum() -> Self;
}
/// A summary of how a timestamp advances along a timely dataflow path.
pub trait PathSummary<T> : Clone+'static+Eq+PartialOrder+Debug+Default {
/// Advances a timestamp according to the timestamp actions on the path.
///
/// The path may advance the timestamp sufficiently that it is no longer valid, for example if
/// incrementing fields would result in integer overflow. In this case, `results_in` should
/// return `None`.
///
/// The `feedback` operator, apparently the only point where timestamps are actually incremented
/// in computation, uses this method and will drop messages with timestamps that when advanced
/// result in `None`. Ideally, all other timestamp manipulation should behave similarly.
///
/// # Examples
/// ```
/// use timely::progress::timestamp::PathSummary;
///
/// let timestamp = 3;
///
/// let summary1 = 5;
/// let summary2 = usize::max_value() - 2;
///
/// assert_eq!(summary1.results_in(×tamp), Some(8));
/// assert_eq!(summary2.results_in(×tamp), None);
/// ```
fn results_in(&self, src: &T) -> Option<T>;
/// Composes this path summary with another path summary.
///
/// It is possible that the two composed paths result in an invalid summary, for example when
/// integer additions overflow. If it is correct that all timestamps moved along these paths
/// would also result in overflow and be discarded, `followed_by` can return `None. It is very
/// important that this not be used casually, as this does not prevent the actual movement of
/// data.
///
/// # Examples
/// ```
/// use timely::progress::timestamp::PathSummary;
///
/// let summary1 = 5;
/// let summary2 = usize::max_value() - 3;
///
/// assert_eq!(summary1.followed_by(&summary2), None);
/// ```
fn followed_by(&self, other: &Self) -> Option<Self>;
}
impl Timestamp for () { type Summary = (); fn minimum() -> Self { () }}
impl PathSummary<()> for () {
#[inline] fn results_in(&self, _src: &()) -> Option<()> { Some(()) }
#[inline] fn followed_by(&self, _other: &()) -> Option<()> { Some(()) }
}
/// Implements Timestamp and PathSummary for types with a `checked_add` method.
macro_rules! implement_timestamp_add {
($($index_type:ty,)*) => (
$(
impl Timestamp for $index_type {
type Summary = $index_type;
fn minimum() -> Self { Self::min_value() }
}
impl PathSummary<$index_type> for $index_type {
#[inline]
fn results_in(&self, src: &$index_type) -> Option<$index_type> { self.checked_add(*src) }
#[inline]
fn followed_by(&self, other: &$index_type) -> Option<$index_type> { self.checked_add(*other) }
}
)*
)
}
implement_timestamp_add!(usize, u128, u64, u32, u16, u8, isize, i128, i64, i32, i16, i8,);
impl Timestamp for ::std::time::Duration {
type Summary = ::std::time::Duration;
fn minimum() -> Self { ::std::time::Duration::new(0, 0) }
}
impl PathSummary<::std::time::Duration> for ::std::time::Duration {
#[inline]
fn results_in(&self, src: &::std::time::Duration) -> Option<::std::time::Duration> { self.checked_add(*src) }
#[inline]
fn followed_by(&self, other: &::std::time::Duration) -> Option<::std::time::Duration> { self.checked_add(*other) }
}
pub use self::refines::Refines;
mod refines {
use crate::progress::Timestamp;
/// Conversion between pointstamp types.
///
/// This trait is central to nested scopes, for which the inner timestamp must be
/// related to the outer timestamp. These methods define those relationships.
///
/// It would be ideal to use Rust's From and Into traits, but they seem to be messed
/// up due to coherence: we can't implement `Into` because it induces a from implementation
/// we can't control.
pub trait Refines<T: Timestamp> : Timestamp {
/// Converts the outer timestamp to an inner timestamp.
fn to_inner(other: T) -> Self;
/// Converts the inner timestamp to an outer timestamp.
fn to_outer(self) -> T;
/// Summarizes an inner path summary as an outer path summary.
///
/// It is crucial for correctness that the result of this summarization's `results_in`
/// method is equivalent to `|time| path.results_in(time.to_inner()).to_outer()`, or
/// at least produces times less or equal to that result.
fn summarize(path: <Self as Timestamp>::Summary) -> <T as Timestamp>::Summary;
}
/// All types "refine" themselves,
impl<T: Timestamp> Refines<T> for T {
fn to_inner(other: T) -> T { other }
fn to_outer(self) -> T { self }
fn summarize(path: <T as Timestamp>::Summary) -> <T as Timestamp>::Summary { path }
}
/// Implements `Refines<()>` for most types.
///
/// We have a macro here because a blanket implement would conflict with the "refines self"
/// blanket implementation just above. Waiting on specialization to fix that, I guess.
macro_rules! implement_refines_empty {
($($index_type:ty,)*) => (
$(
impl Refines<()> for $index_type {
fn to_inner(_: ()) -> $index_type { Default::default() }
fn to_outer(self) -> () { () }
fn summarize(_: <$index_type as Timestamp>::Summary) -> () { () }
}
)*
)
}
implement_refines_empty!(usize, u128, u64, u32, u16, u8, isize, i128, i64, i32, i16, i8, ::std::time::Duration,);
}