chore: bump clippy version (#5173)

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ env:
   # Change to specific Rust release to pin
   rust_stable: stable
   rust_nightly: nightly-2022-11-03
-  rust_clippy: 1.56.0
+  rust_clippy: 1.60.0
   # When updating this, also update:
   # - README.md
   # - tokio/README.md

tokio-stream/src/stream_ext/collect.rs

@@ -195,11 +195,7 @@ where
         } else {
             let res = mem::replace(collection, Ok(U::initialize(sealed::Internal, 0, Some(0))));
 
-            if let Err(err) = res {
-                Err(err)
-            } else {
-                unreachable!();
-            }
+            Err(res.map(drop).unwrap_err())
         }
     }
 }

tokio-stream/src/stream_ext/timeout.rs

@@ -24,7 +24,7 @@ pin_project! {
 }
 
 /// Error returned by `Timeout`.
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct Elapsed(());
 
 impl<S: Stream> Timeout<S> {

tokio-stream/src/wrappers/broadcast.rs

@@ -18,7 +18,7 @@ pub struct BroadcastStream<T> {
 }
 
 /// An error returned from the inner stream of a [`BroadcastStream`].
-#[derive(Debug, PartialEq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone)]
 pub enum BroadcastStreamRecvError {
     /// The receiver lagged too far behind. Attempting to receive again will
     /// return the oldest message still retained by the channel.

tokio-util/src/time/delay_queue.rs

@@ -275,7 +275,7 @@ impl<T> SlabStorage<T> {
     fn remap_key(&self, key: &Key) -> Option<KeyInternal> {
         let key_map = &self.key_map;
         if self.compact_called {
-            key_map.get(&*key).copied()
+            key_map.get(key).copied()
         } else {
             Some((*key).into())
         }

tokio-util/tests/context.rs

@@ -21,5 +21,5 @@ fn tokio_context_with_another_runtime() {
 
     // Without the `HandleExt.wrap()` there would be a panic because there is
     // no timer running, since it would be referencing runtime r1.
-    let _ = rt1.block_on(rt2.wrap(async move { sleep(Duration::from_millis(2)).await }));
+    rt1.block_on(rt2.wrap(async move { sleep(Duration::from_millis(2)).await }));
 }

tokio-util/tests/sync_cancellation_token.rs

@@ -206,9 +206,6 @@ fn create_child_token_after_parent_was_cancelled() {
                 parent_fut.as_mut().poll(&mut Context::from_waker(&waker))
             );
             assert_eq!(wake_counter, 0);
-
-            drop(child_fut);
-            drop(parent_fut);
         }
 
         if drop_child_first {

tokio/src/io/ready.rs

@@ -12,7 +12,7 @@ const WRITE_CLOSED: usize = 0b0_1000;
 ///
 /// `Ready` tracks which operation an I/O resource is ready to perform.
 #[cfg_attr(docsrs, doc(cfg(feature = "net")))]
-#[derive(Clone, Copy, PartialEq, PartialOrd)]
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
 pub struct Ready(usize);
 
 impl Ready {

tokio/src/lib.rs

@@ -2,7 +2,8 @@
     clippy::cognitive_complexity,
     clippy::large_enum_variant,
     clippy::module_inception,
-    clippy::needless_doctest_main
+    clippy::needless_doctest_main,
+    clippy::declare_interior_mutable_const
 )]
 #![warn(
     missing_debug_implementations,

tokio/src/macros/thread_local.rs

@@ -13,7 +13,9 @@ macro_rules! tokio_thread_local {
 #[cfg(not(tokio_no_const_thread_local))]
 #[cfg(not(all(loom, test)))]
 macro_rules! tokio_thread_local {
-    ($($tts:tt)+) => { ::std::thread_local!{ $($tts)+ } }
+    ($($tts:tt)+) => {
+        ::std::thread_local!{ $($tts)+ }
+    }
 }
 
 #[cfg(tokio_no_const_thread_local)]

tokio/src/net/addr.rs

@@ -244,7 +244,7 @@ cfg_net! {
         type Future = <str as sealed::ToSocketAddrsPriv>::Future;
 
         fn to_socket_addrs(&self, _: sealed::Internal) -> Self::Future {
-            (&self[..]).to_socket_addrs(sealed::Internal)
+            self[..].to_socket_addrs(sealed::Internal)
         }
     }
 }

tokio/src/net/udp.rs

@@ -925,7 +925,7 @@ impl UdpSocket {
 
                 // Safety: We trust `UdpSocket::recv` to have filled up `n` bytes in the
                 // buffer.
-                let n = (&*self.io).recv(dst)?;
+                let n = (*self.io).recv(dst)?;
 
                 unsafe {
                     buf.advance_mut(n);
@@ -989,7 +989,7 @@ impl UdpSocket {
 
                 // Safety: We trust `UdpSocket::recv_from` to have filled up `n` bytes in the
                 // buffer.
-                let (n, addr) = (&*self.io).recv_from(dst)?;
+                let (n, addr) = (*self.io).recv_from(dst)?;
 
                 unsafe {
                     buf.advance_mut(n);

tokio/src/net/unix/datagram/socket.rs

@@ -846,7 +846,7 @@ impl UnixDatagram {
 
                 // Safety: We trust `UnixDatagram::recv_from` to have filled up `n` bytes in the
                 // buffer.
-                let (n, addr) = (&*self.io).recv_from(dst)?;
+                let (n, addr) = (*self.io).recv_from(dst)?;
 
                 unsafe {
                     buf.advance_mut(n);
@@ -909,7 +909,7 @@ impl UnixDatagram {
 
                 // Safety: We trust `UnixDatagram::recv` to have filled up `n` bytes in the
                 // buffer.
-                let n = (&*self.io).recv(dst)?;
+                let n = (*self.io).recv(dst)?;
 
                 unsafe {
                     buf.advance_mut(n);

tokio/src/runtime/io/mod.rs

@@ -165,11 +165,11 @@ impl Driver {
             self.resources.compact()
         }
 
-        let mut events = &mut self.events;
+        let events = &mut self.events;
 
         // Block waiting for an event to happen, peeling out how many events
         // happened.
-        match self.poll.poll(&mut events, max_wait) {
+        match self.poll.poll(events, max_wait) {
             Ok(_) => {}
             Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
             #[cfg(tokio_wasi)]

tokio/src/sync/batch_semaphore.rs

@@ -49,7 +49,7 @@ struct Waitlist {
 /// Error returned from the [`Semaphore::try_acquire`] function.
 ///
 /// [`Semaphore::try_acquire`]: crate::sync::Semaphore::try_acquire
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Eq)]
 pub enum TryAcquireError {
     /// The semaphore has been [closed] and cannot issue new permits.
     ///

tokio/src/sync/broadcast.rs

@@ -230,7 +230,7 @@ pub mod error {
     ///
     /// [`recv`]: crate::sync::broadcast::Receiver::recv
     /// [`Receiver`]: crate::sync::broadcast::Receiver
-    #[derive(Debug, PartialEq, Clone)]
+    #[derive(Debug, PartialEq, Eq, Clone)]
     pub enum RecvError {
         /// There are no more active senders implying no further messages will ever
         /// be sent.
@@ -258,7 +258,7 @@ pub mod error {
     ///
     /// [`try_recv`]: crate::sync::broadcast::Receiver::try_recv
     /// [`Receiver`]: crate::sync::broadcast::Receiver
-    #[derive(Debug, PartialEq, Clone)]
+    #[derive(Debug, PartialEq, Eq, Clone)]
     pub enum TryRecvError {
         /// The channel is currently empty. There are still active
         /// [`Sender`] handles, so data may yet become available.

tokio/src/sync/once_cell.rs

@@ -106,7 +106,7 @@ impl<T> Drop for OnceCell<T> {
         if self.initialized_mut() {
             unsafe {
                 self.value
-                    .with_mut(|ptr| ptr::drop_in_place((&mut *ptr).as_mut_ptr()));
+                    .with_mut(|ptr| ptr::drop_in_place((*ptr).as_mut_ptr()));
             };
         }
     }
@@ -416,7 +416,7 @@ unsafe impl<T: Send> Send for OnceCell<T> {}
 /// Errors that can be returned from [`OnceCell::set`].
 ///
 /// [`OnceCell::set`]: crate::sync::OnceCell::set
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Eq)]
 pub enum SetError<T> {
     /// The cell was already initialized when [`OnceCell::set`] was called.
     ///

tokio/src/sync/oneshot.rs

@@ -407,21 +407,21 @@ impl Task {
         F: FnOnce(&Waker) -> R,
     {
         self.0.with(|ptr| {
-            let waker: *const Waker = (&*ptr).as_ptr();
+            let waker: *const Waker = (*ptr).as_ptr();
             f(&*waker)
         })
     }
 
     unsafe fn drop_task(&self) {
         self.0.with_mut(|ptr| {
-            let ptr: *mut Waker = (&mut *ptr).as_mut_ptr();
+            let ptr: *mut Waker = (*ptr).as_mut_ptr();
             ptr.drop_in_place();
         });
     }
 
     unsafe fn set_task(&self, cx: &mut Context<'_>) {
         self.0.with_mut(|ptr| {
-            let ptr: *mut Waker = (&mut *ptr).as_mut_ptr();
+            let ptr: *mut Waker = (*ptr).as_mut_ptr();
             ptr.write(cx.waker().clone());
         });
     }

tokio/src/time/error.rs

@@ -44,7 +44,7 @@ impl From<Kind> for Error {
 ///
 /// This error is returned when a timeout expires before the function was able
 /// to finish.
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct Elapsed(());
 
 #[derive(Debug)]

tokio/tests/fs_file.rs

@@ -73,7 +73,7 @@ async fn coop() {
         let mut buf = [0; 1024];
 
         loop {
-            file.read(&mut buf).await.unwrap();
+            let _ = file.read(&mut buf).await.unwrap();
             file.seek(std::io::SeekFrom::Start(0)).await.unwrap();
         }
     });

tokio/tests/io_buf_writer.rs

@@ -70,15 +70,15 @@ where
 async fn buf_writer() {
     let mut writer = BufWriter::with_capacity(2, Vec::new());
 
-    writer.write(&[0, 1]).await.unwrap();
+    assert_eq!(writer.write(&[0, 1]).await.unwrap(), 2);
     assert_eq!(writer.buffer(), []);
     assert_eq!(*writer.get_ref(), [0, 1]);
 
-    writer.write(&[2]).await.unwrap();
+    assert_eq!(writer.write(&[2]).await.unwrap(), 1);
     assert_eq!(writer.buffer(), [2]);
     assert_eq!(*writer.get_ref(), [0, 1]);
 
-    writer.write(&[3]).await.unwrap();
+    assert_eq!(writer.write(&[3]).await.unwrap(), 1);
     assert_eq!(writer.buffer(), [2, 3]);
     assert_eq!(*writer.get_ref(), [0, 1]);
 
@@ -86,20 +86,20 @@ async fn buf_writer() {
     assert_eq!(writer.buffer(), []);
     assert_eq!(*writer.get_ref(), [0, 1, 2, 3]);
 
-    writer.write(&[4]).await.unwrap();
-    writer.write(&[5]).await.unwrap();
+    assert_eq!(writer.write(&[4]).await.unwrap(), 1);
+    assert_eq!(writer.write(&[5]).await.unwrap(), 1);
     assert_eq!(writer.buffer(), [4, 5]);
     assert_eq!(*writer.get_ref(), [0, 1, 2, 3]);
 
-    writer.write(&[6]).await.unwrap();
+    assert_eq!(writer.write(&[6]).await.unwrap(), 1);
     assert_eq!(writer.buffer(), [6]);
     assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5]);
 
-    writer.write(&[7, 8]).await.unwrap();
+    assert_eq!(writer.write(&[7, 8]).await.unwrap(), 2);
     assert_eq!(writer.buffer(), []);
     assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5, 6, 7, 8]);
 
-    writer.write(&[9, 10, 11]).await.unwrap();
+    assert_eq!(writer.write(&[9, 10, 11]).await.unwrap(), 3);
     assert_eq!(writer.buffer(), []);
     assert_eq!(*writer.get_ref(), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]);
 
@@ -111,7 +111,7 @@ async fn buf_writer() {
 #[tokio::test]
 async fn buf_writer_inner_flushes() {
     let mut w = BufWriter::with_capacity(3, Vec::new());
-    w.write(&[0, 1]).await.unwrap();
+    assert_eq!(w.write(&[0, 1]).await.unwrap(), 2);
     assert_eq!(*w.get_ref(), []);
     w.flush().await.unwrap();
     let w = w.into_inner();
@@ -135,15 +135,15 @@ async fn buf_writer_seek() {
 async fn maybe_pending_buf_writer() {
     let mut writer = BufWriter::with_capacity(2, MaybePending::new(Vec::new()));
 
-    writer.write(&[0, 1]).await.unwrap();
+    assert_eq!(writer.write(&[0, 1]).await.unwrap(), 2);
     assert_eq!(writer.buffer(), []);
     assert_eq!(&writer.get_ref().inner, &[0, 1]);
 
-    writer.write(&[2]).await.unwrap();
+    assert_eq!(writer.write(&[2]).await.unwrap(), 1);
     assert_eq!(writer.buffer(), [2]);
     assert_eq!(&writer.get_ref().inner, &[0, 1]);
 
-    writer.write(&[3]).await.unwrap();
+    assert_eq!(writer.write(&[3]).await.unwrap(), 1);
     assert_eq!(writer.buffer(), [2, 3]);
     assert_eq!(&writer.get_ref().inner, &[0, 1]);
 
@@ -151,20 +151,20 @@ async fn maybe_pending_buf_writer() {
     assert_eq!(writer.buffer(), []);
     assert_eq!(&writer.get_ref().inner, &[0, 1, 2, 3]);
 
-    writer.write(&[4]).await.unwrap();
-    writer.write(&[5]).await.unwrap();
+    assert_eq!(writer.write(&[4]).await.unwrap(), 1);
+    assert_eq!(writer.write(&[5]).await.unwrap(), 1);
     assert_eq!(writer.buffer(), [4, 5]);
     assert_eq!(&writer.get_ref().inner, &[0, 1, 2, 3]);
 
-    writer.write(&[6]).await.unwrap();
+    assert_eq!(writer.write(&[6]).await.unwrap(), 1);
     assert_eq!(writer.buffer(), [6]);
     assert_eq!(writer.get_ref().inner, &[0, 1, 2, 3, 4, 5]);
 
-    writer.write(&[7, 8]).await.unwrap();
+    assert_eq!(writer.write(&[7, 8]).await.unwrap(), 2);
     assert_eq!(writer.buffer(), []);
     assert_eq!(writer.get_ref().inner, &[0, 1, 2, 3, 4, 5, 6, 7, 8]);
 
-    writer.write(&[9, 10, 11]).await.unwrap();
+    assert_eq!(writer.write(&[9, 10, 11]).await.unwrap(), 3);
     assert_eq!(writer.buffer(), []);
     assert_eq!(
         writer.get_ref().inner,
@@ -182,7 +182,7 @@ async fn maybe_pending_buf_writer() {
 #[tokio::test]
 async fn maybe_pending_buf_writer_inner_flushes() {
     let mut w = BufWriter::with_capacity(3, MaybePending::new(Vec::new()));
-    w.write(&[0, 1]).await.unwrap();
+    assert_eq!(w.write(&[0, 1]).await.unwrap(), 2);
     assert_eq!(&w.get_ref().inner, &[]);
     w.flush().await.unwrap();
     let w = w.into_inner().inner;

tokio/tests/io_mem_stream.rs

@@ -113,7 +113,7 @@ async fn duplex_is_cooperative() {
                 let buf = [3u8; 4096];
                 tx.write_all(&buf).await.unwrap();
                 let mut buf = [0u8; 4096];
-                rx.read(&mut buf).await.unwrap();
+                let _ = rx.read(&mut buf).await.unwrap();
             }
         } => {},
         _ = tokio::task::yield_now() => {}

tokio/tests/macros_test.rs

@@ -50,6 +50,7 @@ pub async fn issue_4175_test() -> std::io::Result<()> {
 }
 
 // https://github.com/tokio-rs/tokio/issues/4175
+#[allow(clippy::let_unit_value)]
 pub mod clippy_semicolon_if_nothing_returned {
     #![deny(clippy::semicolon_if_nothing_returned)]
 

tokio/tests/sync_mpsc.rs

@@ -840,8 +840,8 @@ impl Drop for Msg {
 async fn test_msgs_dropped_on_rx_drop() {
     let (tx, mut rx) = mpsc::channel(3);
 
-    let _ = tx.send(Msg {}).await.unwrap();
-    let _ = tx.send(Msg {}).await.unwrap();
+    tx.send(Msg {}).await.unwrap();
+    tx.send(Msg {}).await.unwrap();
 
     // This msg will be pending and should be dropped when `rx` is dropped
     let sent_fut = tx.send(Msg {});
@@ -849,7 +849,7 @@ async fn test_msgs_dropped_on_rx_drop() {
     let _ = rx.recv().await.unwrap();
     let _ = rx.recv().await.unwrap();
 
-    let _ = sent_fut.await.unwrap();
+    sent_fut.await.unwrap();
 
     drop(rx);
 
@@ -936,7 +936,7 @@ async fn test_tx_capacity() {
     assert_eq!(tx.capacity(), 9);
     assert_eq!(tx.max_capacity(), 10);
 
-    let _sent = tx.send(()).await.unwrap();
+    tx.send(()).await.unwrap();
     // after send, capacity should drop by one again
     assert_eq!(tx.capacity(), 8);
     assert_eq!(tx.max_capacity(), 10);

tokio/tests/task_local.rs

@@ -1,4 +1,5 @@
 #![cfg(all(feature = "full", not(tokio_wasi)))] // Wasi doesn't support threads
+#![allow(clippy::declare_interior_mutable_const)]
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};

tokio/tests/task_panic.rs

@@ -1,4 +1,5 @@
 #![warn(rust_2018_idioms)]
+#![allow(clippy::declare_interior_mutable_const)]
 #![cfg(all(feature = "full", not(tokio_wasi)))]
 
 use futures::future;
@@ -78,7 +79,7 @@ fn local_key_sync_scope_panic_caller() -> Result<(), Box<dyn Error>> {
     let panic_location_file = test_panic(|| {
         NUMBER.sync_scope(1, || {
             NUMBER.with(|_| {
-                let _ = NUMBER.sync_scope(1, || {});
+                NUMBER.sync_scope(1, || {});
             });
         });
     });

tokio/tests/tcp_split.rs

@@ -36,7 +36,7 @@ async fn split() -> Result<()> {
     assert_eq!(peek_len1, read_len);
     assert_eq!(&read_buf[..read_len], MSG);
 
-    write_half.write(MSG).await?;
+    assert_eq!(write_half.write(MSG).await?, MSG.len());
     handle.join().unwrap();
     Ok(())
 }

tokio/tests/time_sleep.rs

@@ -315,18 +315,18 @@ async fn drop_from_wake() {
 
     tokio::time::pause();
 
-    let mut lock = list.lock().unwrap();
+    {
+        let mut lock = list.lock().unwrap();
 
-    for _ in 0..100 {
-        let mut timer = Box::pin(tokio::time::sleep(Duration::from_millis(10)));
+        for _ in 0..100 {
+            let mut timer = Box::pin(tokio::time::sleep(Duration::from_millis(10)));
 
-        let _ = timer.as_mut().poll(&mut Context::from_waker(&arc_wake));
+            let _ = timer.as_mut().poll(&mut Context::from_waker(&arc_wake));
 
-        lock.push(timer);
+            lock.push(timer);
+        }
     }
 
-    drop(lock);
-
     tokio::time::sleep(Duration::from_millis(11)).await;
 
     assert!(

tokio/tests/udp.rs

@@ -39,7 +39,7 @@ async fn send_recv_poll() -> std::io::Result<()> {
 
     let mut recv_buf = [0u8; 32];
     let mut read = ReadBuf::new(&mut recv_buf);
-    let _len = poll_fn(|cx| receiver.poll_recv(cx, &mut read)).await?;
+    poll_fn(|cx| receiver.poll_recv(cx, &mut read)).await?;
 
     assert_eq!(read.filled(), MSG);
     Ok(())

tokio/tests/uds_datagram.rs

@@ -177,7 +177,7 @@ async fn send_recv_poll() -> std::io::Result<()> {
 
     let mut recv_buf = [0u8; 32];
     let mut read = ReadBuf::new(&mut recv_buf);
-    let _len = poll_fn(|cx| receiver.poll_recv(cx, &mut read)).await?;
+    poll_fn(|cx| receiver.poll_recv(cx, &mut read)).await?;
 
     assert_eq!(read.filled(), msg);
     Ok(())