timely/dataflow/operators/core/
exchange.rs

1//! Exchange records between workers.
2
3use crate::Container;
4use crate::progress::Timestamp;
5use crate::container::{DrainContainer, SizableContainer, PushInto};
6use crate::dataflow::channels::pact::ExchangeCore;
7use crate::dataflow::operators::generic::operator::Operator;
8use crate::dataflow::Stream;
9
10/// Exchange records between workers.
11pub trait Exchange<C: DrainContainer> {
12    /// Exchange records between workers.
13    ///
14    /// The closure supplied should map a reference to a record to a `u64`,
15    /// whose value determines to which worker the record will be routed.
16    ///
17    /// # Examples
18    /// ```
19    /// use timely::dataflow::operators::{ToStream, Exchange, Inspect};
20    ///
21    /// timely::example(|scope| {
22    ///     (0..10).to_stream(scope)
23    ///            .container::<Vec<_>>()
24    ///            .exchange(|x| *x)
25    ///            .inspect(|x| println!("seen: {:?}", x));
26    /// });
27    /// ```
28    fn exchange<F>(self, route: F) -> Self
29    where
30        for<'a> F: FnMut(&C::Item<'a>) -> u64 + 'static;
31}
32
33impl<T: Timestamp, C> Exchange<C> for Stream<'_, T, C>
34where
35    C: Container
36        + SizableContainer
37        + DrainContainer
38        + Send
39        + crate::dataflow::channels::ContainerBytes
40        + for<'a> PushInto<C::Item<'a>>,
41{
42    fn exchange<F>(self, route: F) -> Self
43    where
44        for<'a> F: FnMut(&C::Item<'a>) -> u64 + 'static,
45    {
46        self.unary(ExchangeCore::new(route), "Exchange", |_, _| {
47            move |input, output| {
48                input.for_each_time(|time, data| {
49                    output.session(&time).give_containers(data);
50                });
51            }
52        })
53    }
54}
timely/dataflow/operators/core/exchange.rs

timely/dataflow/operators/core/
exchange.rs
