fix new lints since rust-1.90.0 upgrade (#1438)

Summary:
we upgraded the rust toolchain to 1.90.0 last week introducing new lints.
```bash
cargo clippy --fix --lib -p ndslice
cargo clippy --fix --lib -p hyperactor
cargo clippy --fix --lib -p hyperactor_mesh
```

Pull Request resolved: #1438

Reviewed By: dulinriley, mariusae

Differential Revision: D83939684

Pulled By: shayne-fletcher

fbshipit-source-id: 3f47c3bc497b630ffd837fd1494238c428300b7a

Author: shayne-fletcher

hyperactor/src/actor.rs

@@ -492,11 +492,9 @@ impl ActorStatus {
             Self::Initializing => Self::Initializing,
             Self::Client => Self::Client,
             Self::Idle => Self::Idle,
-            Self::Processing(instant, handler) => {
-                Self::Processing(instant.clone(), handler.clone())
-            }
-            Self::Saving(instant) => Self::Saving(instant.clone()),
-            Self::Loading(instant) => Self::Loading(instant.clone()),
+            Self::Processing(instant, handler) => Self::Processing(*instant, handler.clone()),
+            Self::Saving(instant) => Self::Saving(*instant),
+            Self::Loading(instant) => Self::Loading(*instant),
             Self::Stopping => Self::Stopping,
             Self::Stopped => Self::Stopped,
             Self::Failed(err) => Self::Failed(err.clone()),
@@ -521,7 +519,7 @@ impl fmt::Display for ActorStatus {
                     "processing for {}ms",
                     RealClock
                         .system_time_now()
-                        .duration_since(instant.clone())
+                        .duration_since(*instant)
                         .unwrap_or_default()
                         .as_millis()
                 )
@@ -533,7 +531,7 @@ impl fmt::Display for ActorStatus {
                     handler,
                     RealClock
                         .system_time_now()
-                        .duration_since(instant.clone())
+                        .duration_since(*instant)
                         .unwrap_or_default()
                         .as_millis()
                 )
@@ -546,7 +544,7 @@ impl fmt::Display for ActorStatus {
                     arm,
                     RealClock
                         .system_time_now()
-                        .duration_since(instant.clone())
+                        .duration_since(*instant)
                         .unwrap_or_default()
                         .as_millis()
                 )
@@ -557,7 +555,7 @@ impl fmt::Display for ActorStatus {
                     "saving for {}ms",
                     RealClock
                         .system_time_now()
-                        .duration_since(instant.clone())
+                        .duration_since(*instant)
                         .unwrap_or_default()
                         .as_millis()
                 )
@@ -568,7 +566,7 @@ impl fmt::Display for ActorStatus {
                     "loading for {}ms",
                     RealClock
                         .system_time_now()
-                        .duration_since(instant.clone())
+                        .duration_since(*instant)
                         .unwrap_or_default()
                         .as_millis()
                 )

hyperactor/src/attrs.rs

@@ -343,7 +343,7 @@ impl AttrValue for std::time::Duration {
 
 impl AttrValue for std::time::SystemTime {
     fn display(&self) -> String {
-        let datetime: DateTime<Utc> = self.clone().into();
+        let datetime: DateTime<Utc> = (*self).into();
         datetime.to_rfc3339()
     }
 

hyperactor/src/channel/net.rs

@@ -393,17 +393,14 @@ impl<M: RemoteMessage> NetTx<M> {
             }
 
             fn requeue_unacked(&mut self, unacked: MessageDeque<M>) {
-                match (unacked.back(), self.deque.front()) {
-                    (Some(last), Some(first)) => {
-                        assert!(
-                            last.seq < first.seq,
-                            "{}: seq should be in ascending order, but got {} vs {}",
-                            self.log_id,
-                            last.seq,
-                            first.seq,
-                        );
-                    }
-                    _ => (),
+                if let (Some(last), Some(first)) = (unacked.back(), self.deque.front()) {
+                    assert!(
+                        last.seq < first.seq,
+                        "{}: seq should be in ascending order, but got {} vs {}",
+                        self.log_id,
+                        last.seq,
+                        first.seq,
+                    );
                 }
 
                 let mut outbox = unacked;
@@ -537,7 +534,7 @@ impl<M: RemoteMessage> NetTx<M> {
                     Some(msg) => {
                         RealClock
                             .sleep_until(
-                                msg.received_at.clone()
+                                msg.received_at
                                     + config::global::get(config::MESSAGE_DELIVERY_TIMEOUT),
                             )
                             .await
@@ -964,10 +961,10 @@ impl<M: RemoteMessage> NetTx<M> {
                 }
             };
 
-            if !matches!(state, State::Closing { .. }) {
-                if let Conn::Disconnected(ref mut backoff) = conn {
-                    RealClock.sleep(backoff.next_backoff().unwrap()).await;
-                }
+            if !matches!(state, State::Closing { .. })
+                && let Conn::Disconnected(ref mut backoff) = conn
+            {
+                RealClock.sleep(backoff.next_backoff().unwrap()).await;
             }
         }; // loop
         tracing::debug!("{log_id}: NetTx exited its loop with state: {state}");
@@ -1435,13 +1432,10 @@ impl<S: AsyncRead + AsyncWrite + Send + 'static + Unpin> ServerConn<S> {
         let mut final_ack = final_next.ack;
         // Flush any ongoing write.
         if self.write_state.is_writing() {
-            match self.write_state.send().await {
-                Ok(acked_seq) => {
-                    if acked_seq > final_ack {
-                        final_ack = acked_seq;
-                    }
+            if let Ok(acked_seq) = self.write_state.send().await {
+                if acked_seq > final_ack {
+                    final_ack = acked_seq;
                 }
-                Err(_) => (),
             };
         }
         // best effort: "flush" any remaining ack before closing this session
@@ -1480,26 +1474,23 @@ impl<S: AsyncRead + AsyncWrite + Send + 'static + Unpin> ServerConn<S> {
             let Ok(writer) = replace(&mut self.write_state, WriteState::Broken).into_idle() else {
                 panic!("illegal state");
             };
-            match serialize_response(NetRxResponse::Reject) {
-                Ok(data) => {
-                    match FrameWrite::new(
-                        writer,
-                        data,
-                        config::global::get(config::CODEC_MAX_FRAME_LENGTH),
-                    ) {
-                        Ok(fw) => {
-                            self.write_state = WriteState::Writing(fw, 0);
-                            let _ = self.write_state.send().await;
-                        }
-                        Err((w, e)) => {
-                            debug_assert_eq!(e.kind(), io::ErrorKind::InvalidData);
-                            tracing::debug!("failed to create reject frame (should be tiny): {e}");
-                            self.write_state = WriteState::Idle(w);
-                            // drop the reject; we're closing anyway
-                        }
+            if let Ok(data) = serialize_response(NetRxResponse::Reject) {
+                match FrameWrite::new(
+                    writer,
+                    data,
+                    config::global::get(config::CODEC_MAX_FRAME_LENGTH),
+                ) {
+                    Ok(fw) => {
+                        self.write_state = WriteState::Writing(fw, 0);
+                        let _ = self.write_state.send().await;
+                    }
+                    Err((w, e)) => {
+                        debug_assert_eq!(e.kind(), io::ErrorKind::InvalidData);
+                        tracing::debug!("failed to create reject frame (should be tiny): {e}");
+                        self.write_state = WriteState::Idle(w);
+                        // drop the reject; we're closing anyway
                     }
                 }
-                Err(_) => (),
             };
         }
 
@@ -2104,7 +2095,7 @@ pub(crate) mod tcp {
         type Stream = TcpStream;
 
         fn dest(&self) -> ChannelAddr {
-            ChannelAddr::Tcp(self.0.clone())
+            ChannelAddr::Tcp(self.0)
         }
 
         async fn connect(&self) -> Result<Self::Stream, ClientError> {

hyperactor/src/clock.rs

@@ -43,7 +43,7 @@ struct SimTime {
 static SIM_TIME: LazyLock<SimTime> = LazyLock::new(|| {
     let now = tokio::time::Instant::now();
     SimTime {
-        start: now.clone(),
+        start: now,
         now: Mutex::new(now),
         system_start: SystemTime::now(),
     }
@@ -222,11 +222,11 @@ impl Clock for SimClock {
     }
     /// Get the current time according to the simnet
     fn now(&self) -> tokio::time::Instant {
-        SIM_TIME.now.lock().unwrap().clone()
+        *SIM_TIME.now.lock().unwrap()
     }
 
     fn system_time_now(&self) -> SystemTime {
-        SIM_TIME.system_start.clone() + self.now().duration_since(SIM_TIME.start)
+        SIM_TIME.system_start + self.now().duration_since(SIM_TIME.start)
     }
 
     #[allow(clippy::disallowed_methods)]
@@ -290,7 +290,7 @@ impl SimClock {
 
     /// Instant marking the start of the simulation
     pub fn start(&self) -> tokio::time::Instant {
-        SIM_TIME.start.clone()
+        SIM_TIME.start
     }
 }
 

hyperactor/src/config/global.rs

@@ -353,7 +353,7 @@ fn test_override_index(layers: &Layers) -> Option<usize> {
         .position(|l| matches!(l.source, Source::TestOverride))
 }
 
-fn ensure_test_override_layer_mut<'a>(layers: &'a mut Layers) -> &'a mut Attrs {
+fn ensure_test_override_layer_mut(layers: &mut Layers) -> &mut Attrs {
     if let Some(i) = test_override_index(layers) {
         return &mut layers.ordered[i].attrs;
     }

hyperactor/src/mailbox.rs

@@ -1719,10 +1719,10 @@ impl<M> PortReceiver<M> {
     pub fn try_recv(&mut self) -> Result<Option<M>, MailboxError> {
         let mut next = self.receiver.try_recv();
         // To coalesce, drain the mpsc queue and only keep the last one.
-        if self.coalesce {
-            if let Some(latest) = self.drain().pop() {
-                next = Ok(latest);
-            }
+        if self.coalesce
+            && let Some(latest) = self.drain().pop()
+        {
+            next = Ok(latest);
         }
         match next {
             Ok(msg) => Ok(Some(msg)),
@@ -1740,10 +1740,10 @@ impl<M> PortReceiver<M> {
         let mut next = self.receiver.recv().await;
         // To coalesce, get the last message from the queue if there are
         // more on the mspc queue.
-        if self.coalesce {
-            if let Some(latest) = self.drain().pop() {
-                next = Some(latest);
-            }
+        if self.coalesce
+            && let Some(latest) = self.drain().pop()
+        {
+            next = Some(latest);
         }
         next.ok_or(MailboxError::new(
             self.actor_id().clone(),
@@ -2152,6 +2152,12 @@ pub struct MailboxMuxer {
     mailboxes: Arc<DashMap<ActorId, Box<dyn MailboxSender + Send + Sync>>>,
 }
 
+impl Default for MailboxMuxer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl MailboxMuxer {
     /// Create a new, empty, muxer.
     pub fn new() -> Self {
@@ -2216,6 +2222,12 @@ pub struct MailboxRouter {
     entries: Arc<RwLock<BTreeMap<Reference, Arc<dyn MailboxSender + Send + Sync>>>>,
 }
 
+impl Default for MailboxRouter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl MailboxRouter {
     /// Create a new, empty router.
     pub fn new() -> Self {
@@ -2364,6 +2376,12 @@ pub struct DialMailboxRouter {
     direct_addressed_remote_only: bool,
 }
 
+impl Default for DialMailboxRouter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl DialMailboxRouter {
     /// Create a new [`DialMailboxRouter`] with an empty routing table.
     pub fn new() -> Self {
@@ -2403,11 +2421,11 @@ impl DialMailboxRouter {
     /// cache to ensure fresh routing on next use.
     pub fn bind(&self, dest: Reference, addr: ChannelAddr) {
         if let Ok(mut w) = self.address_book.write() {
-            if let Some(old_addr) = w.insert(dest.clone(), addr.clone()) {
-                if old_addr != addr {
-                    tracing::info!("rebinding {:?} from {:?} to {:?}", dest, old_addr, addr);
-                    self.sender_cache.remove(&old_addr);
-                }
+            if let Some(old_addr) = w.insert(dest.clone(), addr.clone())
+                && old_addr != addr
+            {
+                tracing::info!("rebinding {:?} from {:?} to {:?}", dest, old_addr, addr);
+                self.sender_cache.remove(&old_addr);
             }
         } else {
             tracing::error!("address book poisoned during bind of {:?}", dest);

hyperactor/src/mailbox/durable_mailbox_sender.rs

@@ -185,6 +185,12 @@ pub mod test_utils {
         observer: Option<mpsc::UnboundedSender<(String, M)>>,
     }
 
+    impl<M: RemoteMessage> Default for TestLog<M> {
+        fn default() -> Self {
+            Self::new()
+        }
+    }
+
     impl<M: RemoteMessage> TestLog<M> {
         /// Create a new, empty [`TestLog`].
         pub fn new() -> Self {
@@ -255,10 +261,10 @@ pub mod test_utils {
                 .map(|(seq_id, msg)| Ok((*seq_id, msg.clone())))
                 .collect();
             for entry in filtered_items.iter() {
-                if let Some(observer) = &self.observer {
-                    if let Ok((_, msg)) = entry.as_ref() {
-                        observer.send(("read".to_string(), msg.clone())).unwrap();
-                    }
+                if let Some(observer) = &self.observer
+                    && let Ok((_, msg)) = entry.as_ref()
+                {
+                    observer.send(("read".to_string(), msg.clone())).unwrap();
                 }
             }
             Ok(futures::stream::iter(filtered_items.into_iter()))

hyperactor/src/proc.rs

@@ -712,7 +712,7 @@ impl Proc {
             .iter()
             .filter(|(actor_id, _)| !stopped_actors.contains(actor_id))
             .map(|(actor_id, _)| {
-                let f = self.abort_root_actor(actor_id, this_handle.clone());
+                let f = self.abort_root_actor(actor_id, this_handle);
                 async move {
                     let _ = if let Some(f) = f { Some(f.await) } else { None };
                     // If `is_none(&_)` then the proc's `ledger.roots`
@@ -1200,11 +1200,8 @@ impl<A: Actor> Instance<A> {
 
         let (mut signal_receiver, _) = actor_loop_receivers;
         while self.cell.child_count() > 0 {
-            match signal_receiver.recv().await? {
-                Signal::ChildStopped(pid) => {
-                    assert!(self.cell.get_child(pid).is_none());
-                }
-                _ => (),
+            if let Signal::ChildStopped(pid) = signal_receiver.recv().await? {
+                assert!(self.cell.get_child(pid).is_none());
             }
         }
 
@@ -1345,7 +1342,7 @@ impl<A: Actor> Instance<A> {
             )
         });
 
-        let _ = self.change_status(ActorStatus::Processing(
+        self.change_status(ActorStatus::Processing(
             self.clock().system_time_now(),
             handler,
         ));
@@ -1427,7 +1424,7 @@ impl<A: Actor> Instance<A> {
             ports: self.ports.clone(),
             status_tx: self.status_tx.clone(),
             sequencer: self.sequencer.clone(),
-            id: self.id.clone(),
+            id: self.id,
         }
     }
 
@@ -1801,6 +1798,12 @@ pub struct WeakInstanceCell {
     inner: Weak<InstanceState>,
 }
 
+impl Default for WeakInstanceCell {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl WeakInstanceCell {
     /// Create a new weak instance cell that is never upgradeable.
     pub fn new() -> Self {