timely/dataflow/scopes/
child.rs

Help
1//! A child dataflow scope, used to build nested dataflow scopes.
2
3use std::rc::Rc;
4use std::cell::RefCell;
5
6use crate::communication::{Exchangeable, 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<T>>,
38}
39
40impl<G, T> Child<'_, 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<G, T> AsWorker for Child<'_, 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: Exchangeable>(&mut self, identifier: usize, address: Rc<[usize]>) -> (Vec<Box<dyn Push<D>>>, Box<dyn Pull<D>>) {
62        self.parent.allocate(identifier, address)
63    }
64    fn pipeline<D: 'static>(&mut self, identifier: usize, address: Rc<[usize]>) -> (ThreadPusher<D>, ThreadPuller<D>) {
65        self.parent.pipeline(identifier, address)
66    }
67    fn broadcast<D: Exchangeable + Clone>(&mut self, identifier: usize, address: Rc<[usize]>) -> (Box<dyn Push<D>>, Box<dyn Pull<D>>) {
68        self.parent.broadcast(identifier, address)
69    }
70    fn new_identifier(&mut self) -> usize {
71        self.parent.new_identifier()
72    }
73    fn peek_identifier(&self) -> usize {
74        self.parent.peek_identifier()
75    }
76    fn log_register(&self) -> ::std::cell::RefMut<crate::logging_core::Registry> {
77        self.parent.log_register()
78    }
79}
80
81impl<G, T> Scheduler for Child<'_, G, T>
82where
83    G: ScopeParent,
84    T: Timestamp+Refines<G::Timestamp>
85{
86    fn activations(&self) -> Rc<RefCell<Activations>> {
87        self.parent.activations()
88    }
89}
90
91impl<G, T> ScopeParent for Child<'_, G, T>
92where
93    G: ScopeParent,
94    T: Timestamp+Refines<G::Timestamp>
95{
96    type Timestamp = T;
97}
98
99impl<G, T> Scope for Child<'_, G, T>
100where
101    G: ScopeParent,
102    T: Timestamp+Refines<G::Timestamp>,
103{
104    fn name(&self) -> String { self.subgraph.borrow().name.clone() }
105    fn addr(&self) -> Rc<[usize]> { Rc::clone(&self.subgraph.borrow().path) }
106
107    fn addr_for_child(&self, index: usize) -> Rc<[usize]> {
108        let path = &self.subgraph.borrow().path[..];
109        let mut addr = Vec::with_capacity(path.len() + 1);
110        addr.extend_from_slice(path);
111        addr.push(index);
112        addr.into()
113    }
114
115    fn add_edge(&self, source: Source, target: Target) {
116        self.subgraph.borrow_mut().connect(source, target);
117    }
118
119    fn add_operator_with_indices(&mut self, operator: Box<dyn Operate<Self::Timestamp>>, local: usize, global: usize) {
120        self.subgraph.borrow_mut().add_child(operator, local, global);
121    }
122
123    fn allocate_operator_index(&mut self) -> usize {
124        self.subgraph.borrow_mut().allocate_child_id()
125    }
126
127    #[inline]
128    fn scoped<T2, R, F>(&mut self, name: &str, func: F) -> R
129    where
130        T2: Timestamp+Refines<T>,
131        F: FnOnce(&mut Child<Self, T2>) -> R,
132    {
133        let index = self.subgraph.borrow_mut().allocate_child_id();
134        let identifier = self.new_identifier();
135        let path = self.addr_for_child(index);
136
137        let type_name = std::any::type_name::<T2>();
138        let progress_logging = self.log_register().get(&format!("timely/progress/{type_name}"));
139        let summary_logging = self.log_register().get(&format!("timely/summary/{type_name}"));
140
141        let subscope = RefCell::new(SubgraphBuilder::new_from(path, identifier, self.logging(), summary_logging, name));
142        let result = {
143            let mut builder = Child {
144                subgraph: &subscope,
145                parent: self.clone(),
146                logging: self.logging.clone(),
147                progress_logging,
148            };
149            func(&mut builder)
150        };
151        let subscope = subscope.into_inner().build(self);
152
153        self.add_operator_with_indices(Box::new(subscope), index, identifier);
154
155        result
156    }
157}
158
159impl<G, T> Clone for Child<'_, G, T>
160where
161    G: ScopeParent,
162    T: Timestamp+Refines<G::Timestamp>
163{
164    fn clone(&self) -> Self {
165        Child {
166            subgraph: self.subgraph,
167            parent: self.parent.clone(),
168            logging: self.logging.clone(),
169            progress_logging: self.progress_logging.clone(),
170        }
171    }
172}
timely/dataflow/scopes/child.rs

timely/dataflow/scopes/
child.rs
