mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-09-09 00:00:08 +02:00
Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
42de3bc7a4 | ||
|
|
20a2b9e263 | ||
|
|
ae4e8d7ad1 | ||
|
|
7a18ca2be0 | ||
|
|
4a7b7c52d1 | ||
|
|
fede3db76a | ||
|
|
2b23aa7389 | ||
|
|
382ee6bf5d |
@@ -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
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
@@ -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));
|
||||
|
||||
@@ -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
@@ -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"]
|
||||
|
||||
@@ -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
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
|
||||
@@ -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.");
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user