From b5c45283d7bde398a185efb768bbc06de7ecb864 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 27 Feb 2026 18:50:38 -0500
Subject: [PATCH] Avoid allocations when sorting

---
 communication/src/allocator/zero_copy/push_pull.rs   | 2 +-
 timely/src/dataflow/operators/generic/handles.rs     | 2 +-
 timely/src/dataflow/operators/generic/notificator.rs | 2 +-
 timely/src/progress/change_batch.rs                  | 4 ++--
 timely/src/progress/subgraph.rs                      | 4 ++--
 timely/src/scheduling/activate.rs                    | 4 ++--
 timely/src/synchronization/sequence.rs               | 2 +-
 7 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/communication/src/allocator/zero_copy/push_pull.rs b/communication/src/allocator/zero_copy/push_pull.rs
index 758aad014..0e7ada2d3 100644
--- a/communication/src/allocator/zero_copy/push_pull.rs
+++ b/communication/src/allocator/zero_copy/push_pull.rs
@@ -36,7 +36,7 @@ impl<T, P: BytesPush> Pusher<T, P> {
 impl<T: Bytesable, P: BytesPush> Push<T> for Pusher<T, P> {
     #[inline]
     fn push(&mut self, element: &mut Option<T>) {
-        if let Some(ref mut element) = *element {
+        if let Some(ref element) = *element {
 
             // determine byte lengths and build header.
             let mut header = self.header;
diff --git a/timely/src/dataflow/operators/generic/handles.rs b/timely/src/dataflow/operators/generic/handles.rs
index a98850c0a..4ecf53d07 100644
--- a/timely/src/dataflow/operators/generic/handles.rs
+++ b/timely/src/dataflow/operators/generic/handles.rs
@@ -58,7 +58,7 @@ impl<T: Timestamp, C: Accountable, P: Pull<Message<T, C>>> InputHandleCore<T, C,
             let data = std::mem::take(data);
             self.staging.push_back((cap, data));
         }
-        self.staging.make_contiguous().sort_by(|x,y| x.0.time().cmp(&y.0.time()));
+        self.staging.make_contiguous().sort_unstable_by(|x,y| x.0.time().cmp(&y.0.time()));
 
         while let Some((cap, data)) = self.staging.pop_front() {
             self.staged.push(data);
diff --git a/timely/src/dataflow/operators/generic/notificator.rs b/timely/src/dataflow/operators/generic/notificator.rs
index e62fee677..a50d342c4 100644
--- a/timely/src/dataflow/operators/generic/notificator.rs
+++ b/timely/src/dataflow/operators/generic/notificator.rs
@@ -308,7 +308,7 @@ impl<T: Timestamp> FrontierNotificator<T> {
 
         if !self.pending.is_empty() {
 
-            self.pending.sort_by(|x,y| x.0.time().cmp(y.0.time()));
+            self.pending.sort_unstable_by(|x,y| x.0.time().cmp(y.0.time()));
             for i in 0 .. self.pending.len() - 1 {
                 if self.pending[i].0.time() == self.pending[i+1].0.time() {
                     self.pending[i+1].1 += self.pending[i].1;
diff --git a/timely/src/progress/change_batch.rs b/timely/src/progress/change_batch.rs
index 6a3c633b4..8da46c966 100644
--- a/timely/src/progress/change_batch.rs
+++ b/timely/src/progress/change_batch.rs
@@ -89,7 +89,7 @@ impl<T, const X: usize> ChangeBatch<T, X>
 where
     T: Ord,
 {
-    
+
     /// Allocates a new `ChangeBatch` with a single entry.
     ///
     /// # Examples
@@ -291,7 +291,7 @@ where
     #[inline]
     pub fn compact(&mut self) {
         if self.clean < self.updates.len() && self.updates.len() > 1 {
-            self.updates.sort_by(|x,y| x.0.cmp(&y.0));
+            self.updates.sort_unstable_by(|x,y| x.0.cmp(&y.0));
             for i in 0 .. self.updates.len() - 1 {
                 if self.updates[i].0 == self.updates[i+1].0 {
                     self.updates[i+1].1 += self.updates[i].1;
diff --git a/timely/src/progress/subgraph.rs b/timely/src/progress/subgraph.rs
index 868fd10d9..647b4c7c6 100644
--- a/timely/src/progress/subgraph.rs
+++ b/timely/src/progress/subgraph.rs
@@ -156,7 +156,7 @@ where
         // will depend on child summaries and capabilities, as well as edges in the subgraph.
 
         // perhaps first check that the children are sanely identified
-        self.children.sort_by(|x,y| x.index.cmp(&y.index));
+        self.children.sort_unstable_by(|x,y| x.index.cmp(&y.index));
         assert!(self.children.iter().enumerate().all(|(i,x)| i == x.index));
 
         let inputs = self.input_messages.len();
@@ -487,7 +487,7 @@ where
         }
 
         // Consider scheduling each recipient of progress information to shut down.
-        self.maybe_shutdown.sort();
+        self.maybe_shutdown.sort_unstable();
         self.maybe_shutdown.dedup();
         for child_index in self.maybe_shutdown.drain(..) {
             let child_state = self.pointstamp_tracker.node_state(child_index);
diff --git a/timely/src/scheduling/activate.rs b/timely/src/scheduling/activate.rs
index e073f823a..c94d7434d 100644
--- a/timely/src/scheduling/activate.rs
+++ b/timely/src/scheduling/activate.rs
@@ -86,7 +86,7 @@ impl Activations {
                 let moment = timer.elapsed() + delay;
                 self.queue.push(Reverse((moment, path.to_vec())));
             }
-        } 
+        }
         else {
             self.activate(path);
         }
@@ -115,7 +115,7 @@ impl Activations {
 
         {   // Scoped, to allow borrow to drop.
             let slices = &self.slices[..];
-            self.bounds.sort_by_key(|x| &slices[x.0 .. (x.0 + x.1)]);
+            self.bounds.sort_unstable_by_key(|x| &slices[x.0 .. (x.0 + x.1)]);
             self.bounds.dedup_by_key(|x| &slices[x.0 .. (x.0 + x.1)]);
         }
 
diff --git a/timely/src/synchronization/sequence.rs b/timely/src/synchronization/sequence.rs
index c02ea2506..e607badd6 100644
--- a/timely/src/synchronization/sequence.rs
+++ b/timely/src/synchronization/sequence.rs
@@ -182,7 +182,7 @@ impl<T: ExchangeData+Clone> Sequencer<T> {
                         }
                     });
 
-                    recvd.sort_by(|x,y| x.0.cmp(&y.0));
+                    recvd.sort_unstable_by(|x,y| x.0.cmp(&y.0));
 
                     if let Some(last) = recvd.last() {
                         let mut activator_borrow = activator_sink.borrow_mut();