chore: clippy and doc fixes (#6081)

tokio-macros/src/entry.rs

@@ -586,6 +586,6 @@ impl ToTokens for Body<'_> {
             for stmt in self.stmts {
                 stmt.to_tokens(tokens);
             }
-        })
+        });
     }
 }

tokio-macros/src/select.rs

@@ -73,27 +73,27 @@ fn clean_pattern(pat: &mut syn::Pat) {
             }
         }
         syn::Pat::Or(or) => {
-            for case in or.cases.iter_mut() {
+            for case in &mut or.cases {
                 clean_pattern(case);
             }
         }
         syn::Pat::Slice(slice) => {
-            for elem in slice.elems.iter_mut() {
+            for elem in &mut slice.elems {
                 clean_pattern(elem);
             }
         }
         syn::Pat::Struct(struct_pat) => {
-            for field in struct_pat.fields.iter_mut() {
+            for field in &mut struct_pat.fields {
                 clean_pattern(&mut field.pat);
             }
         }
         syn::Pat::Tuple(tuple) => {
-            for elem in tuple.elems.iter_mut() {
+            for elem in &mut tuple.elems {
                 clean_pattern(elem);
             }
         }
         syn::Pat::TupleStruct(tuple) => {
-            for elem in tuple.elems.iter_mut() {
+            for elem in &mut tuple.elems {
                 clean_pattern(elem);
             }
         }

tokio-stream/src/once.rs

@@ -35,7 +35,7 @@ impl<I> Unpin for Once<I> {}
 /// ```
 pub fn once<T>(value: T) -> Once<T> {
     Once {
-        iter: crate::iter(Some(value).into_iter()),
+        iter: crate::iter(Some(value)),
     }
 }
 

tokio-stream/src/wrappers/mpsc_bounded.rs

@@ -34,7 +34,7 @@ impl<T> ReceiverStream<T> {
     ///
     /// [`Permit`]: struct@tokio::sync::mpsc::Permit
     pub fn close(&mut self) {
-        self.inner.close()
+        self.inner.close();
     }
 }
 

tokio-stream/src/wrappers/mpsc_unbounded.rs

@@ -28,7 +28,7 @@ impl<T> UnboundedReceiverStream<T> {
     /// This prevents any further messages from being sent on the channel while
     /// still enabling the receiver to drain messages that are buffered.
     pub fn close(&mut self) {
-        self.inner.close()
+        self.inner.close();
     }
 }
 

tokio-test/src/io.rs

@@ -74,7 +74,7 @@ struct Inner {
 }
 
 impl Builder {
-    /// Return a new, empty `Builder.
+    /// Return a new, empty `Builder`.
     pub fn new() -> Self {
         Self::default()
     }
@@ -478,7 +478,7 @@ impl Drop for Mock {
             Action::Read(data) => assert!(data.is_empty(), "There is still data left to read."),
             Action::Write(data) => assert!(data.is_empty(), "There is still data left to write."),
             _ => (),
-        })
+        });
     }
 }
 /*

tokio-test/src/task.rs

@@ -127,7 +127,7 @@ impl<T: Future> Spawn<T> {
 }
 
 impl<T: Stream> Spawn<T> {
-    /// If `T` is a [`Stream`] then poll_next it. This will handle pinning and the context
+    /// If `T` is a [`Stream`] then `poll_next` it. This will handle pinning and the context
     /// type for the stream.
     pub fn poll_next(&mut self) -> Poll<Option<T::Item>> {
         let stream = self.future.as_mut();

tokio-util/src/either.rs

@@ -116,7 +116,7 @@ where
     }
 
     fn consume(self: Pin<&mut Self>, amt: usize) {
-        delegate_call!(self.consume(amt))
+        delegate_call!(self.consume(amt));
     }
 }
 

tokio-util/src/sync/cancellation_token.rs

@@ -133,7 +133,7 @@ impl Default for CancellationToken {
 }
 
 impl CancellationToken {
-    /// Creates a new CancellationToken in the non-cancelled state.
+    /// Creates a new `CancellationToken` in the non-cancelled state.
     pub fn new() -> CancellationToken {
         CancellationToken {
             inner: Arc::new(tree_node::TreeNode::new()),

tokio-util/src/sync/cancellation_token/tree_node.rs

@@ -1,12 +1,12 @@
 //! This mod provides the logic for the inner tree structure of the CancellationToken.
 //!
-//! CancellationTokens are only light handles with references to TreeNode.
-//! All the logic is actually implemented in the TreeNode.
+//! CancellationTokens are only light handles with references to [`TreeNode`].
+//! All the logic is actually implemented in the [`TreeNode`].
 //!
-//! A TreeNode is part of the cancellation tree and may have one parent and an arbitrary number of
+//! A [`TreeNode`] is part of the cancellation tree and may have one parent and an arbitrary number of
 //! children.
 //!
-//! A TreeNode can receive the request to perform a cancellation through a CancellationToken.
+//! A [`TreeNode`] can receive the request to perform a cancellation through a CancellationToken.
 //! This cancellation request will cancel the node and all of its descendants.
 //!
 //! As soon as a node cannot get cancelled any more (because it was already cancelled or it has no

tokio-util/src/sync/poll_semaphore.rs

@@ -29,7 +29,7 @@ impl PollSemaphore {
 
     /// Closes the semaphore.
     pub fn close(&self) {
-        self.semaphore.close()
+        self.semaphore.close();
     }
 
     /// Obtain a clone of the inner semaphore.

tokio/src/fs/dir_builder.rs

@@ -35,7 +35,7 @@ impl DirBuilder {
     /// let builder = DirBuilder::new();
     /// ```
     pub fn new() -> Self {
-        Default::default()
+        DirBuilder::default()
     }
 
     /// Indicates whether to create directories recursively (including all parent directories).

tokio/src/fs/file.rs

@@ -126,7 +126,7 @@ impl File {
     ///
     /// This function will return an error if called from outside of the Tokio
     /// runtime or if path does not already exist. Other errors may also be
-    /// returned according to OpenOptions::open.
+    /// returned according to `OpenOptions::open`.
     ///
     /// # Examples
     ///
@@ -367,7 +367,7 @@ impl File {
             } else {
                 std.set_len(size)
             }
-            .map(|_| 0); // the value is discarded later
+            .map(|()| 0); // the value is discarded later
 
             // Return the result as a seek
             (Operation::Seek(res), buf)
@@ -562,7 +562,7 @@ impl AsyncRead for File {
                             inner.state = State::Idle(Some(buf));
                             return Poll::Ready(Err(e));
                         }
-                        Operation::Write(Ok(_)) => {
+                        Operation::Write(Ok(())) => {
                             assert!(buf.is_empty());
                             inner.state = State::Idle(Some(buf));
                             continue;
@@ -877,7 +877,7 @@ impl Inner {
     async fn complete_inflight(&mut self) {
         use crate::future::poll_fn;
 
-        poll_fn(|cx| self.poll_complete_inflight(cx)).await
+        poll_fn(|cx| self.poll_complete_inflight(cx)).await;
     }
 
     fn poll_complete_inflight(&mut self, cx: &mut Context<'_>) -> Poll<()> {

tokio/src/future/maybe_done.rs

@@ -1,4 +1,4 @@
-//! Definition of the MaybeDone combinator.
+//! Definition of the [`MaybeDone`] combinator.
 
 use std::future::Future;
 use std::mem;

tokio/src/io/async_buf_read.rs

@@ -92,7 +92,7 @@ where
     }
 
     fn consume(self: Pin<&mut Self>, amt: usize) {
-        self.get_mut().as_mut().consume(amt)
+        self.get_mut().as_mut().consume(amt);
     }
 }
 
@@ -112,6 +112,6 @@ impl<T: AsRef<[u8]> + Unpin> AsyncBufRead for io::Cursor<T> {
     }
 
     fn consume(self: Pin<&mut Self>, amt: usize) {
-        io::BufRead::consume(self.get_mut(), amt)
+        io::BufRead::consume(self.get_mut(), amt);
     }
 }

tokio/src/io/async_fd.rs

@@ -13,15 +13,15 @@ use std::{task::Context, task::Poll};
 /// `kqueue`, etc), such as a network socket or pipe, and the file descriptor
 /// must have the nonblocking mode set to true.
 ///
-/// Creating an AsyncFd registers the file descriptor with the current tokio
+/// Creating an [`AsyncFd`] registers the file descriptor with the current tokio
 /// Reactor, allowing you to directly await the file descriptor being readable
 /// or writable. Once registered, the file descriptor remains registered until
-/// the AsyncFd is dropped.
+/// the [`AsyncFd`] is dropped.
 ///
-/// The AsyncFd takes ownership of an arbitrary object to represent the IO
+/// The [`AsyncFd`] takes ownership of an arbitrary object to represent the IO
 /// object. It is intended that this object will handle closing the file
 /// descriptor when it is dropped, avoiding resource leaks and ensuring that the
-/// AsyncFd can clean up the registration before closing the file descriptor.
+/// [`AsyncFd`] can clean up the registration before closing the file descriptor.
 /// The [`AsyncFd::into_inner`] function can be used to extract the inner object
 /// to retake control from the tokio IO reactor.
 ///
@@ -204,7 +204,7 @@ pub struct AsyncFdReadyMutGuard<'a, T: AsRawFd> {
 }
 
 impl<T: AsRawFd> AsyncFd<T> {
-    /// Creates an AsyncFd backed by (and taking ownership of) an object
+    /// Creates an [`AsyncFd`] backed by (and taking ownership of) an object
     /// implementing [`AsRawFd`]. The backing file descriptor is cached at the
     /// time of creation.
     ///
@@ -226,7 +226,7 @@ impl<T: AsRawFd> AsyncFd<T> {
         Self::with_interest(inner, Interest::READABLE | Interest::WRITABLE)
     }
 
-    /// Creates an AsyncFd backed by (and taking ownership of) an object
+    /// Creates an [`AsyncFd`] backed by (and taking ownership of) an object
     /// implementing [`AsRawFd`], with a specific [`Interest`]. The backing
     /// file descriptor is cached at the time of creation.
     ///

tokio/src/io/blocking.rs

@@ -116,7 +116,7 @@ where
 
                     self.state = State::Busy(sys::run(move || {
                         let n = buf.len();
-                        let res = buf.write_to(&mut inner).map(|_| n);
+                        let res = buf.write_to(&mut inner).map(|()| n);
 
                         (res, buf, inner)
                     }));
@@ -147,7 +147,7 @@ where
                         let mut inner = self.inner.take().unwrap();
 
                         self.state = State::Busy(sys::run(move || {
-                            let res = inner.flush().map(|_| 0);
+                            let res = inner.flush().map(|()| 0);
                             (res, buf, inner)
                         }));
 

tokio/src/io/interest.rs

@@ -279,7 +279,7 @@ impl ops::BitOr for Interest {
 impl ops::BitOrAssign for Interest {
     #[inline]
     fn bitor_assign(&mut self, other: Self) {
-        *self = *self | other
+        *self = *self | other;
     }
 }
 

tokio/src/io/stdio_common.rs

@@ -3,8 +3,8 @@ use crate::io::AsyncWrite;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 /// # Windows
-/// AsyncWrite adapter that finds last char boundary in given buffer and does not write the rest,
-/// if buffer contents seems to be utf8. Otherwise it only trims buffer down to MAX_BUF.
+/// [`AsyncWrite`] adapter that finds last char boundary in given buffer and does not write the rest,
+/// if buffer contents seems to be utf8. Otherwise it only trims buffer down to `MAX_BUF`.
 /// That's why, wrapped writer will always receive well-formed utf-8 bytes.
 /// # Other platforms
 /// Passes data to `inner` as is.

tokio/src/io/util/async_buf_read_ext.rs

@@ -294,7 +294,7 @@ cfg_io_util! {
         where
             Self: Unpin,
         {
-            std::pin::Pin::new(self).consume(amt)
+            std::pin::Pin::new(self).consume(amt);
         }
 
         /// Returns a stream over the lines of this reader.

tokio/src/io/util/buf_stream.rs

@@ -192,7 +192,7 @@ impl<RW: AsyncRead + AsyncWrite> AsyncBufRead for BufStream<RW> {
     }
 
     fn consume(self: Pin<&mut Self>, amt: usize) {
-        self.project().inner.consume(amt)
+        self.project().inner.consume(amt);
     }
 }
 

tokio/src/io/util/buf_writer.rs

@@ -282,7 +282,7 @@ impl<W: AsyncWrite + AsyncBufRead> AsyncBufRead for BufWriter<W> {
     }
 
     fn consume(self: Pin<&mut Self>, amt: usize) {
-        self.get_pin_mut().consume(amt)
+        self.get_pin_mut().consume(amt);
     }
 }
 

tokio/src/io/util/copy.rs

@@ -40,7 +40,7 @@ impl CopyBuffer {
         buf.set_filled(me.cap);
 
         let res = reader.poll_read(cx, &mut buf);
-        if let Poll::Ready(Ok(_)) = res {
+        if let Poll::Ready(Ok(())) = res {
             let filled_len = buf.filled().len();
             me.read_done = me.cap == filled_len;
             me.cap = filled_len;
@@ -90,7 +90,7 @@ impl CopyBuffer {
                 self.cap = 0;
 
                 match self.poll_fill_buf(cx, reader.as_mut()) {
-                    Poll::Ready(Ok(_)) => (),
+                    Poll::Ready(Ok(())) => (),
                     Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
                     Poll::Pending => {
                         // Try flushing when the reader has no progress to avoid deadlock

tokio/src/io/util/read_to_end.rs

@@ -54,7 +54,7 @@ pub(super) fn read_to_end_internal<V: VecU8, R: AsyncRead + ?Sized>(
     }
 }
 
-/// Tries to read from the provided AsyncRead.
+/// Tries to read from the provided [`AsyncRead`].
 ///
 /// The length of the buffer is increased by the number of bytes read.
 fn poll_read_to_end<V: VecU8, R: AsyncRead + ?Sized>(

tokio/src/io/util/take.rs

@@ -43,7 +43,7 @@ impl<R: AsyncRead> Take<R> {
     /// the amount of bytes read and the previous limit value don't matter when
     /// calling this method.
     pub fn set_limit(&mut self, limit: u64) {
-        self.limit_ = limit
+        self.limit_ = limit;
     }
 
     /// Gets a reference to the underlying reader.

tokio/src/net/tcp/listener.rs

@@ -268,7 +268,7 @@ impl TcpListener {
             use std::os::unix::io::{FromRawFd, IntoRawFd};
             self.io
                 .into_inner()
-                .map(|io| io.into_raw_fd())
+                .map(IntoRawFd::into_raw_fd)
                 .map(|raw_fd| unsafe { std::net::TcpListener::from_raw_fd(raw_fd) })
         }
 

tokio/src/net/tcp/socket.rs

@@ -378,13 +378,13 @@ impl TcpSocket {
         self.inner.recv_buffer_size().map(|n| n as u32)
     }
 
-    /// Sets the linger duration of this socket by setting the SO_LINGER option.
+    /// Sets the linger duration of this socket by setting the `SO_LINGER` option.
     ///
     /// This option controls the action taken when a stream has unsent messages and the stream is
-    /// closed. If SO_LINGER is set, the system shall block the process until it can transmit the
+    /// closed. If `SO_LINGER` is set, the system shall block the process until it can transmit the
     /// data or until the time expires.
     ///
-    /// If SO_LINGER is not specified, and the socket is closed, the system handles the call in a
+    /// If `SO_LINGER` is not specified, and the socket is closed, the system handles the call in a
     /// way that allows the process to continue as quickly as possible.
     pub fn set_linger(&self, dur: Option<Duration>) -> io::Result<()> {
         self.inner.set_linger(dur)

tokio/src/net/tcp/stream.rs

@@ -249,7 +249,7 @@ impl TcpStream {
             use std::os::unix::io::{FromRawFd, IntoRawFd};
             self.io
                 .into_inner()
-                .map(|io| io.into_raw_fd())
+                .map(IntoRawFd::into_raw_fd)
                 .map(|raw_fd| unsafe { std::net::TcpStream::from_raw_fd(raw_fd) })
         }
 

tokio/src/net/udp.rs

@@ -251,7 +251,7 @@ impl UdpSocket {
             use std::os::unix::io::{FromRawFd, IntoRawFd};
             self.io
                 .into_inner()
-                .map(|io| io.into_raw_fd())
+                .map(IntoRawFd::into_raw_fd)
                 .map(|raw_fd| unsafe { std::net::UdpSocket::from_raw_fd(raw_fd) })
         }
 
@@ -342,7 +342,7 @@ impl UdpSocket {
 
         for addr in addrs {
             match self.io.connect(addr) {
-                Ok(_) => return Ok(()),
+                Ok(()) => return Ok(()),
                 Err(e) => last_err = Some(e),
             }
         }
@@ -1506,7 +1506,7 @@ impl UdpSocket {
     /// # Notes
     ///
     /// On Windows, if the data is larger than the buffer specified, the buffer
-    /// is filled with the first part of the data, and peek_from returns the error
+    /// is filled with the first part of the data, and `peek_from` returns the error
     /// WSAEMSGSIZE(10040). The excess data is lost.
     /// Make sure to always use a sufficiently large buffer to hold the
     /// maximum UDP packet size, which can be up to 65536 bytes in size.

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

@@ -423,7 +423,7 @@ impl UnixDatagram {
         Ok((a, b))
     }
 
-    /// Creates new `UnixDatagram` from a `std::os::unix::net::UnixDatagram`.
+    /// Creates new [`UnixDatagram`] from a [`std::os::unix::net::UnixDatagram`].
     ///
     /// This function is intended to be used to wrap a UnixDatagram from the
     /// standard library in the Tokio equivalent.
@@ -498,7 +498,7 @@ impl UnixDatagram {
     pub fn into_std(self) -> io::Result<std::os::unix::net::UnixDatagram> {
         self.io
             .into_inner()
-            .map(|io| io.into_raw_fd())
+            .map(IntoRawFd::into_raw_fd)
             .map(|raw_fd| unsafe { std::os::unix::net::UnixDatagram::from_raw_fd(raw_fd) })
     }
 

tokio/src/net/unix/listener.rs

@@ -70,7 +70,7 @@ impl UnixListener {
         Ok(UnixListener { io })
     }
 
-    /// Creates new `UnixListener` from a `std::os::unix::net::UnixListener `.
+    /// Creates new [`UnixListener`] from a [`std::os::unix::net::UnixListener`].
     ///
     /// This function is intended to be used to wrap a UnixListener from the
     /// standard library in the Tokio equivalent.
@@ -137,7 +137,7 @@ impl UnixListener {
     pub fn into_std(self) -> io::Result<std::os::unix::net::UnixListener> {
         self.io
             .into_inner()
-            .map(|io| io.into_raw_fd())
+            .map(IntoRawFd::into_raw_fd)
             .map(|raw_fd| unsafe { net::UnixListener::from_raw_fd(raw_fd) })
     }
 

tokio/src/net/unix/pipe.rs

@@ -1188,19 +1188,19 @@ fn get_file_flags(file: &File) -> io::Result<libc::c_int> {
     }
 }
 
-/// Checks for O_RDONLY or O_RDWR access mode.
+/// Checks for `O_RDONLY` or `O_RDWR` access mode.
 fn has_read_access(flags: libc::c_int) -> bool {
     let mode = flags & libc::O_ACCMODE;
     mode == libc::O_RDONLY || mode == libc::O_RDWR
 }
 
-/// Checks for O_WRONLY or O_RDWR access mode.
+/// Checks for `O_WRONLY` or `O_RDWR` access mode.
 fn has_write_access(flags: libc::c_int) -> bool {
     let mode = flags & libc::O_ACCMODE;
     mode == libc::O_WRONLY || mode == libc::O_RDWR
 }
 
-/// Sets file's flags with O_NONBLOCK by fcntl.
+/// Sets file's flags with `O_NONBLOCK` by fcntl.
 fn set_nonblocking(file: &mut File, current_flags: libc::c_int) -> io::Result<()> {
     let fd = file.as_raw_fd();
 

tokio/src/net/unix/split.rs

@@ -35,7 +35,7 @@ pub struct ReadHalf<'a>(&'a UnixStream);
 /// Borrowed write half of a [`UnixStream`], created by [`split`].
 ///
 /// Note that in the [`AsyncWrite`] implementation of this type, [`poll_shutdown`] will
-/// shut down the UnixStream stream in the write direction.
+/// shut down the [`UnixStream`] stream in the write direction.
 ///
 /// Writing to an `WriteHalf` is usually done using the convenience methods found
 /// on the [`AsyncWriteExt`] trait.

tokio/src/net/unix/stream.rs

@@ -742,7 +742,7 @@ impl UnixStream {
             .await
     }
 
-    /// Creates new `UnixStream` from a `std::os::unix::net::UnixStream`.
+    /// Creates new [`UnixStream`] from a [`std::os::unix::net::UnixStream`].
     ///
     /// This function is intended to be used to wrap a UnixStream from the
     /// standard library in the Tokio equivalent.
@@ -828,7 +828,7 @@ impl UnixStream {
     pub fn into_std(self) -> io::Result<std::os::unix::net::UnixStream> {
         self.io
             .into_inner()
-            .map(|io| io.into_raw_fd())
+            .map(IntoRawFd::into_raw_fd)
             .map(|raw_fd| unsafe { std::os::unix::net::UnixStream::from_raw_fd(raw_fd) })
     }
 

tokio/src/process/unix/mod.rs

@@ -89,13 +89,13 @@ impl fmt::Debug for GlobalOrphanQueue {
 
 impl GlobalOrphanQueue {
     pub(crate) fn reap_orphans(handle: &SignalHandle) {
-        get_orphan_queue().reap_orphans(handle)
+        get_orphan_queue().reap_orphans(handle);
     }
 }
 
 impl OrphanQueue<StdChild> for GlobalOrphanQueue {
     fn push_orphan(&self, orphan: StdChild) {
-        get_orphan_queue().push_orphan(orphan)
+        get_orphan_queue().push_orphan(orphan);
     }
 }
 

tokio/src/process/unix/orphan.rs

@@ -70,7 +70,7 @@ impl<T> OrphanQueueImpl<T> {
     where
         T: Wait,
     {
-        self.queue.lock().push(orphan)
+        self.queue.lock().push(orphan);
     }
 
     /// Attempts to reap every process in the queue, ignoring any errors and

tokio/src/runtime/blocking/pool.rs

@@ -228,7 +228,7 @@ impl BlockingPool {
                     before_stop: builder.before_stop.clone(),
                     thread_cap,
                     keep_alive,
-                    metrics: Default::default(),
+                    metrics: SpawnerMetrics::default(),
                 }),
             },
             shutdown_rx,
@@ -259,14 +259,14 @@ impl BlockingPool {
         drop(shared);
 
         if self.shutdown_rx.wait(timeout) {
-            let _ = last_exited_thread.map(|th| th.join());
+            let _ = last_exited_thread.map(thread::JoinHandle::join);
 
             // Loom requires that execution be deterministic, so sort by thread ID before joining.
             // (HashMaps use a randomly-seeded hash function, so the order is nondeterministic)
             let mut workers: Vec<(usize, thread::JoinHandle<()>)> = workers.into_iter().collect();
             workers.sort_by_key(|(id, _)| *id);
 
-            for (_id, handle) in workers.into_iter() {
+            for (_id, handle) in workers {
                 let _ = handle.join();
             }
         }
@@ -499,7 +499,7 @@ fn is_temporary_os_thread_error(error: &std::io::Error) -> bool {
 impl Inner {
     fn run(&self, worker_thread_id: usize) {
         if let Some(f) = &self.after_start {
-            f()
+            f();
         }
 
         let mut shared = self.shared.lock();
@@ -575,9 +575,10 @@ impl Inner {
         // with a descriptive message if it is not the
         // case.
         let prev_idle = self.metrics.dec_num_idle_threads();
-        if prev_idle < self.metrics.num_idle_threads() {
-            panic!("num_idle_threads underflowed on thread exit")
-        }
+        assert!(
+            prev_idle >= self.metrics.num_idle_threads(),
+            "num_idle_threads underflowed on thread exit"
+        );
 
         if shared.shutdown && self.metrics.num_threads() == 0 {
             self.condvar.notify_one();
@@ -586,7 +587,7 @@ impl Inner {
         drop(shared);
 
         if let Some(f) = &self.before_stop {
-            f()
+            f();
         }
 
         if let Some(handle) = join_on_thread {

tokio/src/runtime/builder.rs

@@ -312,7 +312,7 @@ impl Builder {
 
             metrics_poll_count_histogram_enable: false,
 
-            metrics_poll_count_histogram: Default::default(),
+            metrics_poll_count_histogram: HistogramBuilder::default(),
 
             disable_lifo_slot: false,
         }

tokio/src/runtime/context/blocking.rs

@@ -115,7 +115,7 @@ impl Drop for DisallowBlockInPlaceGuard {
                         allow_block_in_place: true,
                     });
                 }
-            })
+            });
         }
     }
 }

tokio/src/runtime/context/current.rs

@@ -50,9 +50,7 @@ impl Context {
         let old_handle = self.current.handle.borrow_mut().replace(handle.clone());
         let depth = self.current.depth.get();
 
-        if depth == usize::MAX {
-            panic!("reached max `enter` depth");
-        }
+        assert!(depth != usize::MAX, "reached max `enter` depth");
 
         let depth = depth + 1;
         self.current.depth.set(depth);

tokio/src/runtime/coop.rs

@@ -62,7 +62,7 @@ impl Budget {
     }
 
     fn has_remaining(self) -> bool {
-        self.0.map(|budget| budget > 0).unwrap_or(true)
+        self.0.map_or(true, |budget| budget > 0)
     }
 }
 

tokio/src/runtime/driver.rs

@@ -66,15 +66,15 @@ impl Driver {
     }
 
     pub(crate) fn park(&mut self, handle: &Handle) {
-        self.inner.park(handle)
+        self.inner.park(handle);
     }
 
     pub(crate) fn park_timeout(&mut self, handle: &Handle, duration: Duration) {
-        self.inner.park_timeout(handle, duration)
+        self.inner.park_timeout(handle, duration);
     }
 
     pub(crate) fn shutdown(&mut self, handle: &Handle) {
-        self.inner.shutdown(handle)
+        self.inner.shutdown(handle);
     }
 }
 

tokio/src/runtime/handle.rs

@@ -228,7 +228,7 @@ impl Handle {
     /// When this is used on a `current_thread` runtime, only the
     /// [`Runtime::block_on`] method can drive the IO and timer drivers, but the
     /// `Handle::block_on` method cannot drive them. This means that, when using
-    /// this method on a current_thread runtime, anything that relies on IO or
+    /// this method on a `current_thread` runtime, anything that relies on IO or
     /// timers will not work unless there is another thread currently calling
     /// [`Runtime::block_on`] on the same runtime.
     ///

tokio/src/runtime/io/driver.rs

@@ -154,7 +154,7 @@ impl Driver {
         // Block waiting for an event to happen, peeling out how many events
         // happened.
         match self.poll.poll(events, max_wait) {
-            Ok(_) => {}
+            Ok(()) => {}
             Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
             #[cfg(target_os = "wasi")]
             Err(e) if e.kind() == io::ErrorKind::InvalidInput => {

tokio/src/runtime/io/scheduled_io.rs

@@ -180,7 +180,7 @@ impl Default for ScheduledIo {
         ScheduledIo {
             linked_list_pointers: UnsafeCell::new(linked_list::Pointers::new()),
             readiness: AtomicUsize::new(0),
-            waiters: Mutex::new(Default::default()),
+            waiters: Mutex::new(Waiters::default()),
         }
     }
 }

tokio/src/runtime/park.rs

@@ -197,7 +197,7 @@ impl Inner {
         // to release `lock`.
         drop(self.mutex.lock());
 
-        self.condvar.notify_one()
+        self.condvar.notify_one();
     }
 
     fn shutdown(&self) {
@@ -243,11 +243,11 @@ impl CachedParkThread {
     }
 
     pub(crate) fn waker(&self) -> Result<Waker, AccessError> {
-        self.unpark().map(|unpark| unpark.into_waker())
+        self.unpark().map(UnparkThread::into_waker)
     }
 
     fn unpark(&self) -> Result<UnparkThread, AccessError> {
-        self.with_current(|park_thread| park_thread.unpark())
+        self.with_current(ParkThread::unpark)
     }
 
     pub(crate) fn park(&mut self) {

tokio/src/runtime/process.rs

@@ -39,6 +39,6 @@ impl Driver {
     }
 
     pub(crate) fn shutdown(&mut self, handle: &driver::Handle) {
-        self.park.shutdown(handle)
+        self.park.shutdown(handle);
     }
 }

tokio/src/runtime/runtime.rs

@@ -450,7 +450,7 @@ impl Runtime {
     /// }
     /// ```
     pub fn shutdown_background(self) {
-        self.shutdown_timeout(Duration::from_nanos(0))
+        self.shutdown_timeout(Duration::from_nanos(0));
     }
 }
 

tokio/src/runtime/scheduler/current_thread/mod.rs

@@ -354,7 +354,7 @@ impl Context {
             // Incorrect lint, the closures are actually different types so `f`
             // cannot be passed as an argument to `enter`.
             #[allow(clippy::redundant_closure)]
-            let (c, _) = self.enter(core, || f());
+            let (c, ()) = self.enter(core, || f());
             core = c;
         }
 
@@ -365,7 +365,7 @@ impl Context {
             core.metrics.about_to_park();
             core.submit_metrics(handle);
 
-            let (c, _) = self.enter(core, || {
+            let (c, ()) = self.enter(core, || {
                 driver.park(&handle.driver);
                 self.defer.wake();
             });
@@ -377,7 +377,7 @@ impl Context {
             // Incorrect lint, the closures are actually different types so `f`
             // cannot be passed as an argument to `enter`.
             #[allow(clippy::redundant_closure)]
-            let (c, _) = self.enter(core, || f());
+            let (c, ()) = self.enter(core, || f());
             core = c;
         }
 
@@ -391,7 +391,7 @@ impl Context {
 
         core.submit_metrics(handle);
 
-        let (mut core, _) = self.enter(core, || {
+        let (mut core, ()) = self.enter(core, || {
             driver.park_timeout(&handle.driver, Duration::from_millis(0));
             self.defer.wake();
         });
@@ -627,7 +627,7 @@ impl Schedule for Arc<Handle> {
 
 impl Wake for Handle {
     fn wake(arc_self: Arc<Self>) {
-        Wake::wake_by_ref(&arc_self)
+        Wake::wake_by_ref(&arc_self);
     }
 
     /// Wake by reference
@@ -702,7 +702,7 @@ impl CoreGuard<'_> {
 
                     let task = context.handle.shared.owned.assert_owner(task);
 
-                    let (c, _) = context.run_task(core, || {
+                    let (c, ()) = context.run_task(core, || {
                         task.run();
                     });
 
@@ -758,7 +758,7 @@ impl Drop for CoreGuard<'_> {
             self.scheduler.core.set(core);
 
             // Wake up other possible threads that could steal the driver.
-            self.scheduler.notify.notify_one()
+            self.scheduler.notify.notify_one();
         }
     }
 }

tokio/src/runtime/scheduler/defer.rs

@@ -8,7 +8,7 @@ pub(crate) struct Defer {
 impl Defer {
     pub(crate) fn new() -> Defer {
         Defer {
-            deferred: Default::default(),
+            deferred: RefCell::default(),
         }
     }
 

tokio/src/runtime/scheduler/mod.rs

@@ -222,7 +222,7 @@ cfg_rt! {
         }
 
         pub(crate) fn defer(&self, waker: &Waker) {
-            match_flavor!(self, Context(context) => context.defer(waker))
+            match_flavor!(self, Context(context) => context.defer(waker));
         }
 
         cfg_rt_multi_thread! {

tokio/src/runtime/scheduler/multi_thread/park.rs

@@ -72,7 +72,7 @@ impl Parker {
         assert_eq!(duration, Duration::from_millis(0));
 
         if let Some(mut driver) = self.inner.shared.driver.try_lock() {
-            driver.park_timeout(handle, duration)
+            driver.park_timeout(handle, duration);
         }
     }
 
@@ -219,7 +219,7 @@ impl Inner {
         // to release `lock`.
         drop(self.mutex.lock());
 
-        self.condvar.notify_one()
+        self.condvar.notify_one();
     }
 
     fn shutdown(&self, handle: &driver::Handle) {

tokio/src/runtime/scheduler/multi_thread/queue.rs

@@ -121,7 +121,7 @@ impl<T> Local<T> {
 
     /// Returns false if there are any entries in the queue
     ///
-    /// Separate to is_stealable so that refactors of is_stealable to "protect"
+    /// Separate to `is_stealable` so that refactors of `is_stealable` to "protect"
     /// some tasks from stealing won't affect this
     pub(crate) fn has_tasks(&self) -> bool {
         !self.inner.is_empty()

tokio/src/runtime/scheduler/multi_thread/worker.rs

@@ -1025,7 +1025,7 @@ impl Handle {
             // Otherwise, use the inject queue.
             self.push_remote_task(task);
             self.notify_parked_remote();
-        })
+        });
     }
 
     pub(super) fn schedule_option_task_without_yield(&self, task: Option<Notified>) {

tokio/src/runtime/signal/mod.rs

@@ -99,7 +99,7 @@ impl Driver {
     }
 
     pub(crate) fn shutdown(&mut self, handle: &driver::Handle) {
-        self.io.shutdown(handle)
+        self.io.shutdown(handle);
     }
 
     fn process(&mut self) {

tokio/src/runtime/task/core.rs

@@ -379,7 +379,7 @@ impl<T: Future, S: Schedule> Core<T, S> {
 
     unsafe fn set_stage(&self, stage: Stage<T>) {
         let _guard = TaskIdGuard::enter(self.task_id);
-        self.stage.stage.with_mut(|ptr| *ptr = stage)
+        self.stage.stage.with_mut(|ptr| *ptr = stage);
     }
 }
 

tokio/src/runtime/task/mod.rs

@@ -448,7 +448,7 @@ impl<S: Schedule> UnownedTask<S> {
     }
 
     pub(crate) fn shutdown(self) {
-        self.into_task().shutdown()
+        self.into_task().shutdown();
     }
 }
 

tokio/src/runtime/task/raw.rs

@@ -6,6 +6,7 @@ use std::ptr::NonNull;
 use std::task::{Poll, Waker};
 
 /// Raw task handle
+#[derive(Clone)]
 pub(crate) struct RawTask {
     ptr: NonNull<Header>,
 }
@@ -162,7 +163,7 @@ impl RawTask {
         S: Schedule,
     {
         let ptr = Box::into_raw(Cell::<_, S>::new(task, scheduler, State::new(), id));
-        let ptr = unsafe { NonNull::new_unchecked(ptr as *mut Header) };
+        let ptr = unsafe { NonNull::new_unchecked(ptr.cast()) };
 
         RawTask { ptr }
     }
@@ -263,12 +264,6 @@ impl RawTask {
     }
 }
 
-impl Clone for RawTask {
-    fn clone(&self) -> Self {
-        RawTask { ptr: self.ptr }
-    }
-}
-
 impl Copy for RawTask {}
 
 unsafe fn poll<T: Future, S: Schedule>(ptr: NonNull<Header>) {
@@ -303,7 +298,7 @@ unsafe fn try_read_output<T: Future, S: Schedule>(
 
 unsafe fn drop_join_handle_slow<T: Future, S: Schedule>(ptr: NonNull<Header>) {
     let harness = Harness::<T, S>::from_raw(ptr);
-    harness.drop_join_handle_slow()
+    harness.drop_join_handle_slow();
 }
 
 unsafe fn drop_abort_handle<T: Future, S: Schedule>(ptr: NonNull<Header>) {
@@ -313,5 +308,5 @@ unsafe fn drop_abort_handle<T: Future, S: Schedule>(ptr: NonNull<Header>) {
 
 unsafe fn shutdown<T: Future, S: Schedule>(ptr: NonNull<Header>) {
     let harness = Harness::<T, S>::from_raw(ptr);
-    harness.shutdown()
+    harness.shutdown();
 }

tokio/src/runtime/task/state.rs

@@ -368,7 +368,7 @@ impl State {
             .map_err(|_| ())
     }
 
-    /// Tries to unset the JOIN_INTEREST flag.
+    /// Tries to unset the `JOIN_INTEREST` flag.
     ///
     /// Returns `Ok` if the operation happens before the task transitions to a
     /// completed state, `Err` otherwise.
@@ -522,11 +522,11 @@ impl Snapshot {
     }
 
     fn unset_notified(&mut self) {
-        self.0 &= !NOTIFIED
+        self.0 &= !NOTIFIED;
     }
 
     fn set_notified(&mut self) {
-        self.0 |= NOTIFIED
+        self.0 |= NOTIFIED;
     }
 
     pub(super) fn is_running(self) -> bool {
@@ -559,7 +559,7 @@ impl Snapshot {
     }
 
     fn unset_join_interested(&mut self) {
-        self.0 &= !JOIN_INTEREST
+        self.0 &= !JOIN_INTEREST;
     }
 
     pub(super) fn is_join_waker_set(self) -> bool {
@@ -571,7 +571,7 @@ impl Snapshot {
     }
 
     fn unset_join_waker(&mut self) {
-        self.0 &= !JOIN_WAKER
+        self.0 &= !JOIN_WAKER;
     }
 
     pub(super) fn ref_count(self) -> usize {
@@ -585,7 +585,7 @@ impl Snapshot {
 
     pub(super) fn ref_dec(&mut self) {
         assert!(self.ref_count() > 0);
-        self.0 -= REF_ONE
+        self.0 -= REF_ONE;
     }
 }
 

tokio/src/runtime/time/entry.rs

@@ -206,7 +206,7 @@ impl StateCell {
     /// Fires the timer, setting the result to the provided result.
     ///
     /// Returns:
-    /// * `Some(waker) - if fired and a waker needs to be invoked once the
+    /// * `Some(waker)` - if fired and a waker needs to be invoked once the
     ///   driver lock is released
     /// * `None` - if fired and a waker does not need to be invoked, or if
     ///   already fired
@@ -553,9 +553,11 @@ impl TimerEntry {
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), super::Error>> {
-        if self.driver().is_shutdown() {
-            panic!("{}", crate::util::error::RUNTIME_SHUTTING_DOWN_ERROR);
-        }
+        assert!(
+            !self.driver().is_shutdown(),
+            "{}",
+            crate::util::error::RUNTIME_SHUTTING_DOWN_ERROR
+        );
 
         if !self.registered {
             let deadline = self.deadline;
@@ -639,6 +641,6 @@ impl TimerHandle {
 
 impl Drop for TimerEntry {
     fn drop(&mut self) {
-        unsafe { Pin::new_unchecked(self) }.as_mut().cancel()
+        unsafe { Pin::new_unchecked(self) }.as_mut().cancel();
     }
 }

tokio/src/runtime/time/mod.rs

@@ -149,11 +149,11 @@ impl Driver {
     }
 
     pub(crate) fn park(&mut self, handle: &driver::Handle) {
-        self.park_internal(handle, None)
+        self.park_internal(handle, None);
     }
 
     pub(crate) fn park_timeout(&mut self, handle: &driver::Handle, duration: Duration) {
-        self.park_internal(handle, Some(duration))
+        self.park_internal(handle, Some(duration));
     }
 
     pub(crate) fn shutdown(&mut self, rt_handle: &driver::Handle) {
@@ -253,7 +253,7 @@ impl Handle {
     pub(self) fn process(&self, clock: &Clock) {
         let now = self.time_source().now(clock);
 
-        self.process_at_time(now)
+        self.process_at_time(now);
     }
 
     pub(self) fn process_at_time(&self, mut now: u64) {
@@ -305,7 +305,7 @@ impl Handle {
 
         drop(lock);
 
-        for waker in waker_list[0..waker_idx].iter_mut() {
+        for waker in &mut waker_list[0..waker_idx] {
             waker.take().unwrap().wake();
         }
     }

tokio/src/signal/mod.rs

@@ -70,10 +70,8 @@ struct RxFuture {
 }
 
 async fn make_future(mut rx: Receiver<()>) -> Receiver<()> {
-    match rx.changed().await {
-        Ok(()) => rx,
-        Err(_) => panic!("signal sender went away"),
-    }
+    rx.changed().await.expect("signal sender went away");
+    rx
 }
 
 impl RxFuture {

tokio/src/signal/registry.rs

@@ -48,7 +48,7 @@ impl Storage for Vec<EventInfo> {
     where
         F: FnMut(&'a EventInfo),
     {
-        self.iter().for_each(f)
+        self.iter().for_each(f);
     }
 }
 
@@ -87,7 +87,7 @@ impl<S: Storage> Registry<S> {
     /// any listeners.
     fn record_event(&self, event_id: EventId) {
         if let Some(event_info) = self.storage.event_info(event_id) {
-            event_info.pending.store(true, Ordering::SeqCst)
+            event_info.pending.store(true, Ordering::SeqCst);
         }
     }
 

tokio/src/signal/unix.rs

@@ -43,7 +43,7 @@ impl Storage for OsStorage {
     where
         F: FnMut(&'a EventInfo),
     {
-        self.iter().map(|si| &si.event_info).for_each(f)
+        self.iter().map(|si| &si.event_info).for_each(f);
     }
 }
 
@@ -224,7 +224,7 @@ pub(crate) struct SignalInfo {
 impl Default for SignalInfo {
     fn default() -> SignalInfo {
         SignalInfo {
-            event_info: Default::default(),
+            event_info: EventInfo::default(),
             init: Once::new(),
             initialized: AtomicBool::new(false),
         }

tokio/src/sync/batch_semaphore.rs

@@ -474,8 +474,7 @@ impl Semaphore {
             // Do we need to register the new waker?
             if waker
                 .as_ref()
-                .map(|waker| !waker.will_wake(cx.waker()))
-                .unwrap_or(true)
+                .map_or(true, |waker| !waker.will_wake(cx.waker()))
             {
                 old_waker = std::mem::replace(waker, Some(cx.waker().clone()));
             }

tokio/src/sync/broadcast.rs

@@ -299,7 +299,7 @@ pub mod error {
     impl std::error::Error for TryRecvError {}
 }
 
-use self::error::*;
+use self::error::{RecvError, SendError, TryRecvError};
 
 /// Data shared between senders and receivers.
 struct Shared<T> {
@@ -817,9 +817,7 @@ impl<T> Sender<T> {
 fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
     let mut tail = shared.tail.lock();
 
-    if tail.rx_cnt == MAX_RECEIVERS {
-        panic!("max receivers");
-    }
+    assert!(tail.rx_cnt != MAX_RECEIVERS, "max receivers");
 
     tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");
 

tokio/src/sync/mpsc/block.rs

@@ -272,10 +272,9 @@ impl<T> Block<T> {
         let ret = NonNull::new(self.header.next.load(ordering));
 
         debug_assert!(unsafe {
-            ret.map(|block| {
+            ret.map_or(true, |block| {
                 block.as_ref().header.start_index == self.header.start_index.wrapping_add(BLOCK_CAP)
             })
-            .unwrap_or(true)
         });
 
         ret
@@ -290,7 +289,7 @@ impl<T> Block<T> {
     ///
     /// # Ordering
     ///
-    /// This performs a compare-and-swap on `next` using AcqRel ordering.
+    /// This performs a compare-and-swap on `next` using `AcqRel` ordering.
     ///
     /// # Safety
     ///
@@ -326,7 +325,7 @@ impl<T> Block<T> {
     ///
     /// It is assumed that `self.next` is null. A new block is allocated with
     /// `start_index` set to be the next block. A compare-and-swap is performed
-    /// with AcqRel memory ordering. If the compare-and-swap is successful, the
+    /// with `AcqRel` memory ordering. If the compare-and-swap is successful, the
     /// newly allocated block is released to other threads walking the block
     /// linked list. If the compare-and-swap fails, the current thread acquires
     /// the next block in the linked list, allowing the current thread to access
@@ -382,7 +381,7 @@ impl<T> Block<T> {
             let actual = unsafe { curr.as_ref().try_push(&mut new_block, AcqRel, Acquire) };
 
             curr = match actual {
-                Ok(_) => {
+                Ok(()) => {
                     return next;
                 }
                 Err(curr) => curr,

tokio/src/sync/mpsc/bounded.rs

@@ -35,7 +35,7 @@ pub struct Sender<T> {
 /// [`Sender`]: Sender
 /// [`WeakSender::upgrade`]: WeakSender::upgrade
 ///
-/// #Examples
+/// # Examples
 ///
 /// ```
 /// use tokio::sync::mpsc::channel;
@@ -522,7 +522,7 @@ impl<T> Sender<T> {
     /// }
     /// ```
     pub async fn closed(&self) {
-        self.chan.closed().await
+        self.chan.closed().await;
     }
 
     /// Attempts to immediately send a message on this `Sender`
@@ -585,7 +585,7 @@ impl<T> Sender<T> {
     /// ```
     pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
         match self.chan.semaphore().semaphore.try_acquire(1) {
-            Ok(_) => {}
+            Ok(()) => {}
             Err(TryAcquireError::Closed) => return Err(TrySendError::Closed(message)),
             Err(TryAcquireError::NoPermits) => return Err(TrySendError::Full(message)),
         }
@@ -868,7 +868,7 @@ impl<T> Sender<T> {
         crate::trace::async_trace_leaf().await;
 
         match self.chan.semaphore().semaphore.acquire(1).await {
-            Ok(_) => Ok(()),
+            Ok(()) => Ok(()),
             Err(_) => Err(SendError(())),
         }
     }
@@ -918,7 +918,7 @@ impl<T> Sender<T> {
     /// ```
     pub fn try_reserve(&self) -> Result<Permit<'_, T>, TrySendError<()>> {
         match self.chan.semaphore().semaphore.try_acquire(1) {
-            Ok(_) => {}
+            Ok(()) => {}
             Err(TryAcquireError::Closed) => return Err(TrySendError::Closed(())),
             Err(TryAcquireError::NoPermits) => return Err(TrySendError::Full(())),
         }
@@ -983,7 +983,7 @@ impl<T> Sender<T> {
     /// ```
     pub fn try_reserve_owned(self) -> Result<OwnedPermit<T>, TrySendError<Self>> {
         match self.chan.semaphore().semaphore.try_acquire(1) {
-            Ok(_) => {}
+            Ok(()) => {}
             Err(TryAcquireError::Closed) => return Err(TrySendError::Closed(self)),
             Err(TryAcquireError::NoPermits) => return Err(TrySendError::Full(self)),
         }
@@ -1118,7 +1118,7 @@ impl<T> Clone for WeakSender<T> {
 }
 
 impl<T> WeakSender<T> {
-    /// Tries to convert a WeakSender into a [`Sender`]. This will return `Some`
+    /// Tries to convert a `WeakSender` into a [`Sender`]. This will return `Some`
     /// if there are other `Sender` instances alive and the channel wasn't
     /// previously dropped, otherwise `None` is returned.
     pub fn upgrade(&self) -> Option<Sender<T>> {

tokio/src/sync/mpsc/chan.rs

@@ -351,7 +351,7 @@ impl<T, S: Semaphore> Drop for Rx<T, S> {
             while let Some(Value(_)) = rx_fields.list.pop(&self.inner.tx) {
                 self.inner.semaphore.add_permit();
             }
-        })
+        });
     }
 }
 
@@ -386,7 +386,7 @@ impl<T, S> Drop for Chan<T, S> {
 
 impl Semaphore for bounded::Semaphore {
     fn add_permit(&self) {
-        self.semaphore.release(1)
+        self.semaphore.release(1);
     }
 
     fn is_idle(&self) -> bool {

tokio/src/sync/mpsc/list.rs

@@ -82,7 +82,7 @@ impl<T> Tx<T> {
     /// Closes the send half of the list.
     ///
     /// Similar process as pushing a value, but instead of writing the value &
-    /// setting the ready flag, the TX_CLOSED flag is set on the block.
+    /// setting the ready flag, the `TX_CLOSED` flag is set on the block.
     pub(crate) fn close(&self) {
         // First, claim a slot for the value. This is the last slot that will be
         // claimed.
@@ -204,7 +204,7 @@ impl<T> Tx<T> {
         // TODO: Unify this logic with Block::grow
         for _ in 0..3 {
             match curr.as_ref().try_push(&mut block, AcqRel, Acquire) {
-                Ok(_) => {
+                Ok(()) => {
                     reused = true;
                     break;
                 }

tokio/src/sync/mpsc/unbounded.rs

@@ -25,7 +25,7 @@ pub struct UnboundedSender<T> {
 /// [`UnboundedSender`]: UnboundedSender
 /// [`WeakUnboundedSender::upgrade`]: WeakUnboundedSender::upgrade
 ///
-/// #Examples
+/// # Examples
 ///
 /// ```
 /// use tokio::sync::mpsc::unbounded_channel;
@@ -379,7 +379,7 @@ impl<T> UnboundedSender<T> {
     /// }
     /// ```
     pub async fn closed(&self) {
-        self.chan.closed().await
+        self.chan.closed().await;
     }
 
     /// Checks if the channel has been closed. This happens when the
@@ -440,7 +440,7 @@ impl<T> Clone for WeakUnboundedSender<T> {
 }
 
 impl<T> WeakUnboundedSender<T> {
-    /// Tries to convert a WeakUnboundedSender into an [`UnboundedSender`].
+    /// Tries to convert a `WeakUnboundedSender` into an [`UnboundedSender`].
     /// This will return `Some` if there are other `Sender` instances alive and
     /// the channel wasn't previously dropped, otherwise `None` is returned.
     pub fn upgrade(&self) -> Option<UnboundedSender<T>> {

tokio/src/sync/mutex.rs

@@ -664,7 +664,7 @@ impl<T: ?Sized> Mutex<T> {
     /// ```
     pub fn try_lock(&self) -> Result<MutexGuard<'_, T>, TryLockError> {
         match self.s.try_acquire(1) {
-            Ok(_) => {
+            Ok(()) => {
                 let guard = MutexGuard {
                     lock: self,
                     #[cfg(all(tokio_unstable, feature = "tracing"))]
@@ -735,7 +735,7 @@ impl<T: ?Sized> Mutex<T> {
     /// # }
     pub fn try_lock_owned(self: Arc<Self>) -> Result<OwnedMutexGuard<T>, TryLockError> {
         match self.s.try_acquire(1) {
-            Ok(_) => {
+            Ok(()) => {
                 let guard = OwnedMutexGuard {
                     #[cfg(all(tokio_unstable, feature = "tracing"))]
                     resource_span: self.resource_span.clone(),

tokio/src/sync/notify.rs

@@ -709,50 +709,47 @@ impl UnwindSafe for Notify {}
 impl RefUnwindSafe for Notify {}
 
 fn notify_locked(waiters: &mut WaitList, state: &AtomicUsize, curr: usize) -> Option<Waker> {
-    loop {
-        match get_state(curr) {
-            EMPTY | NOTIFIED => {
-                let res = state.compare_exchange(curr, set_state(curr, NOTIFIED), SeqCst, SeqCst);
+    match get_state(curr) {
+        EMPTY | NOTIFIED => {
+            let res = state.compare_exchange(curr, set_state(curr, NOTIFIED), SeqCst, SeqCst);
 
-                match res {
-                    Ok(_) => return None,
-                    Err(actual) => {
-                        let actual_state = get_state(actual);
-                        assert!(actual_state == EMPTY || actual_state == NOTIFIED);
-                        state.store(set_state(actual, NOTIFIED), SeqCst);
-                        return None;
-                    }
+            match res {
+                Ok(_) => None,
+                Err(actual) => {
+                    let actual_state = get_state(actual);
+                    assert!(actual_state == EMPTY || actual_state == NOTIFIED);
+                    state.store(set_state(actual, NOTIFIED), SeqCst);
+                    None
                 }
             }
-            WAITING => {
-                // At this point, it is guaranteed that the state will not
-                // concurrently change as holding the lock is required to
-                // transition **out** of `WAITING`.
-                //
-                // Get a pending waiter
-                let waiter = waiters.pop_back().unwrap();
-
-                // Safety: we never make mutable references to waiters.
-                let waiter = unsafe { waiter.as_ref() };
-
-                // Safety: we hold the lock, so we can access the waker.
-                let waker = unsafe { waiter.waker.with_mut(|waker| (*waker).take()) };
-
-                // This waiter is unlinked and will not be shared ever again, release it.
-                waiter.notification.store_release(Notification::One);
-
-                if waiters.is_empty() {
-                    // As this the **final** waiter in the list, the state
-                    // must be transitioned to `EMPTY`. As transitioning
-                    // **from** `WAITING` requires the lock to be held, a
-                    // `store` is sufficient.
-                    state.store(set_state(curr, EMPTY), SeqCst);
-                }
-
-                return waker;
-            }
-            _ => unreachable!(),
         }
+        WAITING => {
+            // At this point, it is guaranteed that the state will not
+            // concurrently change as holding the lock is required to
+            // transition **out** of `WAITING`.
+            //
+            // Get a pending waiter
+            let waiter = waiters.pop_back().unwrap();
+
+            // Safety: we never make mutable references to waiters.
+            let waiter = unsafe { waiter.as_ref() };
+
+            // Safety: we hold the lock, so we can access the waker.
+            let waker = unsafe { waiter.waker.with_mut(|waker| (*waker).take()) };
+
+            // This waiter is unlinked and will not be shared ever again, release it.
+            waiter.notification.store_release(Notification::One);
+
+            if waiters.is_empty() {
+                // As this the **final** waiter in the list, the state
+                // must be transitioned to `EMPTY`. As transitioning
+                // **from** `WAITING` requires the lock to be held, a
+                // `store` is sufficient.
+                state.store(set_state(curr, EMPTY), SeqCst);
+            }
+            waker
+        }
+        _ => unreachable!(),
     }
 }
 

tokio/src/sync/oneshot.rs

@@ -712,7 +712,7 @@ impl<T> Sender<T> {
         #[cfg(not(all(tokio_unstable, feature = "tracing")))]
         let closed = poll_fn(|cx| self.poll_closed(cx));
 
-        closed.await
+        closed.await;
     }
 
     /// Returns `true` if the associated [`Receiver`] handle has been dropped.

tokio/src/sync/semaphore.rs

@@ -611,7 +611,7 @@ impl Semaphore {
     /// [`SemaphorePermit`]: crate::sync::SemaphorePermit
     pub fn try_acquire(&self) -> Result<SemaphorePermit<'_>, TryAcquireError> {
         match self.ll_sem.try_acquire(1) {
-            Ok(_) => Ok(SemaphorePermit {
+            Ok(()) => Ok(SemaphorePermit {
                 sem: self,
                 permits: 1,
             }),
@@ -646,7 +646,7 @@ impl Semaphore {
     /// [`SemaphorePermit`]: crate::sync::SemaphorePermit
     pub fn try_acquire_many(&self, n: u32) -> Result<SemaphorePermit<'_>, TryAcquireError> {
         match self.ll_sem.try_acquire(n) {
-            Ok(_) => Ok(SemaphorePermit {
+            Ok(()) => Ok(SemaphorePermit {
                 sem: self,
                 permits: n,
             }),
@@ -813,7 +813,7 @@ impl Semaphore {
     /// [`OwnedSemaphorePermit`]: crate::sync::OwnedSemaphorePermit
     pub fn try_acquire_owned(self: Arc<Self>) -> Result<OwnedSemaphorePermit, TryAcquireError> {
         match self.ll_sem.try_acquire(1) {
-            Ok(_) => Ok(OwnedSemaphorePermit {
+            Ok(()) => Ok(OwnedSemaphorePermit {
                 sem: self,
                 permits: 1,
             }),
@@ -855,7 +855,7 @@ impl Semaphore {
         n: u32,
     ) -> Result<OwnedSemaphorePermit, TryAcquireError> {
         match self.ll_sem.try_acquire(n) {
-            Ok(_) => Ok(OwnedSemaphorePermit {
+            Ok(()) => Ok(OwnedSemaphorePermit {
                 sem: self,
                 permits: n,
             }),

tokio/src/sync/task/atomic_waker.rs

@@ -363,7 +363,7 @@ trait WakerRef {
 
 impl WakerRef for Waker {
     fn wake(self) {
-        self.wake()
+        self.wake();
     }
 
     fn into_waker(self) -> Waker {
@@ -373,7 +373,7 @@ impl WakerRef for Waker {
 
 impl WakerRef for &Waker {
     fn wake(self) {
-        self.wake_by_ref()
+        self.wake_by_ref();
     }
 
     fn into_waker(self) -> Waker {

tokio/src/sync/watch.rs

@@ -299,7 +299,7 @@ pub mod error {
 }
 
 mod big_notify {
-    use super::*;
+    use super::Notify;
     use crate::sync::notify::Notified;
 
     // To avoid contention on the lock inside the `Notify`, we store multiple
@@ -315,7 +315,7 @@ mod big_notify {
 
     pub(super) struct BigNotify {
         #[cfg(not(all(not(loom), feature = "sync", any(feature = "rt", feature = "macros"))))]
-        next: AtomicUsize,
+        next: std::sync::atomic::AtomicUsize,
         inner: [Notify; 8],
     }
 
@@ -327,7 +327,7 @@ mod big_notify {
                     feature = "sync",
                     any(feature = "rt", feature = "macros")
                 )))]
-                next: AtomicUsize::new(0),
+                next: std::sync::atomic::AtomicUsize::new(0),
                 inner: Default::default(),
             }
         }
@@ -341,7 +341,7 @@ mod big_notify {
         /// This function implements the case where randomness is not available.
         #[cfg(not(all(not(loom), feature = "sync", any(feature = "rt", feature = "macros"))))]
         pub(super) fn notified(&self) -> Notified<'_> {
-            let i = self.next.fetch_add(1, Relaxed) % 8;
+            let i = self.next.fetch_add(1, std::sync::atomic::Ordering::Relaxed) % 8;
             self.inner[i].notified()
         }
 

tokio/src/task/local.rs

@@ -412,7 +412,7 @@ impl Drop for LocalEnterGuard {
                 ctx.set(self.ctx.take());
                 wake_on_schedule.set(self.wake_on_schedule);
             },
-        )
+        );
     }
 }
 
@@ -658,9 +658,7 @@ impl LocalSet {
     fn tick(&self) -> bool {
         for _ in 0..MAX_TASKS_PER_TICK {
             // Make sure we didn't hit an unhandled panic
-            if self.context.unhandled_panic.get() {
-                panic!("a spawned task panicked and the LocalSet is configured to shutdown on unhandled panic");
-            }
+            assert!(!self.context.unhandled_panic.get(), "a spawned task panicked and the LocalSet is configured to shutdown on unhandled panic");
 
             match self.next_task() {
                 // Run the task
@@ -701,7 +699,7 @@ impl LocalSet {
                     .queue
                     .lock()
                     .as_mut()
-                    .and_then(|queue| queue.pop_front())
+                    .and_then(VecDeque::pop_front)
             })
         };
 
@@ -1092,7 +1090,7 @@ impl LocalState {
         // the LocalSet.
         self.assert_called_from_owner_thread();
 
-        self.local_queue.with_mut(|ptr| (*ptr).push_back(task))
+        self.local_queue.with_mut(|ptr| (*ptr).push_back(task));
     }
 
     unsafe fn take_local_queue(&self) -> VecDeque<task::Notified<Arc<Shared>>> {
@@ -1136,7 +1134,7 @@ impl LocalState {
         // the LocalSet.
         self.assert_called_from_owner_thread();
 
-        self.owned.close_and_shutdown_all()
+        self.owned.close_and_shutdown_all();
     }
 
     #[track_caller]

tokio/src/task/task_local.rs

@@ -27,7 +27,7 @@ use std::{fmt, mem, thread};
 /// # fn main() {}
 /// ```
 ///
-/// See [LocalKey documentation][`tokio::task::LocalKey`] for more
+/// See [`LocalKey` documentation][`tokio::task::LocalKey`] for more
 /// information.
 ///
 /// [`tokio::task::LocalKey`]: struct@crate::task::LocalKey

tokio/src/task/yield_now.rs

@@ -60,5 +60,5 @@ pub async fn yield_now() {
         }
     }
 
-    YieldNow { yielded: false }.await
+    YieldNow { yielded: false }.await;
 }

tokio/src/time/clock.rs

@@ -133,7 +133,7 @@ cfg_test_util! {
                 Some(clock) => clock.pause(),
                 None => Err("time cannot be frozen from outside the Tokio runtime"),
             }
-        })
+        });
     }
 
     /// Resumes time.
@@ -161,7 +161,7 @@ cfg_test_util! {
 
             inner.unfrozen = Some(std::time::Instant::now());
             Ok(())
-        })
+        });
     }
 
     /// Advances time.

tokio/src/time/interval.rs

@@ -140,7 +140,7 @@ fn internal_interval_at(
     Interval {
         delay,
         period,
-        missed_tick_behavior: Default::default(),
+        missed_tick_behavior: MissedTickBehavior::default(),
         #[cfg(all(tokio_unstable, feature = "tracing"))]
         resource_span,
     }

tokio/src/time/sleep.rs

@@ -60,7 +60,7 @@ use std::task::{self, Poll};
 #[cfg_attr(docsrs, doc(alias = "delay_until"))]
 #[track_caller]
 pub fn sleep_until(deadline: Instant) -> Sleep {
-    return Sleep::new_timeout(deadline, trace::caller_location());
+    Sleep::new_timeout(deadline, trace::caller_location())
 }
 
 /// Waits until `duration` has elapsed.
@@ -351,7 +351,7 @@ impl Sleep {
     ///
     /// [`Pin::as_mut`]: fn@std::pin::Pin::as_mut
     pub fn reset(self: Pin<&mut Self>, deadline: Instant) {
-        self.reset_inner(deadline)
+        self.reset_inner(deadline);
     }
 
     /// Resets the `Sleep` instance to a new deadline without reregistering it
@@ -360,7 +360,7 @@ impl Sleep {
     /// Calling this function allows changing the instant at which the `Sleep`
     /// future completes without having to create new associated state and
     /// without having it registered. This is required in e.g. the
-    /// [crate::time::Interval] where we want to reset the internal [Sleep]
+    /// [`crate::time::Interval`] where we want to reset the internal [Sleep]
     /// without having it wake up the last task that polled it.
     pub(crate) fn reset_without_reregister(self: Pin<&mut Self>, deadline: Instant) {
         let mut me = self.project();

tokio/src/util/atomic_cell.rs

@@ -32,7 +32,7 @@ impl<T> AtomicCell<T> {
 }
 
 fn to_raw<T>(data: Option<Box<T>>) -> *mut T {
-    data.map(Box::into_raw).unwrap_or(ptr::null_mut())
+    data.map_or(ptr::null_mut(), Box::into_raw)
 }
 
 fn from_raw<T>(val: *mut T) -> Option<Box<T>> {

tokio/src/util/idle_notified_set.rs

@@ -124,7 +124,7 @@ unsafe impl<T> Send for ListEntry<T> {}
 unsafe impl<T> Sync for ListEntry<T> {}
 
 impl<T> IdleNotifiedSet<T> {
-    /// Create a new IdleNotifiedSet.
+    /// Create a new `IdleNotifiedSet`.
     pub(crate) fn new() -> Self {
         let lists = Mutex::new(ListsInner {
             notified: LinkedList::new(),
@@ -433,7 +433,7 @@ impl<T: 'static> Wake for ListEntry<T> {
     }
 
     fn wake(me: Arc<Self>) {
-        Self::wake_by_ref(&me)
+        Self::wake_by_ref(&me);
     }
 }
 

tokio/src/util/linked_list.rs

@@ -154,7 +154,7 @@ impl<L: Link> LinkedList<L, L::Target> {
             if let Some(prev) = L::pointers(last).as_ref().get_prev() {
                 L::pointers(prev).as_mut().set_next(None);
             } else {
-                self.head = None
+                self.head = None;
             }
 
             L::pointers(last).as_mut().set_prev(None);

tokio/src/util/wake.rs

@@ -31,7 +31,7 @@ impl Deref for WakerRef<'_> {
 
 /// Creates a reference to a `Waker` from a reference to `Arc<impl Wake>`.
 pub(crate) fn waker_ref<W: Wake>(wake: &Arc<W>) -> WakerRef<'_> {
-    let ptr = Arc::as_ptr(wake) as *const ();
+    let ptr = Arc::as_ptr(wake).cast::<()>();
 
     let waker = unsafe { Waker::from_raw(RawWaker::new(ptr, waker_vtable::<W>())) };
 
@@ -63,10 +63,10 @@ unsafe fn wake_arc_raw<T: Wake>(data: *const ()) {
 // used by `waker_ref`
 unsafe fn wake_by_ref_arc_raw<T: Wake>(data: *const ()) {
     // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop
-    let arc = ManuallyDrop::new(Arc::<T>::from_raw(data as *const T));
+    let arc = ManuallyDrop::new(Arc::<T>::from_raw(data.cast()));
     Wake::wake_by_ref(&arc);
 }
 
 unsafe fn drop_arc_raw<T: Wake>(data: *const ()) {
-    drop(Arc::<T>::from_raw(data as *const T))
+    drop(Arc::<T>::from_raw(data.cast()));
 }