timely/dataflow/channels/
mod.rs

1//! Structured communication between timely dataflow operators.
2
3use serde::{Deserialize, Serialize};
4use crate::communication::Push;
5
6/// A collection of types that may be pushed at.
7pub mod pushers;
8/// A collection of types that may be pulled from.
9pub mod pullers;
10/// Parallelization contracts, describing how data must be exchanged between operators.
11pub mod pact;
12
13pub use crate::progress::Stamp;
14
15/// A serializable representation of timestamped data.
16#[derive(Clone)]
17pub struct Message<T, C> {
18    /// The multiset of timestamps affixed to the message.
19    ///
20    /// The message may only result in downstream work at times greater or equal
21    /// to some element of the stamp. An empty stamp makes no progress claims.
22    pub stamp: Stamp<T>,
23    /// The data in the message.
24    pub data: C,
25    /// The source worker.
26    pub from: usize,
27    /// A sequence number for this worker-to-worker stream.
28    pub seq: usize,
29}
30
31impl<T, C> Message<T, C> {
32    /// Default buffer size.
33    #[deprecated = "Use timely::buffer::default_capacity instead"]
34    pub fn default_length() -> usize {
35        crate::container::buffer::default_capacity::<C>()
36    }
37}
38
39impl<T, C> Message<T, C> {
40    /// Creates a new message instance from arguments.
41    ///
42    /// Zero values are installed for `from` and `seq`, and are meant to be populated by `LogPusher`.
43    pub fn new(stamp: Stamp<T>, data: C) -> Self {
44        Message { stamp, data, from: 0, seq: 0 }
45    }
46
47    /// Forms a message from borrowed parts, and replaces `buffer` with what is left by the `push` call.
48    /// If the pusher returns nothing, then `buffer` is set to the default for the container.
49    #[inline]
50    pub fn push_at<P: Push<Message<T, C>>>(buffer: &mut C, stamp: Stamp<T>, pusher: &mut P) where C: Default {
51
52        let data = ::std::mem::take(buffer);
53        let message = Message::new(stamp, data);
54        let mut bundle = Some(message);
55
56        pusher.push(&mut bundle);
57
58        if let Some(message) = bundle {
59            *buffer = message.data;
60        }
61    }
62}
63
64// Instructions for serialization of `Message`.
65//
66// Serialization of each field is meant to be `u64` aligned, so that each has the ability
67// to be decoded using safe transmutation, e.g. `bytemuck`.
68impl<T, C> crate::communication::Bytesable for Message<T, C>
69where
70    T: Serialize + for<'a> Deserialize<'a> + Ord,
71    C: ContainerBytes,
72{
73    fn from_bytes(mut bytes: crate::bytes::arc::Bytes) -> Self {
74        use byteorder::ReadBytesExt;
75        let mut slice = &bytes[..];
76        let from: usize = slice.read_u64::<byteorder::LittleEndian>().unwrap().try_into().unwrap();
77        let seq: usize = slice.read_u64::<byteorder::LittleEndian>().unwrap().try_into().unwrap();
78        let stamp: Stamp<T> = ::bincode::deserialize_from(&mut slice).expect("bincode::deserialize() failed");
79        let time_size = ::bincode::serialized_size(&stamp).expect("bincode::serialized_size() failed") as usize;
80        // We expect to find the `data` payload at `8 + 8 + round_up(time_size)`;
81        let bytes_read = 8 + 8 + ((time_size + 7) & !7);
82        bytes.extract_to(bytes_read);
83        let data: C = ContainerBytes::from_bytes(bytes);
84        Self { stamp, data, from, seq }
85    }
86
87    fn length_in_bytes(&self) -> usize {
88        let time_size = ::bincode::serialized_size(&self.stamp).expect("bincode::serialized_size() failed") as usize;
89        // 16 comes from the two `u64` fields: `from` and `seq`.
90        16 + ((time_size + 7) & !7) + self.data.length_in_bytes()
91    }
92
93    fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
94        use byteorder::WriteBytesExt;
95        writer.write_u64::<byteorder::LittleEndian>(self.from.try_into().unwrap()).unwrap();
96        writer.write_u64::<byteorder::LittleEndian>(self.seq.try_into().unwrap()).unwrap();
97        ::bincode::serialize_into(&mut *writer, &self.stamp).expect("bincode::serialize_into() failed");
98        let time_size = ::bincode::serialized_size(&self.stamp).expect("bincode::serialized_size() failed") as usize;
99        let time_slop = ((time_size + 7) & !7) - time_size;
100        writer.write_all(&[0u8; 8][..time_slop]).unwrap();
101        self.data.into_bytes(&mut *writer);
102    }
103}
104
105
106/// A container-oriented version of `Bytesable` that can be implemented here for `Vec<T>` and other containers.
107pub trait ContainerBytes {
108    /// Wrap bytes as `Self`.
109    fn from_bytes(bytes: crate::bytes::arc::Bytes) -> Self;
110
111    /// The number of bytes required to serialize the data.
112    fn length_in_bytes(&self) -> usize;
113
114    /// Writes the binary representation into `writer`.
115    fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W);
116}
117
118mod implementations {
119
120    use std::io::Write;
121
122    use serde::{Serialize, Deserialize};
123    use crate::dataflow::channels::ContainerBytes;
124
125    impl<T: Serialize + for<'a> Deserialize<'a>> ContainerBytes for Vec<T> {
126        fn from_bytes(bytes: crate::bytes::arc::Bytes) -> Self {
127            ::bincode::deserialize(&bytes[..]).expect("bincode::deserialize() failed")
128        }
129
130        fn length_in_bytes(&self) -> usize {
131            let length = ::bincode::serialized_size(&self).expect("bincode::serialized_size() failed") as usize;
132            (length + 7) & !7
133        }
134
135        fn into_bytes<W: Write>(&self, writer: &mut W) {
136            let mut counter = WriteCounter::new(writer);
137            ::bincode::serialize_into(&mut counter, &self).expect("bincode::serialize_into() failed");
138            let written = counter.count;
139            let written_slop = ((written + 7) & !7) - written;
140            counter.write_all(&[0u8; 8][..written_slop]).unwrap();
141        }
142    }
143
144    use write_counter::WriteCounter;
145    /// A `Write` wrapper that counts the bytes written.
146    mod write_counter {
147
148        use ::std::io::{Write, IoSlice, Result};
149        use std::fmt::Arguments;
150
151        /// A write wrapper that tracks the bytes written.
152        pub struct WriteCounter<W> {
153            inner: W,
154            pub count: usize,
155        }
156
157        impl<W> WriteCounter<W> {
158            /// Creates a new counter wrapper from a writer.
159            pub fn new(inner: W) -> Self {
160                Self { inner, count: 0 }
161            }
162        }
163
164        impl<W: Write> Write for WriteCounter<W> {
165            fn write(&mut self, buf: &[u8]) -> Result<usize> {
166                let written = self.inner.write(buf)?;
167                self.count += written;
168                Ok(written)
169            }
170            fn flush(&mut self) -> Result<()> {
171                self.inner.flush()
172            }
173            fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
174                let written = self.inner.write_vectored(bufs)?;
175                self.count += written;
176                Ok(written)
177            }
178            fn write_all(&mut self, buf: &[u8]) -> Result<()> {
179                self.count += buf.len();
180                self.inner.write_all(buf)
181            }
182            fn write_fmt(&mut self, _fmt: Arguments<'_>) -> Result<()> {
183                unimplemented!()
184            }
185        }
186    }
187}
timely/dataflow/channels/mod.rs

timely/dataflow/channels/
mod.rs
