timely_communication/allocator/
mod.rs

1//! Types and traits for the allocation of channels.
2
3use std::rc::Rc;
4use std::cell::RefCell;
5use std::time::Duration;
6use std::collections::VecDeque;
7
8pub use self::thread::Thread;
9pub use self::process::Process;
10pub use self::generic::{Generic, GenericBuilder};
11
12pub mod thread;
13pub mod process;
14pub mod generic;
15
16pub mod canary;
17pub mod counters;
18
19pub mod zero_copy;
20
21use crate::{Data, Push, Pull, Message};
22
23/// A proto-allocator, which implements `Send` and can be completed with `build`.
24///
25/// This trait exists because some allocators contain elements that do not implement
26/// the `Send` trait, for example `Rc` wrappers for shared state. As such, what we
27/// actually need to create to initialize a computation are builders, which we can
28/// then move into new threads each of which then construct their actual allocator.
29pub trait AllocateBuilder : Send {
30    /// The type of allocator to be built.
31    type Allocator: Allocate;
32    /// Builds allocator, consumes self.
33    fn build(self) -> Self::Allocator;
34}
35
36/// A type capable of allocating channels.
37///
38/// There is some feature creep, in that this contains several convenience methods about the nature
39/// of the allocated channels, and maintenance methods to ensure that they move records around.
40pub trait Allocate {
41    /// The index of the worker out of `(0..self.peers())`.
42    fn index(&self) -> usize;
43    /// The number of workers in the communication group.
44    fn peers(&self) -> usize;
45    /// Constructs several send endpoints and one receive endpoint.
46    fn allocate<T: Data>(&mut self, identifier: usize) -> (Vec<Box<dyn Push<Message<T>>>>, Box<dyn Pull<Message<T>>>);
47    /// A shared queue of communication events with channel identifier.
48    ///
49    /// It is expected that users of the channel allocator will regularly
50    /// drain these events in order to drive their computation. If they
51    /// fail to do so the event queue may become quite large, and turn
52    /// into a performance problem.
53    fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>>;
54
55    /// Awaits communication events.
56    ///
57    /// This method may park the current thread, for at most `duration`,
58    /// until new events arrive.
59    /// The method is not guaranteed to wait for any amount of time, but
60    /// good implementations should use this as a hint to park the thread.
61    fn await_events(&self, _duration: Option<Duration>) { }
62
63    /// Ensure that received messages are surfaced in each channel.
64    ///
65    /// This method should be called to ensure that received messages are
66    /// surfaced in each channel, but failing to call the method does not
67    /// ensure that they are not surfaced.
68    ///
69    /// Generally, this method is the indication that the allocator should
70    /// present messages contained in otherwise scarce resources (for example
71    /// network buffers), under the premise that someone is about to consume
72    /// the messages and release the resources.
73    fn receive(&mut self) { }
74
75    /// Signal the completion of a batch of reads from channels.
76    ///
77    /// Conventionally, this method signals to the communication fabric
78    /// that the worker is taking a break from reading from channels, and
79    /// the fabric should consider re-acquiring scarce resources. This can
80    /// lead to the fabric performing defensive copies out of un-consumed
81    /// buffers, and can be a performance problem if invoked casually.
82    fn release(&mut self) { }
83
84    /// Constructs a pipeline channel from the worker to itself.
85    ///
86    /// By default, this method uses the thread-local channel constructor
87    /// based on a shared `VecDeque` which updates the event queue.
88    fn pipeline<T: 'static>(&mut self, identifier: usize) ->
89        (thread::ThreadPusher<Message<T>>,
90         thread::ThreadPuller<Message<T>>)
91    {
92        thread::Thread::new_from(identifier, self.events().clone())
93    }
94}
95
96/// A communication channel event.
97pub enum Event {
98    /// A number of messages pushed into the channel.
99    Pushed(usize),
100    /// A number of messages pulled from the channel.
101    Pulled(usize),
102}
timely_communication/allocator/mod.rs

timely_communication/allocator/
mod.rs
