From 6871084629ad95c37c7136d865890ab2e371ea12 Mon Sep 17 00:00:00 2001 From: Rafael Bachmann Date: Mon, 16 Oct 2023 17:37:51 +0200 Subject: [PATCH] chore: clippy and doc fixes (#6081) --- tokio-macros/src/entry.rs | 2 +- tokio-macros/src/select.rs | 10 +-- tokio-stream/src/once.rs | 2 +- tokio-stream/src/wrappers/mpsc_bounded.rs | 2 +- tokio-stream/src/wrappers/mpsc_unbounded.rs | 2 +- tokio-test/src/io.rs | 4 +- tokio-test/src/task.rs | 2 +- tokio-util/src/either.rs | 2 +- tokio-util/src/sync/cancellation_token.rs | 2 +- .../src/sync/cancellation_token/tree_node.rs | 8 +- tokio-util/src/sync/poll_semaphore.rs | 2 +- tokio/src/fs/dir_builder.rs | 2 +- tokio/src/fs/file.rs | 8 +- tokio/src/future/maybe_done.rs | 2 +- tokio/src/io/async_buf_read.rs | 4 +- tokio/src/io/async_fd.rs | 12 +-- tokio/src/io/blocking.rs | 4 +- tokio/src/io/interest.rs | 2 +- tokio/src/io/stdio_common.rs | 4 +- tokio/src/io/util/async_buf_read_ext.rs | 2 +- tokio/src/io/util/buf_stream.rs | 2 +- tokio/src/io/util/buf_writer.rs | 2 +- tokio/src/io/util/copy.rs | 4 +- tokio/src/io/util/read_to_end.rs | 2 +- tokio/src/io/util/take.rs | 2 +- tokio/src/net/tcp/listener.rs | 2 +- tokio/src/net/tcp/socket.rs | 6 +- tokio/src/net/tcp/stream.rs | 2 +- tokio/src/net/udp.rs | 6 +- tokio/src/net/unix/datagram/socket.rs | 4 +- tokio/src/net/unix/listener.rs | 4 +- tokio/src/net/unix/pipe.rs | 6 +- tokio/src/net/unix/split.rs | 2 +- tokio/src/net/unix/stream.rs | 4 +- tokio/src/process/unix/mod.rs | 4 +- tokio/src/process/unix/orphan.rs | 2 +- tokio/src/runtime/blocking/pool.rs | 17 ++-- tokio/src/runtime/builder.rs | 2 +- tokio/src/runtime/context/blocking.rs | 2 +- tokio/src/runtime/context/current.rs | 4 +- tokio/src/runtime/coop.rs | 2 +- tokio/src/runtime/driver.rs | 6 +- tokio/src/runtime/handle.rs | 2 +- tokio/src/runtime/io/driver.rs | 2 +- tokio/src/runtime/io/scheduled_io.rs | 2 +- tokio/src/runtime/park.rs | 6 +- tokio/src/runtime/process.rs | 2 +- tokio/src/runtime/runtime.rs | 2 +- .../runtime/scheduler/current_thread/mod.rs | 14 ++-- tokio/src/runtime/scheduler/defer.rs | 2 +- tokio/src/runtime/scheduler/mod.rs | 2 +- .../runtime/scheduler/multi_thread/park.rs | 4 +- .../runtime/scheduler/multi_thread/queue.rs | 2 +- .../runtime/scheduler/multi_thread/worker.rs | 2 +- tokio/src/runtime/signal/mod.rs | 2 +- tokio/src/runtime/task/core.rs | 2 +- tokio/src/runtime/task/mod.rs | 2 +- tokio/src/runtime/task/raw.rs | 13 +--- tokio/src/runtime/task/state.rs | 12 +-- tokio/src/runtime/time/entry.rs | 12 +-- tokio/src/runtime/time/mod.rs | 8 +- tokio/src/signal/mod.rs | 6 +- tokio/src/signal/registry.rs | 4 +- tokio/src/signal/unix.rs | 4 +- tokio/src/sync/batch_semaphore.rs | 3 +- tokio/src/sync/broadcast.rs | 6 +- tokio/src/sync/mpsc/block.rs | 9 +-- tokio/src/sync/mpsc/bounded.rs | 14 ++-- tokio/src/sync/mpsc/chan.rs | 4 +- tokio/src/sync/mpsc/list.rs | 4 +- tokio/src/sync/mpsc/unbounded.rs | 6 +- tokio/src/sync/mutex.rs | 4 +- tokio/src/sync/notify.rs | 77 +++++++++---------- tokio/src/sync/oneshot.rs | 2 +- tokio/src/sync/semaphore.rs | 8 +- tokio/src/sync/task/atomic_waker.rs | 4 +- tokio/src/sync/watch.rs | 8 +- tokio/src/task/local.rs | 12 ++- tokio/src/task/task_local.rs | 2 +- tokio/src/task/yield_now.rs | 2 +- tokio/src/time/clock.rs | 4 +- tokio/src/time/interval.rs | 2 +- tokio/src/time/sleep.rs | 6 +- tokio/src/util/atomic_cell.rs | 2 +- tokio/src/util/idle_notified_set.rs | 4 +- tokio/src/util/linked_list.rs | 2 +- tokio/src/util/wake.rs | 6 +- 87 files changed, 224 insertions(+), 239 deletions(-) diff --git a/tokio-macros/src/entry.rs b/tokio-macros/src/entry.rs index 0e31cebbb..3706026d2 100644 --- a/tokio-macros/src/entry.rs +++ b/tokio-macros/src/entry.rs @@ -586,6 +586,6 @@ impl ToTokens for Body<'_> { for stmt in self.stmts { stmt.to_tokens(tokens); } - }) + }); } } diff --git a/tokio-macros/src/select.rs b/tokio-macros/src/select.rs index dd491f848..324b8f942 100644 --- a/tokio-macros/src/select.rs +++ b/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); } } diff --git a/tokio-stream/src/once.rs b/tokio-stream/src/once.rs index 04b4c052b..c5b19bccd 100644 --- a/tokio-stream/src/once.rs +++ b/tokio-stream/src/once.rs @@ -35,7 +35,7 @@ impl Unpin for Once {} /// ``` pub fn once(value: T) -> Once { Once { - iter: crate::iter(Some(value).into_iter()), + iter: crate::iter(Some(value)), } } diff --git a/tokio-stream/src/wrappers/mpsc_bounded.rs b/tokio-stream/src/wrappers/mpsc_bounded.rs index b5362680e..18d799e98 100644 --- a/tokio-stream/src/wrappers/mpsc_bounded.rs +++ b/tokio-stream/src/wrappers/mpsc_bounded.rs @@ -34,7 +34,7 @@ impl ReceiverStream { /// /// [`Permit`]: struct@tokio::sync::mpsc::Permit pub fn close(&mut self) { - self.inner.close() + self.inner.close(); } } diff --git a/tokio-stream/src/wrappers/mpsc_unbounded.rs b/tokio-stream/src/wrappers/mpsc_unbounded.rs index 54597b7f6..6945b0871 100644 --- a/tokio-stream/src/wrappers/mpsc_unbounded.rs +++ b/tokio-stream/src/wrappers/mpsc_unbounded.rs @@ -28,7 +28,7 @@ impl UnboundedReceiverStream { /// 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(); } } diff --git a/tokio-test/src/io.rs b/tokio-test/src/io.rs index 1fc2f41aa..c31d5be5d 100644 --- a/tokio-test/src/io.rs +++ b/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."), _ => (), - }) + }); } } /* diff --git a/tokio-test/src/task.rs b/tokio-test/src/task.rs index c1cfca162..67d558dde 100644 --- a/tokio-test/src/task.rs +++ b/tokio-test/src/task.rs @@ -127,7 +127,7 @@ impl Spawn { } impl Spawn { - /// 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> { let stream = self.future.as_mut(); diff --git a/tokio-util/src/either.rs b/tokio-util/src/either.rs index 9225e53ca..8a02398bc 100644 --- a/tokio-util/src/either.rs +++ b/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)); } } diff --git a/tokio-util/src/sync/cancellation_token.rs b/tokio-util/src/sync/cancellation_token.rs index 2251736a3..5ef8ba244 100644 --- a/tokio-util/src/sync/cancellation_token.rs +++ b/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()), diff --git a/tokio-util/src/sync/cancellation_token/tree_node.rs b/tokio-util/src/sync/cancellation_token/tree_node.rs index f9068bc09..b7a98059e 100644 --- a/tokio-util/src/sync/cancellation_token/tree_node.rs +++ b/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 diff --git a/tokio-util/src/sync/poll_semaphore.rs b/tokio-util/src/sync/poll_semaphore.rs index 6b44574a1..4960a7c8b 100644 --- a/tokio-util/src/sync/poll_semaphore.rs +++ b/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. diff --git a/tokio/src/fs/dir_builder.rs b/tokio/src/fs/dir_builder.rs index 97168bff7..b2210fe25 100644 --- a/tokio/src/fs/dir_builder.rs +++ b/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). diff --git a/tokio/src/fs/file.rs b/tokio/src/fs/file.rs index 2590d305c..be2d48c27 100644 --- a/tokio/src/fs/file.rs +++ b/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<()> { diff --git a/tokio/src/future/maybe_done.rs b/tokio/src/future/maybe_done.rs index 486efbe01..d5e6fa4be 100644 --- a/tokio/src/future/maybe_done.rs +++ b/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; diff --git a/tokio/src/io/async_buf_read.rs b/tokio/src/io/async_buf_read.rs index ecaafba4c..f235b8081 100644 --- a/tokio/src/io/async_buf_read.rs +++ b/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 + Unpin> AsyncBufRead for io::Cursor { } fn consume(self: Pin<&mut Self>, amt: usize) { - io::BufRead::consume(self.get_mut(), amt) + io::BufRead::consume(self.get_mut(), amt); } } diff --git a/tokio/src/io/async_fd.rs b/tokio/src/io/async_fd.rs index ae8f5c641..b27c60bf6 100644 --- a/tokio/src/io/async_fd.rs +++ b/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 AsyncFd { - /// 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 AsyncFd { 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. /// diff --git a/tokio/src/io/blocking.rs b/tokio/src/io/blocking.rs index b988ec718..b5d7dca2b 100644 --- a/tokio/src/io/blocking.rs +++ b/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) })); diff --git a/tokio/src/io/interest.rs b/tokio/src/io/interest.rs index e823910d1..3879e9ba0 100644 --- a/tokio/src/io/interest.rs +++ b/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; } } diff --git a/tokio/src/io/stdio_common.rs b/tokio/src/io/stdio_common.rs index 06da761b8..792b3a400 100644 --- a/tokio/src/io/stdio_common.rs +++ b/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. diff --git a/tokio/src/io/util/async_buf_read_ext.rs b/tokio/src/io/util/async_buf_read_ext.rs index 50ea2d91a..2aee39258 100644 --- a/tokio/src/io/util/async_buf_read_ext.rs +++ b/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. diff --git a/tokio/src/io/util/buf_stream.rs b/tokio/src/io/util/buf_stream.rs index 595c142ac..facdbdefe 100644 --- a/tokio/src/io/util/buf_stream.rs +++ b/tokio/src/io/util/buf_stream.rs @@ -192,7 +192,7 @@ impl AsyncBufRead for BufStream { } fn consume(self: Pin<&mut Self>, amt: usize) { - self.project().inner.consume(amt) + self.project().inner.consume(amt); } } diff --git a/tokio/src/io/util/buf_writer.rs b/tokio/src/io/util/buf_writer.rs index 8dd1bba60..8f398fecd 100644 --- a/tokio/src/io/util/buf_writer.rs +++ b/tokio/src/io/util/buf_writer.rs @@ -282,7 +282,7 @@ impl AsyncBufRead for BufWriter { } fn consume(self: Pin<&mut Self>, amt: usize) { - self.get_pin_mut().consume(amt) + self.get_pin_mut().consume(amt); } } diff --git a/tokio/src/io/util/copy.rs b/tokio/src/io/util/copy.rs index 55861244f..8bd0bff7f 100644 --- a/tokio/src/io/util/copy.rs +++ b/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 diff --git a/tokio/src/io/util/read_to_end.rs b/tokio/src/io/util/read_to_end.rs index 8edba2a17..b56a940eb 100644 --- a/tokio/src/io/util/read_to_end.rs +++ b/tokio/src/io/util/read_to_end.rs @@ -54,7 +54,7 @@ pub(super) fn read_to_end_internal( } } -/// 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( diff --git a/tokio/src/io/util/take.rs b/tokio/src/io/util/take.rs index ac1f2aef4..0787defbe 100644 --- a/tokio/src/io/util/take.rs +++ b/tokio/src/io/util/take.rs @@ -43,7 +43,7 @@ impl Take { /// 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. diff --git a/tokio/src/net/tcp/listener.rs b/tokio/src/net/tcp/listener.rs index 28da34afb..f1befac26 100644 --- a/tokio/src/net/tcp/listener.rs +++ b/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) }) } diff --git a/tokio/src/net/tcp/socket.rs b/tokio/src/net/tcp/socket.rs index 1497bd612..aa9639a64 100644 --- a/tokio/src/net/tcp/socket.rs +++ b/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) -> io::Result<()> { self.inner.set_linger(dur) diff --git a/tokio/src/net/tcp/stream.rs b/tokio/src/net/tcp/stream.rs index 0a173a3d8..9b604b339 100644 --- a/tokio/src/net/tcp/stream.rs +++ b/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) }) } diff --git a/tokio/src/net/udp.rs b/tokio/src/net/udp.rs index 547aaaee0..74ea41d83 100644 --- a/tokio/src/net/udp.rs +++ b/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. diff --git a/tokio/src/net/unix/datagram/socket.rs b/tokio/src/net/unix/datagram/socket.rs index caebf8e8b..d92ad5940 100644 --- a/tokio/src/net/unix/datagram/socket.rs +++ b/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 { 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) }) } diff --git a/tokio/src/net/unix/listener.rs b/tokio/src/net/unix/listener.rs index 036acd063..a7e9115ea 100644 --- a/tokio/src/net/unix/listener.rs +++ b/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 { 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) }) } diff --git a/tokio/src/net/unix/pipe.rs b/tokio/src/net/unix/pipe.rs index 27620508c..0b2508a92 100644 --- a/tokio/src/net/unix/pipe.rs +++ b/tokio/src/net/unix/pipe.rs @@ -1188,19 +1188,19 @@ fn get_file_flags(file: &File) -> io::Result { } } -/// 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(); diff --git a/tokio/src/net/unix/split.rs b/tokio/src/net/unix/split.rs index 6fc7067a7..8b004c4a5 100644 --- a/tokio/src/net/unix/split.rs +++ b/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. diff --git a/tokio/src/net/unix/stream.rs b/tokio/src/net/unix/stream.rs index 241bcb2e2..4821260ff 100644 --- a/tokio/src/net/unix/stream.rs +++ b/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 { 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) }) } diff --git a/tokio/src/process/unix/mod.rs b/tokio/src/process/unix/mod.rs index b9c2d78e8..5b55b7a52 100644 --- a/tokio/src/process/unix/mod.rs +++ b/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 for GlobalOrphanQueue { fn push_orphan(&self, orphan: StdChild) { - get_orphan_queue().push_orphan(orphan) + get_orphan_queue().push_orphan(orphan); } } diff --git a/tokio/src/process/unix/orphan.rs b/tokio/src/process/unix/orphan.rs index 340719603..b6ca7da23 100644 --- a/tokio/src/process/unix/orphan.rs +++ b/tokio/src/process/unix/orphan.rs @@ -70,7 +70,7 @@ impl OrphanQueueImpl { 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 diff --git a/tokio/src/runtime/blocking/pool.rs b/tokio/src/runtime/blocking/pool.rs index 33778670c..3b6de8d79 100644 --- a/tokio/src/runtime/blocking/pool.rs +++ b/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 { diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs index 03f1678dc..fabafd103 100644 --- a/tokio/src/runtime/builder.rs +++ b/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, } diff --git a/tokio/src/runtime/context/blocking.rs b/tokio/src/runtime/context/blocking.rs index 8ae4f570e..39e14937e 100644 --- a/tokio/src/runtime/context/blocking.rs +++ b/tokio/src/runtime/context/blocking.rs @@ -115,7 +115,7 @@ impl Drop for DisallowBlockInPlaceGuard { allow_block_in_place: true, }); } - }) + }); } } } diff --git a/tokio/src/runtime/context/current.rs b/tokio/src/runtime/context/current.rs index c3dc5c899..d86471117 100644 --- a/tokio/src/runtime/context/current.rs +++ b/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); diff --git a/tokio/src/runtime/coop.rs b/tokio/src/runtime/coop.rs index 15a4d98c0..d9f7ff2af 100644 --- a/tokio/src/runtime/coop.rs +++ b/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) } } diff --git a/tokio/src/runtime/driver.rs b/tokio/src/runtime/driver.rs index 0474c2b3e..64928228b 100644 --- a/tokio/src/runtime/driver.rs +++ b/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); } } diff --git a/tokio/src/runtime/handle.rs b/tokio/src/runtime/handle.rs index 999352d6f..91be19ed8 100644 --- a/tokio/src/runtime/handle.rs +++ b/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. /// diff --git a/tokio/src/runtime/io/driver.rs b/tokio/src/runtime/io/driver.rs index 755cb9d14..d3055862e 100644 --- a/tokio/src/runtime/io/driver.rs +++ b/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 => { diff --git a/tokio/src/runtime/io/scheduled_io.rs b/tokio/src/runtime/io/scheduled_io.rs index 6fa5c4e65..3269d4683 100644 --- a/tokio/src/runtime/io/scheduled_io.rs +++ b/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()), } } } diff --git a/tokio/src/runtime/park.rs b/tokio/src/runtime/park.rs index 3e4374551..6f41e2c6a 100644 --- a/tokio/src/runtime/park.rs +++ b/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 { - self.unpark().map(|unpark| unpark.into_waker()) + self.unpark().map(UnparkThread::into_waker) } fn unpark(&self) -> Result { - self.with_current(|park_thread| park_thread.unpark()) + self.with_current(ParkThread::unpark) } pub(crate) fn park(&mut self) { diff --git a/tokio/src/runtime/process.rs b/tokio/src/runtime/process.rs index df339b0e7..8efb786d0 100644 --- a/tokio/src/runtime/process.rs +++ b/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); } } diff --git a/tokio/src/runtime/runtime.rs b/tokio/src/runtime/runtime.rs index d14b9b6aa..b7dfb5c93 100644 --- a/tokio/src/runtime/runtime.rs +++ b/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)); } } diff --git a/tokio/src/runtime/scheduler/current_thread/mod.rs b/tokio/src/runtime/scheduler/current_thread/mod.rs index 30b17c0e8..bc5b65ad3 100644 --- a/tokio/src/runtime/scheduler/current_thread/mod.rs +++ b/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 { impl Wake for Handle { fn wake(arc_self: Arc) { - 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(); } } } diff --git a/tokio/src/runtime/scheduler/defer.rs b/tokio/src/runtime/scheduler/defer.rs index a4be8ef2e..e7a5dde74 100644 --- a/tokio/src/runtime/scheduler/defer.rs +++ b/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(), } } diff --git a/tokio/src/runtime/scheduler/mod.rs b/tokio/src/runtime/scheduler/mod.rs index d02c0272c..42368e5be 100644 --- a/tokio/src/runtime/scheduler/mod.rs +++ b/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! { diff --git a/tokio/src/runtime/scheduler/multi_thread/park.rs b/tokio/src/runtime/scheduler/multi_thread/park.rs index 0a00ea004..87be200a1 100644 --- a/tokio/src/runtime/scheduler/multi_thread/park.rs +++ b/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) { diff --git a/tokio/src/runtime/scheduler/multi_thread/queue.rs b/tokio/src/runtime/scheduler/multi_thread/queue.rs index dd66fa2dd..41121a370 100644 --- a/tokio/src/runtime/scheduler/multi_thread/queue.rs +++ b/tokio/src/runtime/scheduler/multi_thread/queue.rs @@ -121,7 +121,7 @@ impl Local { /// 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() diff --git a/tokio/src/runtime/scheduler/multi_thread/worker.rs b/tokio/src/runtime/scheduler/multi_thread/worker.rs index 8f3181ffc..22c3e739f 100644 --- a/tokio/src/runtime/scheduler/multi_thread/worker.rs +++ b/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) { diff --git a/tokio/src/runtime/signal/mod.rs b/tokio/src/runtime/signal/mod.rs index 24f2f4c6c..0dea18794 100644 --- a/tokio/src/runtime/signal/mod.rs +++ b/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) { diff --git a/tokio/src/runtime/task/core.rs b/tokio/src/runtime/task/core.rs index 6f5867df5..1903a01aa 100644 --- a/tokio/src/runtime/task/core.rs +++ b/tokio/src/runtime/task/core.rs @@ -379,7 +379,7 @@ impl Core { unsafe fn set_stage(&self, stage: Stage) { let _guard = TaskIdGuard::enter(self.task_id); - self.stage.stage.with_mut(|ptr| *ptr = stage) + self.stage.stage.with_mut(|ptr| *ptr = stage); } } diff --git a/tokio/src/runtime/task/mod.rs b/tokio/src/runtime/task/mod.rs index e73ad93d4..ff1a455df 100644 --- a/tokio/src/runtime/task/mod.rs +++ b/tokio/src/runtime/task/mod.rs @@ -448,7 +448,7 @@ impl UnownedTask { } pub(crate) fn shutdown(self) { - self.into_task().shutdown() + self.into_task().shutdown(); } } diff --git a/tokio/src/runtime/task/raw.rs b/tokio/src/runtime/task/raw.rs index 807885928..6699551f3 100644 --- a/tokio/src/runtime/task/raw.rs +++ b/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
, } @@ -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(ptr: NonNull
) { @@ -303,7 +298,7 @@ unsafe fn try_read_output( unsafe fn drop_join_handle_slow(ptr: NonNull
) { let harness = Harness::::from_raw(ptr); - harness.drop_join_handle_slow() + harness.drop_join_handle_slow(); } unsafe fn drop_abort_handle(ptr: NonNull
) { @@ -313,5 +308,5 @@ unsafe fn drop_abort_handle(ptr: NonNull
) { unsafe fn shutdown(ptr: NonNull
) { let harness = Harness::::from_raw(ptr); - harness.shutdown() + harness.shutdown(); } diff --git a/tokio/src/runtime/task/state.rs b/tokio/src/runtime/task/state.rs index 12f544918..64cfb4b5d 100644 --- a/tokio/src/runtime/task/state.rs +++ b/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; } } diff --git a/tokio/src/runtime/time/entry.rs b/tokio/src/runtime/time/entry.rs index 798d3c11e..634ed2031 100644 --- a/tokio/src/runtime/time/entry.rs +++ b/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> { - 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(); } } diff --git a/tokio/src/runtime/time/mod.rs b/tokio/src/runtime/time/mod.rs index 423ad79ab..cdf9918ce 100644 --- a/tokio/src/runtime/time/mod.rs +++ b/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(); } } diff --git a/tokio/src/signal/mod.rs b/tokio/src/signal/mod.rs index 3aacc60ef..ab47e8af2 100644 --- a/tokio/src/signal/mod.rs +++ b/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 { diff --git a/tokio/src/signal/registry.rs b/tokio/src/signal/registry.rs index 48e98c832..022ad082b 100644 --- a/tokio/src/signal/registry.rs +++ b/tokio/src/signal/registry.rs @@ -48,7 +48,7 @@ impl Storage for Vec { where F: FnMut(&'a EventInfo), { - self.iter().for_each(f) + self.iter().for_each(f); } } @@ -87,7 +87,7 @@ impl Registry { /// 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); } } diff --git a/tokio/src/signal/unix.rs b/tokio/src/signal/unix.rs index ae5c13085..d3d7fd4ab 100644 --- a/tokio/src/signal/unix.rs +++ b/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), } diff --git a/tokio/src/sync/batch_semaphore.rs b/tokio/src/sync/batch_semaphore.rs index b6187b866..35de9a574 100644 --- a/tokio/src/sync/batch_semaphore.rs +++ b/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())); } diff --git a/tokio/src/sync/broadcast.rs b/tokio/src/sync/broadcast.rs index 94df88264..32a9f8938 100644 --- a/tokio/src/sync/broadcast.rs +++ b/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 { @@ -817,9 +817,7 @@ impl Sender { fn new_receiver(shared: Arc>) -> Receiver { 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"); diff --git a/tokio/src/sync/mpsc/block.rs b/tokio/src/sync/mpsc/block.rs index 39c3e1be2..befcfd29e 100644 --- a/tokio/src/sync/mpsc/block.rs +++ b/tokio/src/sync/mpsc/block.rs @@ -272,10 +272,9 @@ impl Block { 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 Block { /// /// # 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 Block { /// /// 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 Block { let actual = unsafe { curr.as_ref().try_push(&mut new_block, AcqRel, Acquire) }; curr = match actual { - Ok(_) => { + Ok(()) => { return next; } Err(curr) => curr, diff --git a/tokio/src/sync/mpsc/bounded.rs b/tokio/src/sync/mpsc/bounded.rs index 8924dc222..a9cd73ee3 100644 --- a/tokio/src/sync/mpsc/bounded.rs +++ b/tokio/src/sync/mpsc/bounded.rs @@ -35,7 +35,7 @@ pub struct Sender { /// [`Sender`]: Sender /// [`WeakSender::upgrade`]: WeakSender::upgrade /// -/// #Examples +/// # Examples /// /// ``` /// use tokio::sync::mpsc::channel; @@ -522,7 +522,7 @@ impl Sender { /// } /// ``` 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 Sender { /// ``` pub fn try_send(&self, message: T) -> Result<(), TrySendError> { 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 Sender { 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 Sender { /// ``` pub fn try_reserve(&self) -> Result, 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 Sender { /// ``` pub fn try_reserve_owned(self) -> Result, TrySendError> { 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 Clone for WeakSender { } impl WeakSender { - /// 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> { diff --git a/tokio/src/sync/mpsc/chan.rs b/tokio/src/sync/mpsc/chan.rs index c7c0caf6c..2540e3c2f 100644 --- a/tokio/src/sync/mpsc/chan.rs +++ b/tokio/src/sync/mpsc/chan.rs @@ -351,7 +351,7 @@ impl Drop for Rx { while let Some(Value(_)) = rx_fields.list.pop(&self.inner.tx) { self.inner.semaphore.add_permit(); } - }) + }); } } @@ -386,7 +386,7 @@ impl Drop for Chan { impl Semaphore for bounded::Semaphore { fn add_permit(&self) { - self.semaphore.release(1) + self.semaphore.release(1); } fn is_idle(&self) -> bool { diff --git a/tokio/src/sync/mpsc/list.rs b/tokio/src/sync/mpsc/list.rs index 10b29575b..a8b48a875 100644 --- a/tokio/src/sync/mpsc/list.rs +++ b/tokio/src/sync/mpsc/list.rs @@ -82,7 +82,7 @@ impl Tx { /// 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 Tx { // 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; } diff --git a/tokio/src/sync/mpsc/unbounded.rs b/tokio/src/sync/mpsc/unbounded.rs index cd83fc125..7ec5faf5b 100644 --- a/tokio/src/sync/mpsc/unbounded.rs +++ b/tokio/src/sync/mpsc/unbounded.rs @@ -25,7 +25,7 @@ pub struct UnboundedSender { /// [`UnboundedSender`]: UnboundedSender /// [`WeakUnboundedSender::upgrade`]: WeakUnboundedSender::upgrade /// -/// #Examples +/// # Examples /// /// ``` /// use tokio::sync::mpsc::unbounded_channel; @@ -379,7 +379,7 @@ impl UnboundedSender { /// } /// ``` 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 Clone for WeakUnboundedSender { } impl WeakUnboundedSender { - /// 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> { diff --git a/tokio/src/sync/mutex.rs b/tokio/src/sync/mutex.rs index 56b54a778..d420431f8 100644 --- a/tokio/src/sync/mutex.rs +++ b/tokio/src/sync/mutex.rs @@ -664,7 +664,7 @@ impl Mutex { /// ``` pub fn try_lock(&self) -> Result, 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 Mutex { /// # } pub fn try_lock_owned(self: Arc) -> Result, TryLockError> { match self.s.try_acquire(1) { - Ok(_) => { + Ok(()) => { let guard = OwnedMutexGuard { #[cfg(all(tokio_unstable, feature = "tracing"))] resource_span: self.resource_span.clone(), diff --git a/tokio/src/sync/notify.rs b/tokio/src/sync/notify.rs index c94c2bc0f..879b89b40 100644 --- a/tokio/src/sync/notify.rs +++ b/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 { - 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!(), } } diff --git a/tokio/src/sync/oneshot.rs b/tokio/src/sync/oneshot.rs index af3cc854f..26bf9f39f 100644 --- a/tokio/src/sync/oneshot.rs +++ b/tokio/src/sync/oneshot.rs @@ -712,7 +712,7 @@ impl Sender { #[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. diff --git a/tokio/src/sync/semaphore.rs b/tokio/src/sync/semaphore.rs index 7a060075d..61896c9d5 100644 --- a/tokio/src/sync/semaphore.rs +++ b/tokio/src/sync/semaphore.rs @@ -611,7 +611,7 @@ impl Semaphore { /// [`SemaphorePermit`]: crate::sync::SemaphorePermit pub fn try_acquire(&self) -> Result, 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, 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) -> Result { 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 { match self.ll_sem.try_acquire(n) { - Ok(_) => Ok(OwnedSemaphorePermit { + Ok(()) => Ok(OwnedSemaphorePermit { sem: self, permits: n, }), diff --git a/tokio/src/sync/task/atomic_waker.rs b/tokio/src/sync/task/atomic_waker.rs index 13aba3544..d06498738 100644 --- a/tokio/src/sync/task/atomic_waker.rs +++ b/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 { diff --git a/tokio/src/sync/watch.rs b/tokio/src/sync/watch.rs index 14b22cbd5..587aa795a 100644 --- a/tokio/src/sync/watch.rs +++ b/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() } diff --git a/tokio/src/task/local.rs b/tokio/src/task/local.rs index c619f1d47..9caaca629 100644 --- a/tokio/src/task/local.rs +++ b/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>> { @@ -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] diff --git a/tokio/src/task/task_local.rs b/tokio/src/task/task_local.rs index 237c4d82e..4abeeb37e 100644 --- a/tokio/src/task/task_local.rs +++ b/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 diff --git a/tokio/src/task/yield_now.rs b/tokio/src/task/yield_now.rs index 428d124c3..70a5de53d 100644 --- a/tokio/src/task/yield_now.rs +++ b/tokio/src/task/yield_now.rs @@ -60,5 +60,5 @@ pub async fn yield_now() { } } - YieldNow { yielded: false }.await + YieldNow { yielded: false }.await; } diff --git a/tokio/src/time/clock.rs b/tokio/src/time/clock.rs index 091cf4b19..50884f972 100644 --- a/tokio/src/time/clock.rs +++ b/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. diff --git a/tokio/src/time/interval.rs b/tokio/src/time/interval.rs index 48f81afcd..768fa2fef 100644 --- a/tokio/src/time/interval.rs +++ b/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, } diff --git a/tokio/src/time/sleep.rs b/tokio/src/time/sleep.rs index 6ea05699b..36f6e83c6 100644 --- a/tokio/src/time/sleep.rs +++ b/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(); diff --git a/tokio/src/util/atomic_cell.rs b/tokio/src/util/atomic_cell.rs index 07e37303a..41e44a850 100644 --- a/tokio/src/util/atomic_cell.rs +++ b/tokio/src/util/atomic_cell.rs @@ -32,7 +32,7 @@ impl AtomicCell { } fn to_raw(data: Option>) -> *mut T { - data.map(Box::into_raw).unwrap_or(ptr::null_mut()) + data.map_or(ptr::null_mut(), Box::into_raw) } fn from_raw(val: *mut T) -> Option> { diff --git a/tokio/src/util/idle_notified_set.rs b/tokio/src/util/idle_notified_set.rs index 19b81e28b..430f2e756 100644 --- a/tokio/src/util/idle_notified_set.rs +++ b/tokio/src/util/idle_notified_set.rs @@ -124,7 +124,7 @@ unsafe impl Send for ListEntry {} unsafe impl Sync for ListEntry {} impl IdleNotifiedSet { - /// 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 Wake for ListEntry { } fn wake(me: Arc) { - Self::wake_by_ref(&me) + Self::wake_by_ref(&me); } } diff --git a/tokio/src/util/linked_list.rs b/tokio/src/util/linked_list.rs index 1f9bdf4b8..cda7e3398 100644 --- a/tokio/src/util/linked_list.rs +++ b/tokio/src/util/linked_list.rs @@ -154,7 +154,7 @@ impl LinkedList { 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); diff --git a/tokio/src/util/wake.rs b/tokio/src/util/wake.rs index c872ce5d6..896ec73e7 100644 --- a/tokio/src/util/wake.rs +++ b/tokio/src/util/wake.rs @@ -31,7 +31,7 @@ impl Deref for WakerRef<'_> { /// Creates a reference to a `Waker` from a reference to `Arc`. pub(crate) fn waker_ref(wake: &Arc) -> 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::())) }; @@ -63,10 +63,10 @@ unsafe fn wake_arc_raw(data: *const ()) { // used by `waker_ref` unsafe fn wake_by_ref_arc_raw(data: *const ()) { // Retain Arc, but don't touch refcount by wrapping in ManuallyDrop - let arc = ManuallyDrop::new(Arc::::from_raw(data as *const T)); + let arc = ManuallyDrop::new(Arc::::from_raw(data.cast())); Wake::wake_by_ref(&arc); } unsafe fn drop_arc_raw(data: *const ()) { - drop(Arc::::from_raw(data as *const T)) + drop(Arc::::from_raw(data.cast())); }