Skip to content

Commit

Permalink
tracing: make Entered !Send (#1001)
Browse files Browse the repository at this point in the history 
## Motivation

The `Entered` guard returned by `Span::enter` represents entering and
exiting a span _on the current thread_. Calling `Span::enter` enters the
span, returning an `Entered`, and dropping the `Entered` ensures that
the span is exited. This ensures that all spans, once entered, are
eventually exited.

However, `Entered` has an auto-impl of `Send`, because it doesn't
contain any `!Send` types. This means that the `Entered` guard _could_
be sent to another thread and dropped. This would cause the original
thread to never exit the span,. and the thread to which the `Entered`
guard was sent would exit a span that it never observed an enter for.
This is incorrect.

## Solution

This PR adds a `*mut ()` field to `Entered` so that it no longer
implements `Send`. There's now a manual `Sync` impl so structs holding
an `Entered` can still be `Sync`.

Fixes #698

Signed-off-by: Eliza Weisman <eliza@buoyant.io>
  • Loading branch information
@hawkw
hawkw authored Sep 30, 2020
1 parent 2f59b32 commit 515255f
Show file tree
Hide file tree
Showing 2 changed files with 96 additions and 67 deletions.
126 changes: 61 additions & 65 deletions examples/examples/hyper-echo.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,78 +16,75 @@ async fn echo(req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
uri = ?req.uri(),
headers = ?req.headers()
);
let _enter = span.enter();
info!("received request");
let mut response = Response::new(Body::empty());
async move {
info!("received request");
let mut response = Response::new(Body::empty());

let (rsp_span, resp) = match (req.method(), req.uri().path()) {
// Serve some instructions at /
(&Method::GET, "/") => {
const BODY: &str = "Try POSTing data to /echo";
*response.body_mut() = Body::from(BODY);
(span!(Level::INFO, "response", body = %(&BODY)), response)
}
match (req.method(), req.uri().path()) {
// Serve some instructions at /
(&Method::GET, "/") => {
const BODY: &str = "Try POSTing data to /echo";
*response.body_mut() = Body::from(BODY);
info!(body = %(&BODY), "response",);
Ok(response)
}

// Simply echo the body back to the client.
(&Method::POST, "/echo") => {
let span = span!(Level::INFO, "response", response_kind = %"echo");
*response.body_mut() = req.into_body();
(span, response)
}
// Simply echo the body back to the client.
(&Method::POST, "/echo") => {
info!(response_kind = %"echo", "response");
*response.body_mut() = req.into_body();
Ok(response)
}

// Convert to uppercase before sending back to client.
(&Method::POST, "/echo/uppercase") => {
let body = hyper::body::to_bytes(req).await?;
let upper = body
.iter()
.map(|byte| byte.to_ascii_uppercase())
.collect::<Vec<u8>>();
debug!(
body = ?str::from_utf8(&body[..]),
uppercased = ?str::from_utf8(&upper[..]),
"uppercased request body"
);
// Convert to uppercase before sending back to client.
(&Method::POST, "/echo/uppercase") => {
let body = hyper::body::to_bytes(req).await?;
let upper = body
.iter()
.map(|byte| byte.to_ascii_uppercase())
.collect::<Vec<u8>>();
debug!(
body = ?str::from_utf8(&body[..]),
uppercased = ?str::from_utf8(&upper[..]),
"uppercased request body"
);

*response.body_mut() = Body::from(upper);
(
span!(Level::INFO, "response", response_kind = %"uppercase"),
response,
)
}
info!(response_kind = %"uppercase", "response");
*response.body_mut() = Body::from(upper);
Ok(response)
}

// Reverse the entire body before sending back to the client.
(&Method::POST, "/echo/reversed") => {
let span = span!(Level::TRACE, "response", response_kind = %"reversed");
let _enter = span.enter();
let body = hyper::body::to_bytes(req).await?;
let reversed = body.iter().rev().cloned().collect::<Vec<u8>>();
debug!(
body = ?str::from_utf8(&body[..]),
"reversed request body"
);
*response.body_mut() = Body::from(reversed);
(
span!(Level::INFO, "reversed", body = ?(&response.body())),
response,
)
}
// Reverse the entire body before sending back to the client.
(&Method::POST, "/echo/reversed") => {
async move {
let body = hyper::body::to_bytes(req).await?;
let reversed = body.iter().rev().cloned().collect::<Vec<u8>>();
debug!(
body = ?str::from_utf8(&body[..]),
"reversed request body"
);
*response.body_mut() = Body::from(reversed);
info!(body = ?(&response.body()), "response");
Ok(response)
}
.instrument(span!(Level::TRACE, "response", response_kind = %"reversed"))
.await
}

// The 404 Not Found route...
_ => {
*response.status_mut() = StatusCode::NOT_FOUND;
(
span!(
Level::TRACE,
"response",
// The 404 Not Found route...
_ => {
*response.status_mut() = StatusCode::NOT_FOUND;
info!(
body = ?(),
status = ?StatusCode::NOT_FOUND,
),
response,
)
"response",
);
Ok(response)
}
}
};
let f = async { resp }.instrument(rsp_span);
Ok(f.await)
}
.instrument(span)
.await
}

#[tokio::main]
Expand All @@ -107,14 +104,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {

let local_addr: std::net::SocketAddr = ([127, 0, 0, 1], 3000).into();
let server_span = span!(Level::TRACE, "server", %local_addr);
let _enter = server_span.enter();

let service = make_service_fn(|_| async { Ok::<_, hyper::Error>(service_fn(echo)) });
let server = Server::bind(&local_addr)
.serve(service)
.instrument(server_span.clone());

info!("listening...");
info!(parent: &server_span, "listening...");
server.await?;

Ok(())
Expand Down
37 changes: 35 additions & 2 deletions tracing/src/span.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -318,6 +318,7 @@ pub use tracing_core::span::{Attributes, Id, Record};
use crate::stdlib::{
cmp, fmt,
hash::{Hash, Hasher},
marker::PhantomData,
};
use crate::{
dispatcher::{self, Dispatch},
Expand Down Expand Up @@ -379,6 +380,7 @@ pub(crate) struct Inner {
#[must_use = "once a span has been entered, it should be exited"]
pub struct Entered<'a> {
span: &'a Span,
_not_send: PhantomData<*mut ()>,
}

/// `log` target for all span lifecycle (creation/enter/exit/close) records.
Expand Down Expand Up @@ -562,7 +564,17 @@ impl Span {
/// will call [`Subscriber::exit`]. If the span is disabled, this does
/// nothing.
///
/// # In Asynchronous Code
/// <div class="information">
/// <div class="tooltip ignore" style="">ⓘ<span class="tooltiptext">Note</span></div>
/// </div>
/// <div class="example-wrap" style="display:inline-block">
/// <pre class="ignore" style="white-space:normal;font:inherit;">
/// <strong>Note</strong>: The returned
/// <a href="../struct.Entered.html"><code>Entered</code></a> guard does not
/// implement <code>Send</code>. Dropping the guard will exit <em>this</em> span,
/// and if the guard is sent to another thread and dropped there, that thread may
/// never have entered this span. Thus, <code>Entered</code> should not be sent
/// between threads.</pre></div>
///
/// **Warning**: in asynchronous code that uses [async/await syntax][syntax],
/// `Span::enter` should be used very carefully or avoided entirely. Holding
Expand Down Expand Up @@ -755,7 +767,10 @@ impl Span {
}
}}

Entered { span: self }
Entered {
span: self,
_not_send: PhantomData,
}
}

/// Executes the given function in the context of this span.
Expand Down Expand Up @@ -1220,6 +1235,24 @@ impl Clone for Inner {

// ===== impl Entered =====

/// # Safety
///
/// Technically, `Entered` _can_ implement both `Send` *and* `Sync` safely. It
/// doesn't, because it has a `PhantomData<*mut ()>` field, specifically added
/// in order to make it `!Send`.
///
/// Sending an `Entered` guard between threads cannot cause memory unsafety.
/// However, it *would* result in incorrect behavior, so we add a
/// `PhantomData<*mut ()>` to prevent it from being sent between threads. This
/// is because it must be *dropped* on the same thread that it was created;
/// otherwise, the span will never be exited on the thread where it was entered,
/// and it will attempt to exit the span on a thread that may never have entered
/// it. However, we still want them to be `Sync` so that a struct holding an
/// `Entered` guard can be `Sync`.
///
/// Thus, this is totally safe.
unsafe impl<'a> Sync for Entered<'a> {}

impl<'a> Drop for Entered<'a> {
#[inline]
fn drop(&mut self) {
Expand Down

0 comments on commit 515255f

Please sign in to comment.