Compare commits

...
Author SHA1 Message Date
Carl Lerche 42de3bc7a4 chore: prepare v0.3.3 release (#3090) 2020-11-02 15:36:17 -08:00
Alice Ryhl 20a2b9e263 rt: add missing Send bound (#3089) 2020-11-02 15:30:07 -08:00
Zeki Sherif ae4e8d7ad1 net: add get/set reuseport, reuseaddr, localaddr for TcpSocket (#3083) 2020-11-02 12:59:56 -08:00
Alice Ryhl 7a18ca2be0 doc: add from_std change to CHANGELOG (#3075) 2020-11-02 10:25:43 -08:00
Naja Melan 4a7b7c52d1 util: copy paste error in documentation for Compat (#3088) 2020-11-02 13:25:51 +01:00
Eliza Weisman fede3db76a tracing: replace future names with spawn locations in task spans (#3074)
## Motivation

Currently, the per-task `tracing` spans generated by tokio's `tracing`
feature flag include the `std::any::type_name` of the future that was
spawned. When future combinators and/or libraries like Tower are in use,
these future names can get _quite_ long. Furthermore, when formatting
the `tracing` spans with their parent spans as context, any other task
spans in the span context where the future was spawned from can _also_
include extremely long future names.

In some cases, this can result in extremely high memory use just to
store the future names. For example, in Linkerd, when we enable
`tokio=trace` to enable the task spans, there's a spawned task whose
future name is _232990 characters long_. A proxy with only 14 spawned
tasks generates a task list that's over 690 KB. Enabling task spans
under load results in the process getting OOM killed very quickly.

## Solution

This branch removes future type names from the spans generated by
`spawn`. As a replacement, to allow identifying which `spawn` call a
span corresponds to, the task span now contains the source code location
where `spawn` was called, when the compiler supports the
`#[track_caller]` attribute. Since `track_caller` was stabilized in Rust
1.46.0, and our minimum supported Rust version is 1.45.0, we can't
assume that `#[track_caller]` is always available. Instead, we have a
RUSTFLAGS cfg, `tokio_track_caller`, that guards whether or not we use
it. I've also added a `build.rs` that detects the compiler minor
version, and sets the cfg flag automatically if the current compiler
version is >= 1.46. This means users shouldn't have to enable
`tokio_track_caller` manually.

Here's the trace output from the `chat` example, before this change:
![Screenshot_20201030_110157](https://user-images.githubusercontent.com/2796466/97741071-6d408800-1a9f-11eb-9ed6-b25e72f58c7b.png)
...and after:
![Screenshot_20201030_110303](https://user-images.githubusercontent.com/2796466/97741112-7e899480-1a9f-11eb-9197-c5a3f9ea1c05.png)

Closes #3073

Signed-off-by: Eliza Weisman <[email protected]>
2020-11-01 10:48:44 -08:00
Dirkjan Ochtman 2b23aa7389 util: add back public poll_read_buf() function (#3079)
This was accidentally removed in #3064.
2020-11-01 10:22:22 +01:00
Finn Behrens 382ee6bf5d net: add pid to tokio::net::unix::UCred (#2633) 2020-10-31 10:30:55 +01:00
21 changed files with 327 additions and 58 deletions
+3 -3
View File
@@ -1,11 +1,11 @@
### Added
- io: `poll_read_buf` util fn (#2972).
# 0.5.0 (October 30, 2020)
### Changed
- io: update `bytes` to 0.6 (#3071).
### Added
- io: `poll_read_buf` util fn (#2972).
# 0.4.0 (October 15, 2020)
### Added
+1 -1
View File
@@ -150,7 +150,7 @@ where
// got room for at least one byte to read to ensure that we don't
// get a spurious 0 that looks like EOF
state.buffer.reserve(1);
let bytect = match poll_read_buf(cx, pinned.inner.as_mut(), &mut state.buffer)? {
let bytect = match poll_read_buf(pinned.inner.as_mut(), cx, &mut state.buffer)? {
Poll::Ready(ct) => ct,
Poll::Pending => return Poll::Pending,
};
+1 -1
View File
@@ -20,7 +20,7 @@ pin_project! {
/// `futures_io::AsyncRead` to implement `tokio::io::AsyncRead`.
pub trait FuturesAsyncReadCompatExt: futures_io::AsyncRead {
/// Wraps `self` with a compatibility layer that implements
/// `tokio_io::AsyncWrite`.
/// `tokio_io::AsyncRead`.
fn compat(self) -> Compat<Self>
where
Self: Sized,
+1
View File
@@ -13,3 +13,4 @@ mod stream_reader;
pub use self::read_buf::read_buf;
pub use self::reader_stream::ReaderStream;
pub use self::stream_reader::StreamReader;
pub use crate::util::poll_read_buf;
+1 -1
View File
@@ -59,7 +59,7 @@ where
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = &mut *self;
crate::util::poll_read_buf(cx, Pin::new(this.0), this.1)
crate::util::poll_read_buf(Pin::new(this.0), cx, this.1)
}
}
}
+1 -1
View File
@@ -83,7 +83,7 @@ impl<R: AsyncRead> Stream for ReaderStream<R> {
this.buf.reserve(CAPACITY);
}
match poll_read_buf(cx, reader, &mut this.buf) {
match poll_read_buf(reader, cx, &mut this.buf) {
Poll::Pending => Poll::Pending,
Poll::Ready(Err(err)) => {
self.project().reader.set(None);
+42 -3
View File
@@ -69,10 +69,49 @@ mod util {
use std::pin::Pin;
use std::task::{Context, Poll};
pub(crate) fn poll_read_buf<T: AsyncRead>(
cx: &mut Context<'_>,
/// Try to read data from an `AsyncRead` into an implementer of the [`Buf`] trait.
///
/// [`Buf`]: bytes::Buf
///
/// # Example
///
/// ```
/// use bytes::{Bytes, BytesMut};
/// use tokio::stream;
/// use tokio::io::Result;
/// use tokio_util::io::{StreamReader, poll_read_buf};
/// use futures::future::poll_fn;
/// use std::pin::Pin;
/// # #[tokio::main]
/// # async fn main() -> std::io::Result<()> {
///
/// // Create a reader from an iterator. This particular reader will always be
/// // ready.
/// let mut read = StreamReader::new(stream::iter(vec![Result::Ok(Bytes::from_static(&[0, 1, 2, 3]))]));
///
/// let mut buf = BytesMut::new();
/// let mut reads = 0;
///
/// loop {
/// reads += 1;
/// let n = poll_fn(|cx| poll_read_buf(Pin::new(&mut read), cx, &mut buf)).await?;
///
/// if n == 0 {
/// break;
/// }
/// }
///
/// // one or more reads might be necessary.
/// assert!(reads >= 1);
/// assert_eq!(&buf[..], &[0, 1, 2, 3]);
/// # Ok(())
/// # }
/// ```
#[cfg_attr(not(feature = "io"), allow(unreachable_pub))]
pub fn poll_read_buf<T: AsyncRead, B: BufMut>(
io: Pin<&mut T>,
buf: &mut impl BufMut,
cx: &mut Context<'_>,
buf: &mut B,
) -> Poll<io::Result<usize>> {
if !buf.has_remaining_mut() {
return Poll::Ready(Ok(0));
+16
View File
@@ -1,3 +1,18 @@
# 0.3.3 (November 2, 2020)
Fixes a soundness hole by adding a missing `Send` bound to
`Runtime::spawn_blocking()`.
### Fixed
- rt: include missing `Send`, fixing soundness hole (#3089).
- tracing: avoid huge trace span names (#3074).
### Added
- net: `TcpSocket::reuseport()`, `TcpSocket::set_reuseport()` (#3083).
- net: `TcpSocket::reuseaddr()` (#3093).
- net: `TcpSocket::local_addr()` (#3093).
- net: add pid to `UCred` (#2633).
# 0.3.2 (October 27, 2020)
Adds `AsyncFd` as a replacement for v0.2's `PollEvented`.
@@ -66,6 +81,7 @@ Biggest changes are:
- fs: `File` operations take `&self` (#2930).
- rt: runtime API, and `#[tokio::main]` macro polish (#2876)
- rt: `Runtime::enter` uses an RAII guard instead of a closure (#2954).
- net: the `from_std` function on all sockets no longer sets socket into non-blocking mode (#2893)
### Added
- sync: `map` function to lock guards (#2445).
+7 -4
View File
@@ -8,12 +8,12 @@ name = "tokio"
# - README.md
# - Update CHANGELOG.md.
# - Create "v0.3.x" git tag.
version = "0.3.2"
version = "0.3.3"
edition = "2018"
authors = ["Tokio Contributors <[email protected]>"]
license = "MIT"
readme = "README.md"
documentation = "https://docs.rs/tokio/0.3.2/tokio/"
documentation = "https://docs.rs/tokio/0.3.3/tokio/"
repository = "https://github.com/tokio-rs/tokio"
homepage = "https://tokio.rs"
description = """
@@ -98,11 +98,11 @@ bytes = { version = "0.6.0", optional = true }
futures-core = { version = "0.3.0", optional = true }
lazy_static = { version = "1.0.2", optional = true }
memchr = { version = "2.2", optional = true }
mio = { version = "0.7.3", optional = true }
mio = { version = "0.7.5", optional = true }
num_cpus = { version = "1.8.0", optional = true }
parking_lot = { version = "0.11.0", optional = true } # Not in full
slab = { version = "0.4.1", optional = true }
tracing = { version = "0.1.16", default-features = false, features = ["std"], optional = true } # Not in full
tracing = { version = "0.1.21", default-features = false, features = ["std"], optional = true } # Not in full
[target.'cfg(unix)'.dependencies]
libc = { version = "0.2.42", optional = true }
@@ -126,6 +126,9 @@ tempfile = "3.1.0"
[target.'cfg(loom)'.dev-dependencies]
loom = { version = "0.3.5", features = ["futures", "checkpoint"] }
[build-dependencies]
autocfg = "1" # Needed for conditionally enabling `track-caller`
[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs"]
+22
View File
@@ -0,0 +1,22 @@
use autocfg::AutoCfg;
fn main() {
match AutoCfg::new() {
Ok(ac) => {
// The #[track_caller] attribute was stabilized in rustc 1.46.0.
if ac.probe_rustc_version(1, 46) {
autocfg::emit("tokio_track_caller")
}
}
Err(e) => {
// If we couldn't detect the compiler version and features, just
// print a warning. This isn't a fatal error: we can still build
// Tokio, we just can't enable cfgs automatically.
println!(
"cargo:warning=tokio: failed to detect compiler features: {}",
e
);
}
}
}
+1 -1
View File
@@ -1,4 +1,4 @@
#![doc(html_root_url = "https://docs.rs/tokio/0.3.2")]
#![doc(html_root_url = "https://docs.rs/tokio/0.3.3")]
#![allow(
clippy::cognitive_complexity,
clippy::large_enum_variant,
+121
View File
@@ -183,6 +183,127 @@ impl TcpSocket {
self.inner.set_reuseaddr(reuseaddr)
}
/// Retrieves the value set for `SO_REUSEADDR` on this socket
///
/// # Examples
///
/// ```no_run
/// use tokio::net::TcpSocket;
///
/// use std::io;
///
/// #[tokio::main]
/// async fn main() -> io::Result<()> {
/// let addr = "127.0.0.1:8080".parse().unwrap();
///
/// let socket = TcpSocket::new_v4()?;
/// socket.set_reuseaddr(true)?;
/// assert!(socket.reuseaddr().unwrap());
/// socket.bind(addr)?;
///
/// let listener = socket.listen(1024)?;
/// Ok(())
/// }
/// ```
pub fn reuseaddr(&self) -> io::Result<bool> {
self.inner.get_reuseaddr()
}
/// Allow the socket to bind to an in-use port. Only available for unix systems
/// (excluding Solaris & Illumos).
///
/// Behavior is platform specific. Refer to the target platform's
/// documentation for more details.
///
/// # Examples
///
/// ```no_run
/// use tokio::net::TcpSocket;
///
/// use std::io;
///
/// #[tokio::main]
/// async fn main() -> io::Result<()> {
/// let addr = "127.0.0.1:8080".parse().unwrap();
///
/// let socket = TcpSocket::new_v4()?;
/// socket.set_reuseport(true)?;
/// socket.bind(addr)?;
///
/// let listener = socket.listen(1024)?;
/// Ok(())
/// }
/// ```
#[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))]
#[cfg_attr(
docsrs,
doc(cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos"))))
)]
pub fn set_reuseport(&self, reuseport: bool) -> io::Result<()> {
self.inner.set_reuseport(reuseport)
}
/// Allow the socket to bind to an in-use port. Only available for unix systems
/// (excluding Solaris & Illumos).
///
/// Behavior is platform specific. Refer to the target platform's
/// documentation for more details.
///
/// # Examples
///
/// ```no_run
/// use tokio::net::TcpSocket;
///
/// use std::io;
///
/// #[tokio::main]
/// async fn main() -> io::Result<()> {
/// let addr = "127.0.0.1:8080".parse().unwrap();
///
/// let socket = TcpSocket::new_v4()?;
/// socket.set_reuseport(true)?;
/// assert!(socket.reuseport().unwrap());
/// socket.bind(addr)?;
///
/// let listener = socket.listen(1024)?;
/// Ok(())
/// }
/// ```
#[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))]
#[cfg_attr(
docsrs,
doc(cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos"))))
)]
pub fn reuseport(&self) -> io::Result<bool> {
self.inner.get_reuseport()
}
/// Get the local address of this socket.
///
/// Will fail on windows if called before `bind`.
///
/// # Examples
///
/// ```no_run
/// use tokio::net::TcpSocket;
///
/// use std::io;
///
/// #[tokio::main]
/// async fn main() -> io::Result<()> {
/// let addr = "127.0.0.1:8080".parse().unwrap();
///
/// let socket = TcpSocket::new_v4()?;
/// socket.bind(addr)?;
/// assert_eq!(socket.local_addr().unwrap().to_string(), "127.0.0.1:8080");
/// let listener = socket.listen(1024)?;
/// Ok(())
/// }
/// ```
pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.inner.get_localaddr()
}
/// Bind the socket to the given address.
///
/// This calls the `bind(2)` operating-system function. Behavior is
+68 -7
View File
@@ -1,8 +1,10 @@
use libc::{gid_t, uid_t};
use libc::{gid_t, pid_t, uid_t};
/// Credentials of a process
#[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)]
pub struct UCred {
/// PID (process ID) of the process
pid: Option<pid_t>,
/// UID (user ID) of the process
uid: uid_t,
/// GID (group ID) of the process
@@ -19,6 +21,13 @@ impl UCred {
pub fn gid(&self) -> gid_t {
self.gid
}
/// Gets PID (process ID) of the process.
///
/// This is only implemented under linux, android, IOS and MacOS
pub fn pid(&self) -> Option<pid_t> {
self.pid
}
}
#[cfg(any(target_os = "linux", target_os = "android"))]
@@ -26,12 +35,13 @@ pub(crate) use self::impl_linux::get_peer_cred;
#[cfg(any(
target_os = "dragonfly",
target_os = "macos",
target_os = "ios",
target_os = "freebsd",
target_os = "netbsd",
target_os = "openbsd"
))]
pub(crate) use self::impl_bsd::get_peer_cred;
#[cfg(any(target_os = "macos", target_os = "ios"))]
pub(crate) use self::impl_macos::get_peer_cred;
#[cfg(any(target_os = "solaris", target_os = "illumos"))]
@@ -77,6 +87,7 @@ pub(crate) mod impl_linux {
Ok(super::UCred {
uid: ucred.uid,
gid: ucred.gid,
pid: Some(ucred.pid),
})
} else {
Err(io::Error::last_os_error())
@@ -87,13 +98,11 @@ pub(crate) mod impl_linux {
#[cfg(any(
target_os = "dragonfly",
target_os = "macos",
target_os = "ios",
target_os = "freebsd",
target_os = "netbsd",
target_os = "openbsd"
))]
pub(crate) mod impl_macos {
pub(crate) mod impl_bsd {
use crate::net::unix::UnixStream;
use libc::getpeereid;
@@ -114,6 +123,54 @@ pub(crate) mod impl_macos {
Ok(super::UCred {
uid: uid.assume_init(),
gid: gid.assume_init(),
pid: None,
})
} else {
Err(io::Error::last_os_error())
}
}
}
}
#[cfg(any(target_os = "macos", target_os = "ios"))]
pub(crate) mod impl_macos {
use crate::net::unix::UnixStream;
use libc::{c_void, getpeereid, getsockopt, pid_t, LOCAL_PEEREPID, SOL_LOCAL};
use std::io;
use std::mem::size_of;
use std::mem::MaybeUninit;
use std::os::unix::io::AsRawFd;
pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
unsafe {
let raw_fd = sock.as_raw_fd();
let mut uid = MaybeUninit::uninit();
let mut gid = MaybeUninit::uninit();
let mut pid: MaybeUninit<pid_t> = MaybeUninit::uninit();
let mut pid_size: MaybeUninit<u32> = MaybeUninit::new(size_of::<pid_t>() as u32);
if getsockopt(
raw_fd,
SOL_LOCAL,
LOCAL_PEEREPID,
pid.as_mut_ptr() as *mut c_void,
pid_size.as_mut_ptr(),
) != 0
{
return Err(io::Error::last_os_error());
}
assert!(pid_size.assume_init() == (size_of::<pid_t>() as u32));
let ret = getpeereid(raw_fd, uid.as_mut_ptr(), gid.as_mut_ptr());
if ret == 0 {
Ok(super::UCred {
uid: uid.assume_init(),
gid: gid.assume_init(),
pid: Some(pid.assume_init()),
})
} else {
Err(io::Error::last_os_error())
@@ -154,7 +211,11 @@ pub(crate) mod impl_solaris {
ucred_free(cred);
Ok(super::UCred { uid, gid })
Ok(super::UCred {
uid,
gid,
pid: None,
})
} else {
Err(io::Error::last_os_error())
}
+2
View File
@@ -70,6 +70,7 @@ const KEEP_ALIVE: Duration = Duration::from_secs(10);
pub(crate) fn spawn_blocking<F, R>(func: F) -> JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let rt = context::current().expect("not currently running on the Tokio runtime.");
rt.spawn_blocking(func)
@@ -79,6 +80,7 @@ where
pub(crate) fn try_spawn_blocking<F, R>(func: F) -> Result<(), ()>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let rt = context::current().expect("not currently running on the Tokio runtime.");
+2 -1
View File
@@ -19,7 +19,8 @@ impl<T> Unpin for BlockingTask<T> {}
impl<T, R> Future for BlockingTask<T>
where
T: FnOnce() -> R,
T: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
type Output = R;
+15 -1
View File
@@ -39,13 +39,27 @@ impl Handle {
// context::enter(self.clone(), f)
// }
/// Run the provided function on an executor dedicated to blocking operations.
/// Run the provided function on an executor dedicated to blocking
/// operations.
#[cfg_attr(tokio_track_caller, track_caller)]
pub(crate) fn spawn_blocking<F, R>(&self, func: F) -> JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
#[cfg(feature = "tracing")]
let func = {
#[cfg(tokio_track_caller)]
let location = std::panic::Location::caller();
#[cfg(tokio_track_caller)]
let span = tracing::trace_span!(
target: "tokio::task",
"task",
kind = %"blocking",
function = %std::any::type_name::<F>(),
spawn.location = %format_args!("{}:{}:{}", location.file(), location.line(), location.column()),
);
#[cfg(not(tokio_track_caller))]
let span = tracing::trace_span!(
target: "tokio::task",
"task",
+5
View File
@@ -357,11 +357,14 @@ cfg_rt! {
/// });
/// # }
/// ```
#[cfg_attr(tokio_track_caller, track_caller)]
pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
#[cfg(feature = "tracing")]
let future = crate::util::trace::task(future, "task");
match &self.kind {
#[cfg(feature = "rt-multi-thread")]
Kind::ThreadPool(exec) => exec.spawn(future),
@@ -385,9 +388,11 @@ cfg_rt! {
/// println!("now running on a worker thread");
/// });
/// # }
#[cfg_attr(tokio_track_caller, track_caller)]
pub fn spawn_blocking<F, R>(&self, func: F) -> JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
self.handle.spawn_blocking(func)
}
+1
View File
@@ -104,6 +104,7 @@ cfg_rt_multi_thread! {
/// # Ok(())
/// # }
/// ```
#[cfg_attr(tokio_track_caller, track_caller)]
pub fn spawn_blocking<F, R>(f: F) -> JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
+2
View File
@@ -190,6 +190,7 @@ cfg_rt! {
/// }).await;
/// }
/// ```
#[cfg_attr(tokio_track_caller, track_caller)]
pub fn spawn_local<F>(future: F) -> JoinHandle<F::Output>
where
F: Future + 'static,
@@ -273,6 +274,7 @@ impl LocalSet {
/// }
/// ```
/// [`spawn_local`]: fn@spawn_local
#[cfg_attr(tokio_track_caller, track_caller)]
pub fn spawn_local<F>(&self, future: F) -> JoinHandle<F::Output>
where
F: Future + 'static,
+1
View File
@@ -122,6 +122,7 @@ cfg_rt! {
/// ```text
/// error[E0391]: cycle detected when processing `main`
/// ```
#[cfg_attr(tokio_track_caller, track_caller)]
pub fn spawn<T>(task: T) -> JoinHandle<T::Output>
where
T: Future + Send + 'static,
+14 -34
View File
@@ -1,47 +1,27 @@
cfg_trace! {
cfg_rt! {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use pin_project_lite::pin_project;
use tracing::Span;
pin_project! {
/// A future that has been instrumented with a `tracing` span.
#[derive(Debug, Clone)]
pub(crate) struct Instrumented<T> {
#[pin]
inner: T,
span: Span,
}
}
impl<T: Future> Future for Instrumented<T> {
type Output = T::Output;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
let _enter = this.span.enter();
this.inner.poll(cx)
}
}
impl<T> Instrumented<T> {
pub(crate) fn new(inner: T, span: Span) -> Self {
Self { inner, span }
}
}
pub(crate) use tracing::instrument::Instrumented;
#[inline]
#[cfg_attr(tokio_track_caller, track_caller)]
pub(crate) fn task<F>(task: F, kind: &'static str) -> Instrumented<F> {
use tracing::instrument::Instrument;
#[cfg(tokio_track_caller)]
let location = std::panic::Location::caller();
#[cfg(tokio_track_caller)]
let span = tracing::trace_span!(
target: "tokio::task",
"task",
%kind,
future = %std::any::type_name::<F>(),
spawn.location = %format_args!("{}:{}:{}", location.file(), location.line(), location.column()),
);
Instrumented::new(task, span)
#[cfg(not(tokio_track_caller))]
let span = tracing::trace_span!(
target: "tokio::task",
"task",
%kind,
);
task.instrument(span)
}
}
}