timely/dataflow/scopes/
child.rs

1//! A child dataflow scope, used to build nested dataflow scopes.
2
3use std::rc::Rc;
4use std::cell::RefCell;
5
6use crate::communication::{Data, Push, Pull};
7use crate::communication::allocator::thread::{ThreadPusher, ThreadPuller};
8use crate::scheduling::Scheduler;
9use crate::scheduling::activate::Activations;
10use crate::progress::{Timestamp, Operate, SubgraphBuilder};
11use crate::progress::{Source, Target};
12use crate::progress::timestamp::Refines;
13use crate::order::Product;
14use crate::logging::TimelyLogger as Logger;
15use crate::logging::TimelyProgressLogger as ProgressLogger;
16use crate::worker::{AsWorker, Config};
17
18use super::{ScopeParent, Scope};
19
20/// Type alias for iterative child scope.
21pub type Iterative<'a, G, T> = Child<'a, G, Product<<G as ScopeParent>::Timestamp, T>>;
22
23/// A `Child` wraps a `Subgraph` and a parent `G: Scope`. It manages the addition
24/// of `Operate`s to a subgraph, and the connection of edges between them.
25pub struct Child<'a, G, T>
26where
27    G: ScopeParent,
28    T: Timestamp+Refines<G::Timestamp>
29{
30    /// The subgraph under assembly.
31    pub subgraph: &'a RefCell<SubgraphBuilder<G::Timestamp, T>>,
32    /// A copy of the child's parent scope.
33    pub parent:   G,
34    /// The log writer for this scope.
35    pub logging:  Option<Logger>,
36    /// The progress log writer for this scope.
37    pub progress_logging:  Option<ProgressLogger>,
38}
39
40impl<'a, G, T> Child<'a, G, T>
41where
42    G: ScopeParent,
43    T: Timestamp+Refines<G::Timestamp>
44{
45    /// This worker's unique identifier.
46    ///
47    /// Ranges from `0` to `self.peers() - 1`.
48    pub fn index(&self) -> usize { self.parent.index() }
49    /// The total number of workers in the computation.
50    pub fn peers(&self) -> usize { self.parent.peers() }
51}
52
53impl<'a, G, T> AsWorker for Child<'a, G, T>
54where
55    G: ScopeParent,
56    T: Timestamp+Refines<G::Timestamp>
57{
58    fn config(&self) -> &Config { self.parent.config() }
59    fn index(&self) -> usize { self.parent.index() }
60    fn peers(&self) -> usize { self.parent.peers() }
61    fn allocate<D: Data>(&mut self, identifier: usize, address: &[usize]) -> (Vec<Box<dyn Push<Message<D>>>>, Box<dyn Pull<Message<D>>>) {
62        self.parent.allocate(identifier, address)
63    }
64    fn pipeline<D: 'static>(&mut self, identifier: usize, address: &[usize]) -> (ThreadPusher<Message<D>>, ThreadPuller<Message<D>>) {
65        self.parent.pipeline(identifier, address)
66    }
67    fn new_identifier(&mut self) -> usize {
68        self.parent.new_identifier()
69    }
70    fn log_register(&self) -> ::std::cell::RefMut<crate::logging_core::Registry<crate::logging::WorkerIdentifier>> {
71        self.parent.log_register()
72    }
73}
74
75impl<'a, G, T> Scheduler for Child<'a, G, T>
76where
77    G: ScopeParent,
78    T: Timestamp+Refines<G::Timestamp>
79{
80    fn activations(&self) -> Rc<RefCell<Activations>> {
81        self.parent.activations()
82    }
83}
84
85impl<'a, G, T> ScopeParent for Child<'a, G, T>
86where
87    G: ScopeParent,
88    T: Timestamp+Refines<G::Timestamp>
89{
90    type Timestamp = T;
91}
92
93impl<'a, G, T> Scope for Child<'a, G, T>
94where
95    G: ScopeParent,
96    T: Timestamp+Refines<G::Timestamp>,
97{
98    fn name(&self) -> String { self.subgraph.borrow().name.clone() }
99    fn addr(&self) -> Vec<usize> { self.subgraph.borrow().path.clone() }
100    fn add_edge(&self, source: Source, target: Target) {
101        self.subgraph.borrow_mut().connect(source, target);
102    }
103
104    fn add_operator_with_indices(&mut self, operator: Box<dyn Operate<Self::Timestamp>>, local: usize, global: usize) {
105        self.subgraph.borrow_mut().add_child(operator, local, global);
106    }
107
108    fn allocate_operator_index(&mut self) -> usize {
109        self.subgraph.borrow_mut().allocate_child_id()
110    }
111
112    #[inline]
113    fn scoped<T2, R, F>(&mut self, name: &str, func: F) -> R
114    where
115        T2: Timestamp+Refines<T>,
116        F: FnOnce(&mut Child<Self, T2>) -> R,
117    {
118        let index = self.subgraph.borrow_mut().allocate_child_id();
119        let path = self.subgraph.borrow().path.clone();
120
121        let subscope = RefCell::new(SubgraphBuilder::new_from(index, path, self.logging(), self.progress_logging.clone(), name));
122        let result = {
123            let mut builder = Child {
124                subgraph: &subscope,
125                parent: self.clone(),
126                logging: self.logging.clone(),
127                progress_logging: self.progress_logging.clone(),
128            };
129            func(&mut builder)
130        };
131        let subscope = subscope.into_inner().build(self);
132
133        self.add_operator_with_index(Box::new(subscope), index);
134
135        result
136    }
137}
138
139use crate::communication::Message;
140
141impl<'a, G, T> Clone for Child<'a, G, T>
142where
143    G: ScopeParent,
144    T: Timestamp+Refines<G::Timestamp>
145{
146    fn clone(&self) -> Self {
147        Child {
148            subgraph: self.subgraph,
149            parent: self.parent.clone(),
150            logging: self.logging.clone(),
151            progress_logging: self.progress_logging.clone(),
152        }
153    }
154}
timely/dataflow/scopes/child.rs

timely/dataflow/scopes/
child.rs
