timely/progress/
mod.rs

1//! Progress tracking mechanisms to support notification in timely dataflow
2
3use columnar::Columnar;
4use serde::{Deserialize, Serialize};
5pub use self::operate::Operate;
6pub use self::subgraph::{Subgraph, SubgraphBuilder};
7pub use self::timestamp::{Timestamp, PathSummary};
8pub use self::change_batch::ChangeBatch;
9pub use self::frontier::Antichain;
10pub use self::stamp::Stamp;
11
12pub mod change_batch;
13pub mod stamp;
14pub mod frontier;
15pub mod timestamp;
16pub mod operate;
17pub mod broadcast;
18pub mod reachability;
19pub mod subgraph;
20
21/// A timely dataflow location.
22#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Serialize, Deserialize, Columnar)]
23pub struct Location {
24    /// A scope-local operator identifier.
25    pub node: usize,
26    /// An operator port identifier.
27    pub port: Port,
28}
29
30impl Location {
31    /// Creates a new target location (operator input or scope output).
32    pub fn new_target(node: usize, port: usize) -> Location {
33        Location { node, port: Port::Target(port) }
34    }
35    /// Creates a new source location (operator output or scope input).
36    pub fn new_source(node: usize, port: usize) -> Location {
37        Location { node, port: Port::Source(port) }
38    }
39    /// If the location is a target.
40    pub fn is_target(&self) -> bool { matches!(self.port, Port::Target(_)) }
41    /// If the location is a source.
42    pub fn is_source(&self) -> bool { matches!(self.port, Port::Source(_)) }
43}
44
45impl From<Target> for Location {
46    fn from(target: Target) -> Self {
47        Location {
48            node: target.node,
49            port: Port::Target(target.port),
50        }
51    }
52}
53
54impl From<Source> for Location {
55    fn from(source: Source) -> Self {
56        Location {
57            node: source.node,
58            port: Port::Source(source.port),
59        }
60    }
61}
62
63/// An operator port.
64#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Serialize, Deserialize, Columnar)]
65pub enum Port {
66    /// An operator input.
67    Target(usize),
68    /// An operator output.
69    Source(usize),
70}
71
72/// Names a source of a data stream.
73///
74/// A source of data is either a child output, or an input from a parent.
75/// Conventionally, `index` zero is used for parent input.
76#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug)]
77pub struct Source {
78    /// Index of the source operator.
79    pub node: usize,
80    /// Number of the output port from the operator.
81    pub port: usize,
82}
83
84impl Source {
85    /// Creates a new source from node and port identifiers.
86    pub fn new(node: usize, port: usize) -> Self {
87        Self { node, port }
88    }
89}
90
91/// Names a target of a data stream.
92///
93/// A target of data is either a child input, or an output to a parent.
94/// Conventionally, `index` zero is used for parent output.
95#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug)]
96pub struct Target {
97    /// Index of the target operator.
98    pub node: usize,
99    /// Number of the input port to the operator.
100    pub port: usize,
101}
102
103impl Target {
104    /// Creates a new target from node and port identifiers.
105    pub fn new(node: usize, port: usize) -> Self {
106        Self { node, port }
107    }
108}
timely/progress/mod.rs

timely/progress/
mod.rs
