timely/dataflow/operators/core/
exchange.rs

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

timely/dataflow/operators/core/
exchange.rs
