Rename CounterCore to Counter

timely/src/dataflow/channels/pushers/counter.rs

@@ -11,16 +11,13 @@ use crate::Container;
 
 /// A wrapper which updates shared `produced` based on the number of records pushed.
 #[derive(Debug)]
-pub struct CounterCore<T, D, P: Push<Bundle<T, D>>> {
+pub struct Counter<T, D, P: Push<Bundle<T, D>>> {
     pushee: P,
     produced: Rc<RefCell<ChangeBatch<T>>>,
     phantom: PhantomData<D>,
 }
 
-/// A counter specialized to vector.
-pub type Counter<T, D, P> = CounterCore<T, Vec<D>, P>;
-
-impl<T: Timestamp, D: Container, P> Push<Bundle<T, D>> for CounterCore<T, D, P> where P: Push<Bundle<T, D>> {
+impl<T: Timestamp, D: Container, P> Push<Bundle<T, D>> for Counter<T, D, P> where P: Push<Bundle<T, D>> {
     #[inline]
     fn push(&mut self, message: &mut Option<Bundle<T, D>>) {
         if let Some(message) = message {
@@ -34,10 +31,10 @@ impl<T: Timestamp, D: Container, P> Push<Bundle<T, D>> for CounterCore<T, D, P>
     }
 }
 
-impl<T, D, P: Push<Bundle<T, D>>> CounterCore<T, D, P> where T : Ord+Clone+'static {
+impl<T, D, P: Push<Bundle<T, D>>> Counter<T, D, P> where T : Ord+Clone+'static {
     /// Allocates a new `Counter` from a pushee and shared counts.
-    pub fn new(pushee: P) -> CounterCore<T, D, P> {
-        CounterCore {
+    pub fn new(pushee: P) -> Counter<T, D, P> {
+        Counter {
             pushee,
             produced: Rc::new(RefCell::new(ChangeBatch::new())),
             phantom: PhantomData,

timely/src/dataflow/channels/pushers/mod.rs

@@ -1,6 +1,6 @@
 pub use self::tee::{Tee, TeeHelper};
 pub use self::exchange::Exchange;
-pub use self::counter::{Counter, CounterCore};
+pub use self::counter::Counter;
 
 pub mod tee;
 pub mod exchange;

timely/src/dataflow/operators/capture/replay.rs

@@ -39,7 +39,7 @@
 //! than that in which the stream was captured.
 
 use crate::dataflow::{Scope, StreamCore};
-use crate::dataflow::channels::pushers::CounterCore as PushCounter;
+use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::Buffer as PushBuffer;
 use crate::dataflow::operators::generic::builder_raw::OperatorBuilder;
 use crate::progress::Timestamp;

timely/src/dataflow/operators/enterleave.rs

@@ -28,7 +28,7 @@ use crate::progress::{Source, Target};
 use crate::order::Product;
 use crate::{Container, Data};
 use crate::communication::Push;
-use crate::dataflow::channels::pushers::{CounterCore, Tee};
+use crate::dataflow::channels::pushers::{Counter, Tee};
 use crate::dataflow::channels::{Bundle, Message};
 
 use crate::worker::AsWorker;
@@ -90,7 +90,7 @@ impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Data+Container> Enter<G, T
 
         let (targets, registrar) = Tee::<T, C>::new();
         let ingress = IngressNub {
-            targets: CounterCore::new(targets),
+            targets: Counter::new(targets),
             phantom: PhantomData,
             activator: scope.activator_for(&scope.addr()),
             active: false,
@@ -161,7 +161,7 @@ impl<'a, G: Scope, D: Clone+Container, T: Timestamp+Refines<G::Timestamp>> Leave
 
 
 struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TData: Container> {
-    targets: CounterCore<TInner, TData, Tee<TInner, TData>>,
+    targets: Counter<TInner, TData, Tee<TInner, TData>>,
     phantom: ::std::marker::PhantomData<TOuter>,
     activator: crate::scheduling::Activator,
     active: bool,

timely/src/dataflow/operators/generic/builder_rc.rs

@@ -11,7 +11,7 @@ use crate::progress::frontier::{Antichain, MutableAntichain};
 use crate::Container;
 use crate::dataflow::{Scope, StreamCore};
 use crate::dataflow::channels::pushers::Tee;
-use crate::dataflow::channels::pushers::CounterCore as PushCounter;
+use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::Buffer as PushBuffer;
 use crate::dataflow::channels::pact::ParallelizationContract;
 use crate::dataflow::channels::pullers::Counter as PullCounter;

timely/src/dataflow/operators/generic/handles.rs

@@ -11,7 +11,7 @@ use crate::progress::Timestamp;
 use crate::progress::ChangeBatch;
 use crate::progress::frontier::MutableAntichain;
 use crate::dataflow::channels::pullers::Counter as PullCounter;
-use crate::dataflow::channels::pushers::CounterCore as PushCounter;
+use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::{Buffer, Session};
 use crate::dataflow::channels::Bundle;
 use crate::communication::{Push, Pull, message::RefOrMut};

timely/src/dataflow/operators/input.rs

@@ -12,7 +12,7 @@ use crate::progress::Source;
 use crate::{Container, Data};
 use crate::communication::Push;
 use crate::dataflow::{Stream, ScopeParent, Scope, StreamCore};
-use crate::dataflow::channels::pushers::{Tee, CounterCore};
+use crate::dataflow::channels::pushers::{Tee, Counter};
 use crate::dataflow::channels::Message;
 
 
@@ -178,7 +178,7 @@ impl<G: Scope> Input for G where <G as ScopeParent>::Timestamp: TotalOrder {
 
     fn input_from_core<D: Container>(&mut self, handle: &mut HandleCore<<G as ScopeParent>::Timestamp, D>) -> StreamCore<G, D> {
         let (output, registrar) = Tee::<<G as ScopeParent>::Timestamp, D>::new();
-        let counter = CounterCore::new(output);
+        let counter = Counter::new(output);
         let produced = counter.produced().clone();
 
         let index = self.allocate_operator_index();
@@ -249,7 +249,7 @@ impl<T:Timestamp> Operate<T> for Operator<T> {
 pub struct HandleCore<T: Timestamp, C: Container> {
     activate: Vec<Activator>,
     progress: Vec<Rc<RefCell<ChangeBatch<T>>>>,
-    pushers: Vec<CounterCore<T, C, Tee<T, C>>>,
+    pushers: Vec<Counter<T, C, Tee<T, C>>>,
     buffer1: C,
     buffer2: C,
     now_at: T,
@@ -332,7 +332,7 @@ impl<T: Timestamp, D: Container> HandleCore<T, D> {
 
     fn register(
         &mut self,
-        pusher: CounterCore<T, D, Tee<T, D>>,
+        pusher: Counter<T, D, Tee<T, D>>,
         progress: Rc<RefCell<ChangeBatch<T>>>,
     ) {
         // flush current contents, so new registrant does not see existing data.

timely/src/dataflow/operators/probe.rs

@@ -5,7 +5,7 @@ use std::cell::RefCell;
 
 use crate::progress::Timestamp;
 use crate::progress::frontier::{AntichainRef, MutableAntichain};
-use crate::dataflow::channels::pushers::CounterCore as PushCounter;
+use crate::dataflow::channels::pushers::Counter as PushCounter;
 use crate::dataflow::channels::pushers::buffer::Buffer as PushBuffer;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::channels::pullers::Counter as PullCounter;

timely/src/dataflow/operators/unordered_input.rs

@@ -12,7 +12,7 @@ use crate::progress::Source;
 use crate::progress::ChangeBatch;
 
 use crate::Data;
-use crate::dataflow::channels::pushers::{CounterCore as PushCounter, Tee};
+use crate::dataflow::channels::pushers::{Counter as PushCounter, Tee};
 use crate::dataflow::channels::pushers::buffer::{Buffer as PushBuffer, AutoflushSessionCore};
 
 use crate::dataflow::operators::{ActivateCapability, Capability};