From ca6d15912fcd55f52f05b118f926f3359b2a2610 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 17 Sep 2025 15:43:32 -0400 Subject: [PATCH] Remove Container argument from Counter --- timely/src/dataflow/channels/pact.rs | 6 +----- timely/src/dataflow/channels/pushers/counter.rs | 11 ++++------- timely/src/dataflow/operators/core/enterleave.rs | 2 +- timely/src/dataflow/operators/core/input.rs | 6 +++--- timely/src/dataflow/operators/core/unordered_input.rs | 8 ++++---- timely/src/dataflow/operators/generic/handles.rs | 10 +++++----- 6 files changed, 18 insertions(+), 25 deletions(-) diff --git a/timely/src/dataflow/channels/pact.rs b/timely/src/dataflow/channels/pact.rs index e3e8af8a9..296e87a8f 100644 --- a/timely/src/dataflow/channels/pact.rs +++ b/timely/src/dataflow/channels/pact.rs @@ -110,11 +110,7 @@ mod distributor { C: Accountable + ContainerBytes + Send + 'static, D: Distributor + 'static, { - type Pusher = Exchange< - T, - LogPusher>>>, - D - >; + type Pusher = Exchange>>>, D>; type Puller = LogPuller>>>; fn connect(self, allocator: &mut A, identifier: usize, address: Rc<[usize]>, logging: Option) -> (Self::Pusher, Self::Puller) { let (senders, receiver) = allocator.allocate::>(identifier, address); diff --git a/timely/src/dataflow/channels/pushers/counter.rs b/timely/src/dataflow/channels/pushers/counter.rs index 67d39318d..8718c8acd 100644 --- a/timely/src/dataflow/channels/pushers/counter.rs +++ b/timely/src/dataflow/channels/pushers/counter.rs @@ -1,6 +1,5 @@ //! A wrapper which counts the number of records pushed past and updates a shared count map. -use std::marker::PhantomData; use std::rc::Rc; use std::cell::RefCell; @@ -11,13 +10,12 @@ use crate::Accountable; /// A wrapper which updates shared `produced` based on the number of records pushed. #[derive(Debug)] -pub struct Counter>> { +pub struct Counter { pushee: P, produced: Rc>>, - phantom: PhantomData, } -impl Push> for Counter where P: Push> { +impl Push> for Counter where P: Push> { #[inline] fn push(&mut self, message: &mut Option>) { if let Some(message) = message { @@ -31,13 +29,12 @@ impl Push> for Counter w } } -impl>> Counter where T : Ord+Clone+'static { +impl Counter where T : Ord+Clone+'static { /// Allocates a new `Counter` from a pushee and shared counts. - pub fn new(pushee: P) -> Counter { + pub fn new(pushee: P) -> Counter { Counter { pushee, produced: Rc::new(RefCell::new(ChangeBatch::new())), - phantom: PhantomData, } } /// A references to shared changes in counts, for cloning or draining. diff --git a/timely/src/dataflow/operators/core/enterleave.rs b/timely/src/dataflow/operators/core/enterleave.rs index c0889498e..14787459b 100644 --- a/timely/src/dataflow/operators/core/enterleave.rs +++ b/timely/src/dataflow/operators/core/enterleave.rs @@ -131,7 +131,7 @@ impl> Leave for struct IngressNub, TContainer: Container> { - targets: Counter>, + targets: Counter>, phantom: ::std::marker::PhantomData, activator: crate::scheduling::Activator, active: bool, diff --git a/timely/src/dataflow/operators/core/input.rs b/timely/src/dataflow/operators/core/input.rs index 6a71f63b4..8cb95495e 100644 --- a/timely/src/dataflow/operators/core/input.rs +++ b/timely/src/dataflow/operators/core/input.rs @@ -218,7 +218,7 @@ impl Operate for Operator { pub struct Handle { activate: Vec, progress: Vec>>>, - pushers: Vec>>, + pushers: Vec>>, builder: CB, buffer: CB::Container, now_at: T, @@ -341,7 +341,7 @@ impl Handle { fn register( &mut self, - pusher: Counter>, + pusher: Counter>, progress: Rc>>, ) { // flush current contents, so new registrant does not see existing data. @@ -380,7 +380,7 @@ impl Handle { fn send_container( container: &mut CB::Container, buffer: &mut CB::Container, - pushers: &mut [Counter>], + pushers: &mut [Counter>], now_at: &T ) { for index in 0 .. pushers.len() { diff --git a/timely/src/dataflow/operators/core/unordered_input.rs b/timely/src/dataflow/operators/core/unordered_input.rs index 703a461d3..a06633c26 100644 --- a/timely/src/dataflow/operators/core/unordered_input.rs +++ b/timely/src/dataflow/operators/core/unordered_input.rs @@ -148,11 +148,11 @@ impl Operate for UnorderedOperator { /// A handle to an input [StreamCore], used to introduce data to a timely dataflow computation. #[derive(Debug)] pub struct UnorderedHandle { - buffer: PushBuffer>>, + buffer: PushBuffer>>, } impl UnorderedHandle { - fn new(pusher: Counter>) -> UnorderedHandle { + fn new(pusher: Counter>) -> UnorderedHandle { UnorderedHandle { buffer: PushBuffer::new(pusher), } @@ -160,7 +160,7 @@ impl UnorderedHandle { /// Allocates a new automatically flushing session based on the supplied capability. #[inline] - pub fn session_with_builder(&mut self, cap: ActivateCapability) -> ActivateOnDrop>>> { + pub fn session_with_builder(&mut self, cap: ActivateCapability) -> ActivateOnDrop>>> { ActivateOnDrop::new(self.buffer.autoflush_session_with_builder(cap.capability.clone()), Rc::clone(&cap.address), Rc::clone(&cap.activations)) } } @@ -168,7 +168,7 @@ impl UnorderedHandle { impl UnorderedHandle> { /// Allocates a new automatically flushing session based on the supplied capability. #[inline] - pub fn session(&mut self, cap: ActivateCapability) -> ActivateOnDrop, Counter>>> { + pub fn session(&mut self, cap: ActivateCapability) -> ActivateOnDrop, Counter>>> { self.session_with_builder(cap) } } diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs index a10ff2f40..88c43965a 100644 --- a/timely/src/dataflow/operators/generic/handles.rs +++ b/timely/src/dataflow/operators/generic/handles.rs @@ -159,14 +159,14 @@ pub fn new_input_handle>>( /// pusher is flushed (via the `cease` method) once it is no longer used. #[derive(Debug)] pub struct OutputWrapper>> { - push_buffer: Buffer>, + push_buffer: Buffer>, internal_buffer: Rc>>, port: usize, } impl>> OutputWrapper { /// Creates a new output wrapper from a push buffer. - pub fn new(push_buffer: Buffer>, internal_buffer: Rc>>, port: usize) -> Self { + pub fn new(push_buffer: Buffer>, internal_buffer: Rc>>, port: usize) -> Self { OutputWrapper { push_buffer, internal_buffer, @@ -188,7 +188,7 @@ impl>> Out /// Handle to an operator's output stream. pub struct OutputHandleCore<'a, T: Timestamp, CB: ContainerBuilder+'a, P: Push>+'a> { - push_buffer: &'a mut Buffer>, + push_buffer: &'a mut Buffer>, internal_buffer: &'a Rc>>, port: usize, } @@ -220,7 +220,7 @@ impl<'a, T: Timestamp, CB: ContainerBuilder, P: Push>> /// }); /// }); /// ``` - pub fn session_with_builder<'b, CT: CapabilityTrait>(&'b mut self, cap: &'b CT) -> Session<'b, T, CB, PushCounter> where 'a: 'b { + pub fn session_with_builder<'b, CT: CapabilityTrait>(&'b mut self, cap: &'b CT) -> Session<'b, T, CB, PushCounter> where 'a: 'b { debug_assert!(cap.valid_for_output(self.internal_buffer, self.port), "Attempted to open output session with invalid capability"); self.push_buffer.session_with_builder(cap.time()) } @@ -255,7 +255,7 @@ impl<'a, T: Timestamp, C: Container, P: Push>> OutputHandleCore<'a /// }); /// ``` #[inline] - pub fn session<'b, CT: CapabilityTrait>(&'b mut self, cap: &'b CT) -> Session<'b, T, CapacityContainerBuilder, PushCounter> where 'a: 'b { + pub fn session<'b, CT: CapabilityTrait>(&'b mut self, cap: &'b CT) -> Session<'b, T, CapacityContainerBuilder, PushCounter> where 'a: 'b { self.session_with_builder(cap) } }