timely/dataflow/channels/pushers/
exchange.rs

1//! The exchange pattern distributes pushed data between many target pushees.
2
3use crate::communication::Push;
4use crate::container::{ContainerBuilder, PushInto};
5use crate::dataflow::channels::Message;
6use crate::{Container, Data};
7
8// TODO : Software write combining
9/// Distributes records among target pushees according to a distribution function.
10pub struct Exchange<T, CB, P, H>
11where
12    CB: ContainerBuilder,
13    P: Push<Message<T, CB::Container>>,
14    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
15{
16    pushers: Vec<P>,
17    builders: Vec<CB>,
18    current: Option<T>,
19    hash_func: H,
20}
21
22impl<T: Clone, CB, P, H>  Exchange<T, CB, P, H>
23where
24    CB: ContainerBuilder,
25    P: Push<Message<T, CB::Container>>,
26    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
27{
28    /// Allocates a new `Exchange` from a supplied set of pushers and a distribution function.
29    pub fn new(pushers: Vec<P>, key: H) -> Exchange<T, CB, P, H> {
30        let builders = std::iter::repeat_with(Default::default).take(pushers.len()).collect();
31        Exchange {
32            pushers,
33            hash_func: key,
34            builders,
35            current: None,
36        }
37    }
38    #[inline]
39    fn flush(&mut self, index: usize) {
40        while let Some(container) = self.builders[index].finish() {
41            if let Some(ref time) = self.current {
42                Message::push_at(container, time.clone(), &mut self.pushers[index]);
43            }
44        }
45    }
46}
47
48impl<T: Eq+Data, CB, P, H> Push<Message<T, CB::Container>> for Exchange<T, CB, P, H>
49where
50    CB: ContainerBuilder,
51    CB: for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
52    P: Push<Message<T, CB::Container>>,
53    for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
54{
55    #[inline(never)]
56    fn push(&mut self, message: &mut Option<Message<T, CB::Container>>) {
57        // if only one pusher, no exchange
58        if self.pushers.len() == 1 {
59            self.pushers[0].push(message);
60        }
61        else if let Some(message) = message {
62
63            let time = &message.time;
64            let data = &mut message.data;
65
66            // if the time isn't right, flush everything.
67            if self.current.as_ref().is_some_and(|x| x != time) {
68                for index in 0..self.pushers.len() {
69                    self.flush(index);
70                }
71            }
72            self.current = Some(time.clone());
73
74            let hash_func = &mut self.hash_func;
75
76            // if the number of pushers is a power of two, use a mask
77            if self.pushers.len().is_power_of_two() {
78                let mask = (self.pushers.len() - 1) as u64;
79                CB::partition(data, &mut self.builders, |datum| ((hash_func)(datum) & mask) as usize);
80            }
81            // as a last resort, use mod (%)
82            else {
83                let num_pushers = self.pushers.len() as u64;
84                CB::partition(data, &mut self.builders, |datum| ((hash_func)(datum) % num_pushers) as usize);
85            }
86            for (buffer, pusher) in self.builders.iter_mut().zip(self.pushers.iter_mut()) {
87                while let Some(container) = buffer.extract() {
88                    Message::push_at(container, time.clone(), pusher);
89                }
90            }
91        }
92        else {
93            // flush
94            for index in 0..self.pushers.len() {
95                self.flush(index);
96                self.builders[index].relax();
97                self.pushers[index].push(&mut None);
98            }
99        }
100    }
101}
timely/dataflow/channels/pushers/exchange.rs

timely/dataflow/channels/pushers/
exchange.rs
