Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions container/src/columnation.rs
Original file line number Diff line number Diff line change
Expand Up @@ -339,7 +339,7 @@ mod container {

use crate::columnation::{Columnation, TimelyStack};

impl<T: Columnation + 'static> Container for TimelyStack<T> {
impl<T: Columnation> Container for TimelyStack<T> {
type ItemRef<'a> = &'a T where Self: 'a;
type Item<'a> = &'a T where Self: 'a;

Expand All @@ -355,20 +355,20 @@ mod container {
TimelyStack::clear(self)
}

type Iter<'a> = std::slice::Iter<'a, T>;
type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;

fn iter(&self) -> Self::Iter<'_> {
self.deref().iter()
}

type DrainIter<'a> = std::slice::Iter<'a, T>;
type DrainIter<'a> = std::slice::Iter<'a, T> where Self: 'a;

fn drain(&mut self) -> Self::DrainIter<'_> {
(*self).iter()
}
}

impl<T: Columnation + 'static> SizableContainer for TimelyStack<T> {
impl<T: Columnation> SizableContainer for TimelyStack<T> {
fn capacity(&self) -> usize {
self.capacity()
}
Expand Down
8 changes: 4 additions & 4 deletions container/src/flatcontainer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
pub use flatcontainer::*;
use crate::{buffer, Container, SizableContainer, PushInto};

impl<R: Region + Clone + 'static> Container for FlatStack<R> {
impl<R: Region> Container for FlatStack<R> {
type ItemRef<'a> = R::ReadItem<'a> where Self: 'a;
type Item<'a> = R::ReadItem<'a> where Self: 'a;

Expand All @@ -15,20 +15,20 @@ impl<R: Region + Clone + 'static> Container for FlatStack<R> {
self.clear()
}

type Iter<'a> = <&'a Self as IntoIterator>::IntoIter;
type Iter<'a> = <&'a Self as IntoIterator>::IntoIter where Self: 'a;

fn iter(&self) -> Self::Iter<'_> {
IntoIterator::into_iter(self)
}

type DrainIter<'a> = Self::Iter<'a>;
type DrainIter<'a> = Self::Iter<'a> where Self: 'a;

fn drain(&mut self) -> Self::DrainIter<'_> {
IntoIterator::into_iter(&*self)
}
}

impl<R: Region + Clone + 'static> SizableContainer for FlatStack<R> {
impl<R: Region> SizableContainer for FlatStack<R> {
fn capacity(&self) -> usize {
self.capacity()
}
Expand Down
27 changes: 13 additions & 14 deletions container/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,7 @@ pub mod flatcontainer;
/// We require the container to be cloneable to enable efficient copies when providing references
/// of containers to operators. Care must be taken that the type's `clone_from` implementation
/// is efficient (which is not necessarily the case when deriving `Clone`.)
/// TODO: Don't require `Container: Clone`
pub trait Container: Default + Clone + 'static {
pub trait Container: Default {
/// The type of elements when reading non-destructively from the container.
type ItemRef<'a> where Self: 'a;

Expand Down Expand Up @@ -50,13 +49,13 @@ pub trait Container: Default + Clone + 'static {
fn clear(&mut self);

/// Iterator type when reading from the container.
type Iter<'a>: Iterator<Item=Self::ItemRef<'a>>;
type Iter<'a>: Iterator<Item=Self::ItemRef<'a>> where Self: 'a;

/// Returns an iterator that reads the contents of this container.
fn iter(&self) -> Self::Iter<'_>;

/// Iterator type when draining the container.
type DrainIter<'a>: Iterator<Item=Self::Item<'a>>;
type DrainIter<'a>: Iterator<Item=Self::Item<'a>> where Self: 'a;

/// Returns an iterator that drains the contents of this container.
/// Drain leaves the container in an undefined state.
Expand Down Expand Up @@ -104,7 +103,7 @@ pub trait PushInto<T> {
/// decide to represent a push order for `extract` and `finish`, or not.
pub trait ContainerBuilder: Default + 'static {
/// The container type we're building.
type Container: Container;
type Container: Container + Clone + 'static;
/// Extract assembled containers, potentially leaving unfinished data behind. Can
/// be called repeatedly, for example while the caller can send data.
///
Expand Down Expand Up @@ -160,7 +159,7 @@ impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuil
}
}

impl<C: Container> ContainerBuilder for CapacityContainerBuilder<C> {
impl<C: Container + Clone + 'static> ContainerBuilder for CapacityContainerBuilder<C> {
type Container = C;

#[inline]
Expand Down Expand Up @@ -204,7 +203,7 @@ impl<C: Container> CapacityContainerBuilder<C> {
}
}

impl<T: Clone + 'static> Container for Vec<T> {
impl<T> Container for Vec<T> {
type ItemRef<'a> = &'a T where T: 'a;
type Item<'a> = T where T: 'a;

Expand All @@ -218,20 +217,20 @@ impl<T: Clone + 'static> Container for Vec<T> {

fn clear(&mut self) { Vec::clear(self) }

type Iter<'a> = std::slice::Iter<'a, T>;
type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;

fn iter(&self) -> Self::Iter<'_> {
self.as_slice().iter()
}

type DrainIter<'a> = std::vec::Drain<'a, T>;
type DrainIter<'a> = std::vec::Drain<'a, T> where Self: 'a;

fn drain(&mut self) -> Self::DrainIter<'_> {
self.drain(..)
}
}

impl<T: Clone + 'static> SizableContainer for Vec<T> {
impl<T> SizableContainer for Vec<T> {
fn capacity(&self) -> usize {
self.capacity()
}
Expand Down Expand Up @@ -294,13 +293,13 @@ mod rc {
}
}

type Iter<'a> = T::Iter<'a>;
type Iter<'a> = T::Iter<'a> where Self: 'a;

fn iter(&self) -> Self::Iter<'_> {
self.deref().iter()
}

type DrainIter<'a> = T::Iter<'a>;
type DrainIter<'a> = T::Iter<'a> where Self: 'a;

fn drain(&mut self) -> Self::DrainIter<'_> {
self.iter()
Expand Down Expand Up @@ -335,13 +334,13 @@ mod arc {
}
}

type Iter<'a> = T::Iter<'a>;
type Iter<'a> = T::Iter<'a> where Self: 'a;

fn iter(&self) -> Self::Iter<'_> {
self.deref().iter()
}

type DrainIter<'a> = T::Iter<'a>;
type DrainIter<'a> = T::Iter<'a> where Self: 'a;

fn drain(&mut self) -> Self::DrainIter<'_> {
self.iter()
Expand Down
2 changes: 1 addition & 1 deletion timely/src/dataflow/channels/pact.rs
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ pub trait ParallelizationContract<T, C> {
#[derive(Debug)]
pub struct Pipeline;

impl<T: 'static, C: Container> ParallelizationContract<T, C> for Pipeline {
impl<T: 'static, C: Container + 'static> ParallelizationContract<T, C> for Pipeline {
type Pusher = LogPusher<T, C, ThreadPusher<Message<T, C>>>;
type Puller = LogPuller<T, C, ThreadPuller<Message<T, C>>>;
fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: Rc<[usize]>, logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
Expand Down
6 changes: 3 additions & 3 deletions timely/src/dataflow/channels/pushers/buffer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ use crate::container::{ContainerBuilder, CapacityContainerBuilder, PushInto};
use crate::dataflow::channels::Message;
use crate::dataflow::operators::Capability;
use crate::progress::Timestamp;
use crate::Container;
use crate::{Container, Data};

/// Buffers data sent at the same time, for efficient communication.
///
Expand Down Expand Up @@ -44,7 +44,7 @@ impl<T, CB: Default, P> Buffer<T, CB, P> {
}
}

impl<T, C: Container, P: Push<Message<T, C>>> Buffer<T, CapacityContainerBuilder<C>, P> where T: Eq+Clone {
impl<T, C: Container + Data, P: Push<Message<T, C>>> Buffer<T, CapacityContainerBuilder<C>, P> where T: Eq+Clone {
/// Returns a `Session`, which accepts data to send at the associated time
#[inline]
pub fn session(&mut self, time: &T) -> Session<T, CapacityContainerBuilder<C>, P> {
Expand Down Expand Up @@ -133,7 +133,7 @@ pub struct Session<'a, T, CB, P> {
buffer: &'a mut Buffer<T, CB, P>,
}

impl<'a, T, C: Container, P> Session<'a, T, CapacityContainerBuilder<C>, P>
impl<'a, T, C: Container + Data, P> Session<'a, T, CapacityContainerBuilder<C>, P>
where
T: Eq + Clone + 'a,
P: Push<Message<T, C>> + 'a,
Expand Down
2 changes: 1 addition & 1 deletion timely/src/dataflow/channels/pushers/tee.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ pub struct Tee<T, C> {
shared: PushList<T, C>,
}

impl<T: Data, C: Container> Push<Message<T, C>> for Tee<T, C> {
impl<T: Data, C: Container + Data> Push<Message<T, C>> for Tee<T, C> {
#[inline]
fn push(&mut self, message: &mut Option<Message<T, C>>) {
let mut pushers = self.shared.borrow_mut();
Expand Down
2 changes: 1 addition & 1 deletion timely/src/dataflow/operators/branch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ pub trait BranchWhen<T>: Sized {
fn branch_when(&self, condition: impl Fn(&T) -> bool + 'static) -> (Self, Self);
}

impl<S: Scope, C: Container> BranchWhen<S::Timestamp> for StreamCore<S, C> {
impl<S: Scope, C: Container + Data> BranchWhen<S::Timestamp> for StreamCore<S, C> {
fn branch_when(&self, condition: impl Fn(&S::Timestamp) -> bool + 'static) -> (Self, Self) {
let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());

Expand Down
6 changes: 3 additions & 3 deletions timely/src/dataflow/operators/core/capture/capture.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,14 @@ use crate::dataflow::channels::pact::Pipeline;
use crate::dataflow::channels::pullers::Counter as PullCounter;
use crate::dataflow::operators::generic::builder_raw::OperatorBuilder;

use crate::Container;
use crate::{Container, Data};
use crate::progress::ChangeBatch;
use crate::progress::Timestamp;

use super::{Event, EventPusher};

/// Capture a stream of timestamped data for later replay.
pub trait Capture<T: Timestamp, C: Container> {
pub trait Capture<T: Timestamp, C: Container + Data> {
/// Captures a stream of timestamped data for later replay.
///
/// # Examples
Expand Down Expand Up @@ -113,7 +113,7 @@ pub trait Capture<T: Timestamp, C: Container> {
}
}

impl<S: Scope, C: Container> Capture<S::Timestamp, C> for StreamCore<S, C> {
impl<S: Scope, C: Container + Data> Capture<S::Timestamp, C> for StreamCore<S, C> {
fn capture_into<P: EventPusher<S::Timestamp, C>+'static>(&self, mut event_pusher: P) {

let mut builder = OperatorBuilder::new("Capture".to_owned(), self.scope());
Expand Down
2 changes: 1 addition & 1 deletion timely/src/dataflow/operators/core/capture/replay.rs
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ pub trait Replay<T: Timestamp, C> : Sized {
fn replay_core<S: Scope<Timestamp=T>>(self, scope: &mut S, period: Option<std::time::Duration>) -> StreamCore<S, C>;
}

impl<T: Timestamp, C: Container, I> Replay<T, C> for I
impl<T: Timestamp, C: Container + Clone + 'static, I> Replay<T, C> for I
where
I : IntoIterator,
<I as IntoIterator>::Item: EventIterator<T, C>+'static,
Expand Down
8 changes: 4 additions & 4 deletions timely/src/dataflow/operators/core/concat.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
//! Merges the contents of multiple streams.


use crate::Container;
use crate::{Container, Data};
use crate::dataflow::channels::pact::Pipeline;
use crate::dataflow::{StreamCore, Scope};

Expand All @@ -23,7 +23,7 @@ pub trait Concat<G: Scope, C: Container> {
fn concat(&self, _: &StreamCore<G, C>) -> StreamCore<G, C>;
}

impl<G: Scope, C: Container> Concat<G, C> for StreamCore<G, C> {
impl<G: Scope, C: Container + Data> Concat<G, C> for StreamCore<G, C> {
fn concat(&self, other: &StreamCore<G, C>) -> StreamCore<G, C> {
self.scope().concatenate([self.clone(), other.clone()])
}
Expand Down Expand Up @@ -52,7 +52,7 @@ pub trait Concatenate<G: Scope, C: Container> {
I: IntoIterator<Item=StreamCore<G, C>>;
}

impl<G: Scope, C: Container> Concatenate<G, C> for StreamCore<G, C> {
impl<G: Scope, C: Container + Data> Concatenate<G, C> for StreamCore<G, C> {
fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
where
I: IntoIterator<Item=StreamCore<G, C>>
Expand All @@ -62,7 +62,7 @@ impl<G: Scope, C: Container> Concatenate<G, C> for StreamCore<G, C> {
}
}

impl<G: Scope, C: Container> Concatenate<G, C> for G {
impl<G: Scope, C: Container + Data> Concatenate<G, C> for G {
fn concatenate<I>(&self, sources: I) -> StreamCore<G, C>
where
I: IntoIterator<Item=StreamCore<G, C>>
Expand Down
6 changes: 3 additions & 3 deletions timely/src/dataflow/operators/core/enterleave.rs
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ pub trait Leave<G: Scope, C: Container> {
fn leave(&self) -> StreamCore<G, C>;
}

impl<G: Scope, C: Clone+Container, T: Timestamp+Refines<G::Timestamp>> Leave<G, C> for StreamCore<Child<'_, G, T>, C> {
impl<G: Scope, C: Container + Data, T: Timestamp+Refines<G::Timestamp>> Leave<G, C> for StreamCore<Child<'_, G, T>, C> {
fn leave(&self) -> StreamCore<G, C> {

let scope = self.scope();
Expand All @@ -130,14 +130,14 @@ impl<G: Scope, C: Clone+Container, T: Timestamp+Refines<G::Timestamp>> Leave<G,
}


struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> {
struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container + Data> {
targets: Counter<TInner, TContainer, Tee<TInner, TContainer>>,
phantom: ::std::marker::PhantomData<TOuter>,
activator: crate::scheduling::Activator,
active: bool,
}

impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> Push<Message<TOuter, TContainer>> for IngressNub<TOuter, TInner, TContainer> {
impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container + Data> Push<Message<TOuter, TContainer>> for IngressNub<TOuter, TInner, TContainer> {
fn push(&mut self, element: &mut Option<Message<TOuter, TContainer>>) {
if let Some(outer_message) = element {
let data = ::std::mem::take(&mut outer_message.data);
Expand Down
16 changes: 8 additions & 8 deletions timely/src/dataflow/operators/core/feedback.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
//! Create cycles in a timely dataflow graph.

use crate::Container;
use crate::{Container, Data};
use crate::container::CapacityContainerBuilder;
use crate::dataflow::channels::pact::Pipeline;
use crate::dataflow::channels::pushers::Tee;
Expand Down Expand Up @@ -36,7 +36,7 @@ pub trait Feedback<G: Scope> {
/// .connect_loop(handle);
/// });
/// ```
fn feedback<C: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>);
fn feedback<C: Container + Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>);
}

/// Creates a `StreamCore` and a `Handle` to later bind the source of that `StreamCore`.
Expand Down Expand Up @@ -64,12 +64,12 @@ pub trait LoopVariable<'a, G: Scope, T: Timestamp> {
/// });
/// });
/// ```
fn loop_variable<C: Container>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>);
fn loop_variable<C: Container + Data>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>);
}

impl<G: Scope> Feedback<G> for G {

fn feedback<C: Container>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>) {
fn feedback<C: Container + Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>) {

let mut builder = OperatorBuilder::new("Feedback".to_owned(), self.clone());
let (output, stream) = builder.new_output();
Expand All @@ -79,13 +79,13 @@ impl<G: Scope> Feedback<G> for G {
}

impl<'a, G: Scope, T: Timestamp> LoopVariable<'a, G, T> for Iterative<'a, G, T> {
fn loop_variable<C: Container>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>) {
fn loop_variable<C: Container + Data>(&mut self, summary: T::Summary) -> (Handle<Iterative<'a, G, T>, C>, StreamCore<Iterative<'a, G, T>, C>) {
self.feedback(Product::new(Default::default(), summary))
}
}

/// Connect a `Stream` to the input of a loop variable.
pub trait ConnectLoop<G: Scope, C: Container> {
pub trait ConnectLoop<G: Scope, C: Container + Data> {
/// Connect a `Stream` to be the input of a loop variable.
///
/// # Examples
Expand All @@ -106,7 +106,7 @@ pub trait ConnectLoop<G: Scope, C: Container> {
fn connect_loop(&self, handle: Handle<G, C>);
}

impl<G: Scope, C: Container> ConnectLoop<G, C> for StreamCore<G, C> {
impl<G: Scope, C: Container + Data> ConnectLoop<G, C> for StreamCore<G, C> {
fn connect_loop(&self, handle: Handle<G, C>) {

let mut builder = handle.builder;
Expand All @@ -131,7 +131,7 @@ impl<G: Scope, C: Container> ConnectLoop<G, C> for StreamCore<G, C> {

/// A handle used to bind the source of a loop variable.
#[derive(Debug)]
pub struct Handle<G: Scope, C: Container> {
pub struct Handle<G: Scope, C: Container + Data> {
builder: OperatorBuilder<G>,
summary: <G::Timestamp as Timestamp>::Summary,
output: OutputWrapper<G::Timestamp, CapacityContainerBuilder<C>, Tee<G::Timestamp, C>>,
Expand Down
Loading