Merge branch 'master' into miri_test

antiguru · antiguru · commit 91209b667c87 · 2025-03-24T17:04:03.000+01:00
diff --git a/communication/Cargo.toml b/communication/Cargo.toml
@@ -24,7 +24,6 @@ serde = { version = "1.0", features = ["derive"] }
 timely_bytes = { path = "../bytes", version = "0.13" }
 timely_container = { path = "../container", version = "0.15.0" }
 timely_logging = { path = "../logging", version = "0.13" }
-crossbeam-channel = "0.5"
 
 # Lgalloc only supports linux and macos, don't depend on any other OS.
 [target.'cfg(any(target_os = "linux", target_os = "macos"))'.dev-dependencies]
diff --git a/communication/src/allocator/counters.rs b/communication/src/allocator/counters.rs
@@ -2,6 +2,7 @@
 
 use std::rc::Rc;
 use std::cell::RefCell;
+use std::sync::mpsc::Sender;
 
 use crate::{Push, Pull};
 
@@ -51,8 +52,6 @@ impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
     }
 }
 
-use crossbeam_channel::Sender;
-
 /// The push half of an intra-thread channel.
 pub struct ArcPusher<T, P: Push<T>> {
     index: usize,
diff --git a/communication/src/allocator/process.rs b/communication/src/allocator/process.rs
@@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex};
 use std::any::Any;
 use std::time::Duration;
 use std::collections::{HashMap};
-use crossbeam_channel::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver};
 
 use crate::allocator::thread::{ThreadBuilder};
 use crate::allocator::{Allocate, AllocateBuilder, PeerBuilder, Thread};
@@ -80,7 +80,7 @@ impl PeerBuilder for Process {
         let mut counters_send = Vec::with_capacity(peers);
         let mut counters_recv = Vec::with_capacity(peers);
         for _ in 0 .. peers {
-            let (send, recv) = crossbeam_channel::unbounded();
+            let (send, recv) = std::sync::mpsc::channel();
             counters_send.push(send);
             counters_recv.push(recv);
         }
@@ -130,7 +130,7 @@ impl Allocate for Process {
                 let mut pushers = Vec::with_capacity(self.peers);
                 let mut pullers = Vec::with_capacity(self.peers);
                 for buzzer in self.buzzers.iter() {
-                    let (s, r): (Sender<T>, Receiver<T>) = crossbeam_channel::unbounded();
+                    let (s, r): (Sender<T>, Receiver<T>) = std::sync::mpsc::channel();
                     // TODO: the buzzer in the pusher may be redundant, because we need to buzz post-counter.
                     pushers.push((Pusher { target: s }, buzzer.clone()));
                     pullers.push(Puller { source: r, current: None });
diff --git a/communication/src/allocator/zero_copy/allocator.rs b/communication/src/allocator/zero_copy/allocator.rs
@@ -2,7 +2,7 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 use std::collections::{VecDeque, HashMap, hash_map::Entry};
-use crossbeam_channel::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver};
 
 use timely_bytes::arc::Bytes;
 
diff --git a/communication/src/allocator/zero_copy/allocator_process.rs b/communication/src/allocator/zero_copy/allocator_process.rs
@@ -3,7 +3,7 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 use std::collections::{VecDeque, HashMap, hash_map::Entry};
-use crossbeam_channel::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver};
 
 use timely_bytes::arc::Bytes;
 
diff --git a/communication/src/allocator/zero_copy/tcp.rs b/communication/src/allocator/zero_copy/tcp.rs
@@ -1,7 +1,7 @@
 //! Methods related to reading from and writing to TCP connections
 
 use std::io::{self, Write};
-use crossbeam_channel::{Sender, Receiver};
+use std::sync::mpsc::{Sender, Receiver};
 
 use crate::networking::MessageHeader;
 
diff --git a/communication/src/lib.rs b/communication/src/lib.rs
@@ -107,6 +107,8 @@ pub use allocator::Generic as Allocator;
 pub use allocator::{Allocate, Exchangeable};
 pub use initialize::{initialize, initialize_from, Config, WorkerGuards};
 
+use std::sync::mpsc::{Sender, Receiver};
+
 use timely_bytes::arc::Bytes;
 
 /// A type that can be serialized and deserialized through `Bytes`.
@@ -169,8 +171,6 @@ impl<T, P: ?Sized + Pull<T>> Pull<T> for Box<P> {
 }
 
 
-use crossbeam_channel::{Sender, Receiver};
-
 /// Allocate a matrix of send and receive changes to exchange items.
 ///
 /// This method constructs channels for `sends` threads to create and send
@@ -183,7 +183,7 @@ fn promise_futures<T>(sends: usize, recvs: usize) -> (Vec<Vec<Sender<T>>>, Vec<V
 
     for sender in senders.iter_mut() {
         for recver in recvers.iter_mut() {
-            let (send, recv) = crossbeam_channel::unbounded();
+            let (send, recv) = std::sync::mpsc::channel();
             sender.push(send);
             recver.push(recv);
         }
diff --git a/timely/src/dataflow/operators/capability.rs b/timely/src/dataflow/operators/capability.rs
@@ -27,9 +27,9 @@ use std::cell::RefCell;
 use std::fmt::{self, Debug};
 
 use crate::order::PartialOrder;
+use crate::progress::Antichain;
 use crate::progress::Timestamp;
 use crate::progress::ChangeBatch;
-use crate::progress::operate::PortConnectivity;
 use crate::scheduling::Activations;
 use crate::dataflow::channels::pullers::counter::ConsumedGuard;
 
@@ -238,26 +238,26 @@ pub struct InputCapability<T: Timestamp> {
     /// Output capability buffers, for use in minting capabilities.
     internal: CapabilityUpdates<T>,
     /// Timestamp summaries for each output.
-    summaries: Rc<RefCell<PortConnectivity<T::Summary>>>,
+    summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>,
     /// A drop guard that updates the consumed capability this InputCapability refers to on drop
     consumed_guard: ConsumedGuard<T>,
 }
 
 impl<T: Timestamp> CapabilityTrait<T> for InputCapability<T> {
     fn time(&self) -> &T { self.time() }
     fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool {
-        let internal_borrow = self.internal.borrow();
-        self.summaries.borrow().iter().any(|(output, summary)| {
+        let borrow = self.summaries.borrow();
+        self.internal.borrow().iter().enumerate().any(|(index, rc)| {
             // To be valid, the output buffer must match and the timestamp summary needs to be the default.
-            Rc::ptr_eq(&internal_borrow[*output], query_buffer) && summary.len() == 1 && summary[0] == Default::default()
+            Rc::ptr_eq(rc, query_buffer) && borrow[index].len() == 1 && borrow[index][0] == Default::default()
         })
     }
 }
 
 impl<T: Timestamp> InputCapability<T> {
     /// Creates a new capability reference at `time` while incrementing (and keeping a reference to)
     /// the provided [`ChangeBatch`].
-    pub(crate) fn new(internal: CapabilityUpdates<T>, summaries: Rc<RefCell<PortConnectivity<T::Summary>>>, guard: ConsumedGuard<T>) -> Self {
+    pub(crate) fn new(internal: CapabilityUpdates<T>, summaries: Rc<RefCell<Vec<Antichain<T::Summary>>>>, guard: ConsumedGuard<T>) -> Self {
         InputCapability {
             internal,
             summaries,
@@ -281,10 +281,10 @@ impl<T: Timestamp> InputCapability<T> {
     /// Delays capability for a specific output port.
     pub fn delayed_for_output(&self, new_time: &T, output_port: usize) -> Capability<T> {
         use crate::progress::timestamp::PathSummary;
-        if self.summaries.borrow()[&output_port].iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) {
+        if self.summaries.borrow()[output_port].iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) {
             Capability::new(new_time.clone(), self.internal.borrow()[output_port].clone())
         } else {
-            panic!("Attempted to delay to a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", new_time, self.summaries.borrow()[&output_port], self.time());
+            panic!("Attempted to delay to a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", new_time, self.summaries.borrow()[output_port], self.time());
         }
     }
 
@@ -305,11 +305,11 @@ impl<T: Timestamp> InputCapability<T> {
     pub fn retain_for_output(self, output_port: usize) -> Capability<T> {
         use crate::progress::timestamp::PathSummary;
         let self_time = self.time().clone();
-        if self.summaries.borrow()[&output_port].iter().flat_map(|summary| summary.results_in(&self_time)).any(|time| time.less_equal(&self_time)) {
+        if self.summaries.borrow()[output_port].iter().flat_map(|summary| summary.results_in(&self_time)).any(|time| time.less_equal(&self_time)) {
             Capability::new(self_time, self.internal.borrow()[output_port].clone())
         }
         else {
-            panic!("Attempted to retain a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", self_time, self.summaries.borrow()[&output_port], self_time);
+            panic!("Attempted to retain a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", self_time, self.summaries.borrow()[output_port], self_time);
         }
     }
 }
diff --git a/timely/src/dataflow/operators/core/feedback.rs b/timely/src/dataflow/operators/core/feedback.rs
@@ -113,8 +113,7 @@ impl<G: Scope, C: Container + Data> ConnectLoop<G, C> for StreamCore<G, C> {
         let summary = handle.summary;
         let mut output = handle.output;
 
-        let connection = Some((0, Antichain::from_elem(summary.clone()))).into_iter().collect();
-        let mut input = builder.new_input_connection(self, Pipeline, connection);
+        let mut input = builder.new_input_connection(self, Pipeline, vec![Antichain::from_elem(summary.clone())]);
 
         builder.build(move |_capability| move |_frontier| {
             let mut output = output.activate();
diff --git a/timely/src/dataflow/operators/core/input.rs b/timely/src/dataflow/operators/core/input.rs
@@ -7,9 +7,9 @@ use crate::container::{CapacityContainerBuilder, ContainerBuilder, PushInto};
 
 use crate::scheduling::{Schedule, Activator};
 
+use crate::progress::frontier::Antichain;
 use crate::progress::{Operate, operate::SharedProgress, Timestamp, ChangeBatch};
 use crate::progress::Source;
-use crate::progress::operate::Connectivity;
 
 use crate::{Container, Data};
 use crate::communication::Push;
@@ -205,7 +205,7 @@ impl<T:Timestamp> Operate<T> for Operator<T> {
     fn inputs(&self) -> usize { 0 }
     fn outputs(&self) -> usize { 1 }
 
-    fn get_internal_summary(&mut self) -> (Connectivity<T::Summary>, Rc<RefCell<SharedProgress<T>>>) {
+    fn get_internal_summary(&mut self) -> (Vec<Vec<Antichain<<T as Timestamp>::Summary>>>, Rc<RefCell<SharedProgress<T>>>) {
         self.shared_progress.borrow_mut().internals[0].update(T::minimum(), self.copies as i64);
         (Vec::new(), self.shared_progress.clone())
     }
diff --git a/timely/src/dataflow/operators/core/unordered_input.rs b/timely/src/dataflow/operators/core/unordered_input.rs
@@ -7,10 +7,10 @@ use crate::container::{ContainerBuilder, CapacityContainerBuilder};
 
 use crate::scheduling::{Schedule, ActivateOnDrop};
 
+use crate::progress::frontier::Antichain;
 use crate::progress::{Operate, operate::SharedProgress, Timestamp};
 use crate::progress::Source;
 use crate::progress::ChangeBatch;
-use crate::progress::operate::Connectivity;
 
 use crate::dataflow::channels::pushers::{Counter, Tee};
 use crate::dataflow::channels::pushers::buffer::{Buffer as PushBuffer, AutoflushSession};
@@ -134,7 +134,7 @@ impl<T:Timestamp> Operate<T> for UnorderedOperator<T> {
     fn inputs(&self) -> usize { 0 }
     fn outputs(&self) -> usize { 1 }
 
-    fn get_internal_summary(&mut self) -> (Connectivity<T::Summary>, Rc<RefCell<SharedProgress<T>>>) {
+    fn get_internal_summary(&mut self) -> (Vec<Vec<Antichain<<T as Timestamp>::Summary>>>, Rc<RefCell<SharedProgress<T>>>) {
         let mut borrow = self.internal.borrow_mut();
         for (time, count) in borrow.drain() {
             self.shared_progress.borrow_mut().internals[0].update(time, count * (self.peers as i64));
diff --git a/timely/src/dataflow/operators/generic/builder_raw.rs b/timely/src/dataflow/operators/generic/builder_raw.rs
@@ -12,7 +12,6 @@ use crate::scheduling::{Schedule, Activations};
 
 use crate::progress::{Source, Target};
 use crate::progress::{Timestamp, Operate, operate::SharedProgress, Antichain};
-use crate::progress::operate::{Connectivity, PortConnectivity};
 
 use crate::Container;
 use crate::dataflow::{StreamCore, Scope};
@@ -61,7 +60,7 @@ pub struct OperatorBuilder<G: Scope> {
     global: usize,
     address: Rc<[usize]>,    // path to the operator (ending with index).
     shape: OperatorShape,
-    summary: Connectivity<<G::Timestamp as Timestamp>::Summary>,
+    summary: Vec<Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>>,
 }
 
 impl<G: Scope> OperatorBuilder<G> {
@@ -108,12 +107,12 @@ impl<G: Scope> OperatorBuilder<G> {
     pub fn new_input<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P) -> P::Puller
         where
             P: ParallelizationContract<G::Timestamp, C> {
-        let connection = (0 .. self.shape.outputs).map(|o| (o, Antichain::from_elem(Default::default()))).collect();
+        let connection = vec![Antichain::from_elem(Default::default()); self.shape.outputs];
         self.new_input_connection(stream, pact, connection)
     }
 
     /// Adds a new input to a generic operator builder, returning the `Pull` implementor to use.
-    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: PortConnectivity<<G::Timestamp as Timestamp>::Summary>) -> P::Puller
+    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> P::Puller
     where
         P: ParallelizationContract<G::Timestamp, C> {
 
@@ -124,6 +123,7 @@ impl<G: Scope> OperatorBuilder<G> {
         stream.connect_to(target, sender, channel_id);
 
         self.shape.inputs += 1;
+        assert_eq!(self.shape.outputs, connection.len());
         self.summary.push(connection);
 
         receiver
@@ -132,20 +132,21 @@ impl<G: Scope> OperatorBuilder<G> {
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
     pub fn new_output<C: Container>(&mut self) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
 
-        let connection = (0 .. self.shape.inputs).map(|i| (i, Antichain::from_elem(Default::default()))).collect();
+        let connection = vec![Antichain::from_elem(Default::default()); self.shape.inputs];
         self.new_output_connection(connection)
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
-    pub fn new_output_connection<C: Container>(&mut self, connection: PortConnectivity<<G::Timestamp as Timestamp>::Summary>) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
+    pub fn new_output_connection<C: Container>(&mut self, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> (Tee<G::Timestamp, C>, StreamCore<G, C>) {
 
         let (targets, registrar) = Tee::<G::Timestamp,C>::new();
         let source = Source::new(self.index, self.shape.outputs);
         let stream = StreamCore::new(source, registrar, self.scope.clone());
 
         self.shape.outputs += 1;
-        for (input, entry) in connection.into_iter() {
-            self.summary[input].insert(self.shape.outputs - 1, entry);
+        assert_eq!(self.shape.inputs, connection.len());
+        for (summary, entry) in self.summary.iter_mut().zip(connection.into_iter()) {
+            summary.push(entry);
         }
 
         (targets, stream)
@@ -187,7 +188,7 @@ where
     logic: L,
     shared_progress: Rc<RefCell<SharedProgress<T>>>,
     activations: Rc<RefCell<Activations>>,
-    summary: Connectivity<T::Summary>,
+    summary: Vec<Vec<Antichain<T::Summary>>>,
 }
 
 impl<T, L> Schedule for OperatorCore<T, L>
@@ -212,7 +213,7 @@ where
     fn outputs(&self) -> usize { self.shape.outputs }
 
     // announce internal topology as fully connected, and hold all default capabilities.
-    fn get_internal_summary(&mut self) -> (Connectivity<T::Summary>, Rc<RefCell<SharedProgress<T>>>) {
+    fn get_internal_summary(&mut self) -> (Vec<Vec<Antichain<T::Summary>>>, Rc<RefCell<SharedProgress<T>>>) {
 
         // Request the operator to be scheduled at least once.
         self.activations.borrow_mut().activate(&self.address[..]);
diff --git a/timely/src/dataflow/operators/generic/builder_rc.rs b/timely/src/dataflow/operators/generic/builder_rc.rs
@@ -20,7 +20,7 @@ use crate::dataflow::operators::capability::Capability;
 use crate::dataflow::operators::generic::handles::{InputHandleCore, new_input_handle, OutputWrapper};
 use crate::dataflow::operators::generic::operator_info::OperatorInfo;
 use crate::dataflow::operators::generic::builder_raw::OperatorShape;
-use crate::progress::operate::PortConnectivity;
+
 use crate::logging::TimelyLogger as Logger;
 
 use super::builder_raw::OperatorBuilder as OperatorBuilderRaw;
@@ -33,7 +33,7 @@ pub struct OperatorBuilder<G: Scope> {
     consumed: Vec<Rc<RefCell<ChangeBatch<G::Timestamp>>>>,
     internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<G::Timestamp>>>>>>,
     /// For each input, a shared list of summaries to each output.
-    summaries: Vec<Rc<RefCell<PortConnectivity<<G::Timestamp as Timestamp>::Summary>>>>,
+    summaries: Vec<Rc<RefCell<Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>>>>,
     produced: Vec<Rc<RefCell<ChangeBatch<G::Timestamp>>>>,
     logging: Option<Logger>,
 }
@@ -64,7 +64,7 @@ impl<G: Scope> OperatorBuilder<G> {
     where
         P: ParallelizationContract<G::Timestamp, C> {
 
-        let connection = (0..self.builder.shape().outputs()).map(|o| (o, Antichain::from_elem(Default::default()))).collect();
+        let connection = (0..self.builder.shape().outputs()).map(|_| Antichain::from_elem(Default::default())).collect();
         self.new_input_connection(stream, pact, connection)
     }
 
@@ -76,7 +76,7 @@ impl<G: Scope> OperatorBuilder<G> {
     ///
     /// Commonly the connections are either the unit summary, indicating the same timestamp might be produced as output, or an empty
     /// antichain indicating that there is no connection from the input to the output.
-    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: PortConnectivity<<G::Timestamp as Timestamp>::Summary>) -> InputHandleCore<G::Timestamp, C, P::Puller>
+    pub fn new_input_connection<C: Container, P>(&mut self, stream: &StreamCore<G, C>, pact: P, connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>) -> InputHandleCore<G::Timestamp, C, P::Puller>
         where
             P: ParallelizationContract<G::Timestamp, C> {
 
@@ -94,7 +94,7 @@ impl<G: Scope> OperatorBuilder<G> {
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.
     pub fn new_output<CB: ContainerBuilder>(&mut self) -> (OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>, StreamCore<G, CB::Container>) {
-        let connection = (0..self.builder.shape().inputs()).map(|i| (i, Antichain::from_elem(Default::default()))).collect();
+        let connection = (0..self.builder.shape().inputs()).map(|_| Antichain::from_elem(Default::default())).collect();
         self.new_output_connection(connection)
     }
 
@@ -108,7 +108,7 @@ impl<G: Scope> OperatorBuilder<G> {
     /// antichain indicating that there is no connection from the input to the output.
     pub fn new_output_connection<CB: ContainerBuilder>(
         &mut self,
-        connection: PortConnectivity<<G::Timestamp as Timestamp>::Summary>
+        connection: Vec<Antichain<<G::Timestamp as Timestamp>::Summary>>
     ) -> (
         OutputWrapper<G::Timestamp, CB, Tee<G::Timestamp, CB::Container>>,
         StreamCore<G, CB::Container>
@@ -122,9 +122,8 @@ impl<G: Scope> OperatorBuilder<G> {
         let mut buffer = PushBuffer::new(PushCounter::new(tee));
         self.produced.push(buffer.inner().produced().clone());
 
-        for (input, summary) in connection.into_iter() {
-            let mut borrow = self.summaries[input].borrow_mut();
-            borrow.insert(self.shape().outputs() - 1, summary);
+        for (summary, connection) in self.summaries.iter().zip(connection.into_iter()) {
+            summary.borrow_mut().push(connection.clone());
         }
 
         (OutputWrapper::new(buffer, internal), stream)
diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs
diff --git a/timely/src/logging.rs b/timely/src/logging.rs
diff --git a/timely/src/progress/operate.rs b/timely/src/progress/operate.rs
diff --git a/timely/src/progress/reachability.rs b/timely/src/progress/reachability.rs
diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs
diff --git a/timely/tests/shape_scaling.rs b/timely/tests/shape_scaling.rs

Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@`
`2`	`2`
`3`	`3`	`use std::rc::Rc;`
`4`	`4`	`use std::cell::RefCell;`
	`5`	`+use std::sync::mpsc::Sender;`
`5`	`6`
`6`	`7`	`use crate::{Push, Pull};`
`7`	`8`
`@@ -51,8 +52,6 @@ impl<T, P: Push<T>> Push<T> for Pusher<T, P> {`
`51`	`52`	`}`
`52`	`53`	`}`
`53`	`54`
`54`		`-use crossbeam_channel::Sender;`
`55`		`-`
`56`	`55`	`/// The push half of an intra-thread channel.`
`57`	`56`	`pub struct ArcPusher<T, P: Push<T>> {`
`58`	`57`	`index: usize,`