From 9c31797a087165660a689f7430b6be1ce8eeaa63 Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Tue, 27 Aug 2019 17:53:57 -0700 Subject: [PATCH] net: switch from `log` to `tracing` (#1455) * net: switch from `log` to `tracing`. Motivation: The `tracing` crate implements scoped, structured, context-aware diagnostics, which can add significant debugging value over unstructured log messages. `tracing` is part of the Tokio project. As part of the `tokio` 0.2 changes, I thought it would be good to move over from `log` to `tracing` in the tokio runtime. Solution: This branch replaces the use of `log` in `tokio-net` with `tracing`. I've tried to leave all the instrumentation points more or less the same, but modified to use structured fields instead of string interpolation. Notes: I removed the timing in `Reactor::poll` in favor of simply adding a `#[tracing::instrument]` attribute. Since the generated `tracing` span will have enter and exit events, a `tracing::Subscriber` implemementation can use those to record timestamps, and process that timing data in a much more sophisticated manner than including it in a log line. We can add the timestamps back if they're desired. Signed-off-by: Eliza Weisman --- tokio-net/Cargo.toml | 7 ++- tokio-net/src/driver/reactor.rs | 37 ++++++------- tokio-net/src/driver/registration.rs | 3 +- tokio-net/src/lib.rs | 2 + tokio-net/src/process/unix/orphan.rs | 7 ++- tokio-net/src/tcp/stream.rs | 1 + tokio-net/src/tracing.rs | 78 ++++++++++++++++++++++++++++ tokio-net/src/udp/frame.rs | 26 ++++++---- tokio-net/src/uds/frame.rs | 25 ++++++--- tokio-net/tests/process_stdio.rs | 2 +- tokio/Cargo.toml | 1 + 11 files changed, 142 insertions(+), 47 deletions(-) create mode 100644 tokio-net/src/tracing.rs diff --git a/tokio-net/Cargo.toml b/tokio-net/Cargo.toml index f7b39d34e..a2e056382 100644 --- a/tokio-net/Cargo.toml +++ b/tokio-net/Cargo.toml @@ -63,6 +63,7 @@ uds = [ "iovec", "libc", ] +log = ["tracing/log"] [dependencies] tokio-codec = { version = "=0.2.0-alpha.2", path = "../tokio-codec" } @@ -70,11 +71,12 @@ tokio-executor = { version = "=0.2.0-alpha.2", path = "../tokio-executor" } tokio-io = { version = "=0.2.0-alpha.2", path = "../tokio-io" } tokio-sync = { version = "=0.2.0-alpha.2", path = "../tokio-sync" } +tracing = { version = "0.1.5", optional = true } + # driver implementation crossbeam-utils = "0.6.0" futures-core-preview = "=0.3.0-alpha.18" lazy_static = "1.0.2" -log = "0.4.6" mio = "0.6.14" num_cpus = "1.8.0" parking_lot = "0.9" @@ -101,6 +103,9 @@ version = "0.3" default-features = false optional = true +[target.'cfg(test)'.dependencies] +tracing = { version = "0.1.5", features = ["log"] } + [dev-dependencies] tokio = { version = "0.2.0-alpha.1", path = "../tokio" } tokio-test = { version = "0.2.0-alpha.1", path = "../tokio-test" } diff --git a/tokio-net/src/driver/reactor.rs b/tokio-net/src/driver/reactor.rs index 2aa60374d..800ab509c 100644 --- a/tokio-net/src/driver/reactor.rs +++ b/tokio-net/src/driver/reactor.rs @@ -4,7 +4,6 @@ use super::sharded_rwlock::RwLock; use tokio_executor::park::{Park, Unpark}; use tokio_sync::AtomicWaker; -use log::{debug, log_enabled, trace, Level}; use mio::event::Evented; use slab::Slab; use std::cell::RefCell; @@ -16,7 +15,7 @@ use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::{Relaxed, SeqCst}; use std::sync::{Arc, Weak}; use std::task::Waker; -use std::time::{Duration, Instant}; +use std::time::Duration; use std::{fmt, usize}; /// The core reactor, or event loop. @@ -235,6 +234,7 @@ impl Reactor { self.inner.io_dispatch.read().is_empty() } + #[cfg_attr(feature = "tracing", tracing::instrument(level = "debug"))] fn poll(&mut self, max_wait: Option) -> io::Result<()> { // Block waiting for an event to happen, peeling out how many events // happened. @@ -243,18 +243,19 @@ impl Reactor { Err(e) => return Err(e), } - let start = if log_enabled!(Level::Debug) { - Some(Instant::now()) - } else { - None - }; - // Process all the events that came in, dispatching appropriately + + // event count is only used for tracing instrumentation. + #[cfg(feature = "tracing")] let mut events = 0; + for event in self.events.iter() { - events += 1; + #[cfg(feature = "tracing")] + { + events += 1; + } let token = event.token(); - trace!("event {:?} {:?}", event.readiness(), event.token()); + trace!(event.readiness = ?event.readiness(), event.token = ?token); if token == TOKEN_WAKEUP { self.inner @@ -266,15 +267,7 @@ impl Reactor { } } - if let Some(start) = start { - let dur = start.elapsed(); - trace!( - "loop process - {} events, {}.{:03}s", - events, - dur.as_secs(), - dur.subsec_millis() - ); - } + trace!(message = "loop process", events); Ok(()) } @@ -465,7 +458,7 @@ impl Inner { }; let token = aba_guard | key; - debug!("adding I/O source: {}", token); + debug!(message = "adding I/O source", token); self.io.register( source, @@ -483,13 +476,13 @@ impl Inner { } pub(super) fn drop_source(&self, token: usize) { - debug!("dropping I/O source: {}", token); + debug!(message = "dropping I/O source", token); self.io_dispatch.write().remove(token); } /// Registers interest in the I/O resource associated with `token`. pub(super) fn register(&self, token: usize, dir: Direction, w: Waker) { - debug!("scheduling {:?} for: {}", dir, token); + debug!(message = "scheduling", direction = ?dir, token); let io_dispatch = self.io_dispatch.read(); let sched = io_dispatch.get(token).unwrap(); diff --git a/tokio-net/src/driver/registration.rs b/tokio-net/src/driver/registration.rs index eb24d8f78..4d2cd9df0 100644 --- a/tokio-net/src/driver/registration.rs +++ b/tokio-net/src/driver/registration.rs @@ -1,7 +1,6 @@ use super::platform; use super::reactor::{Direction, Handle, HandlePriv}; -use log::debug; use mio::{self, Evented}; use std::cell::UnsafeCell; use std::sync::atomic::AtomicUsize; @@ -524,7 +523,7 @@ impl Inner { if ready.is_empty() { if let Some(cx) = cx { - debug!("scheduling {:?} for: {}", direction, self.token); + debug!(message = "scheduling", ?direction, token = self.token); // Update the task info match direction { Direction::Read => sched.reader.register_by_ref(cx.waker()), diff --git a/tokio-net/src/lib.rs b/tokio-net/src/lib.rs index c071b8739..576aa32b3 100644 --- a/tokio-net/src/lib.rs +++ b/tokio-net/src/lib.rs @@ -35,6 +35,8 @@ //! [`Registration`]: struct.Registration.html //! [`PollEvented`]: struct.PollEvented.html //! [reactor module]: https://docs.rs/tokio/0.1/tokio/reactor/index.html +#[macro_use] +mod tracing; pub mod driver; pub mod util; diff --git a/tokio-net/src/process/unix/orphan.rs b/tokio-net/src/process/unix/orphan.rs index 136b4da9e..69adbbdb3 100644 --- a/tokio-net/src/process/unix/orphan.rs +++ b/tokio-net/src/process/unix/orphan.rs @@ -1,5 +1,4 @@ use crossbeam_queue::SegQueue; -use log::error; use std::io; use std::process::ExitStatus; @@ -71,9 +70,9 @@ impl OrphanQueue for AtomicOrphanQueue { match orphan.try_wait() { Ok(Some(_)) => {} Err(e) => error!( - "leaking orphaned process {} due to try_wait() error: {}", - orphan.id(), - e, + message = "leaking orphaned process due to try_wait() error", + orphan.id =orphan.id(), + error = %e, ), // Still not done yet, we need to put it back in the queue diff --git a/tokio-net/src/tcp/stream.rs b/tokio-net/src/tcp/stream.rs index caaf099d6..8741355f6 100644 --- a/tokio-net/src/tcp/stream.rs +++ b/tokio-net/src/tcp/stream.rs @@ -661,6 +661,7 @@ impl TcpStream { unsafe { buf.advance_mut(n); } + trace!(tcp.written.bytes = n); Poll::Ready(Ok(n)) } Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { diff --git a/tokio-net/src/tracing.rs b/tokio-net/src/tracing.rs new file mode 100644 index 000000000..3034bc435 --- /dev/null +++ b/tokio-net/src/tracing.rs @@ -0,0 +1,78 @@ +//! This module provides a small facade that wraps the `tracing` APIs we use, so +//! that when the `tracing` dependency is disabled, `tracing`'s macros expand to +//! no-ops. +//! +//! This means we don't have to put a `#[cfg(feature = "tracing")]` on every +//! individual use of a `tracing` macro. + +// The macros in this module may or may not be used depending on the combination +// of feature flags enabled. Rather than feature-flagging each individual macro +// to only be defined when the features that use it are enabled, just allow +// unused macros in some cases. +#![allow(unused_macros)] +#![allow(dead_code)] + +#[cfg(not(feature = "tracing"))] +#[derive(Clone, Debug)] +pub(crate) struct Span {} + +#[cfg(feature = "tracing")] +macro_rules! trace { + ($($arg:tt)+) => { + tracing::trace!($($arg)+) + }; +} + +#[cfg(not(feature = "tracing"))] +macro_rules! trace { + ($($arg:tt)+) => {}; +} + +#[cfg(feature = "tracing")] +macro_rules! debug { + ($($arg:tt)+) => { + tracing::debug!($($arg)+) + }; +} + +#[cfg(not(feature = "tracing"))] +macro_rules! debug { + ($($arg:tt)+) => {}; +} + +#[cfg(feature = "tracing")] +macro_rules! error { + ($($arg:tt)+) => { + tracing::error!($($arg)+) + }; +} + +#[cfg(not(feature = "tracing"))] +macro_rules! error { + ($($arg:tt)+) => {}; +} + +#[cfg(feature = "tracing")] +macro_rules! trace_span { + ($($arg:tt)+) => { + tracing::trace_span!($($arg)+) + }; +} + +#[cfg(not(feature = "tracing"))] +macro_rules! trace_span { + ($($arg:tt)+) => { + crate::tracing::Span::new() + }; +} + +#[cfg(not(feature = "tracing"))] +impl Span { + pub(crate) fn new() -> Self { + Span {} + } + + pub(crate) fn enter(&self) -> Span { + Span {} + } +} diff --git a/tokio-net/src/udp/frame.rs b/tokio-net/src/udp/frame.rs index b308aac46..d694e2fed 100644 --- a/tokio-net/src/udp/frame.rs +++ b/tokio-net/src/udp/frame.rs @@ -6,7 +6,6 @@ use bytes::{BufMut, BytesMut}; use core::task::{Context, Poll}; use futures_core::{ready, Stream}; use futures_sink::Sink; -use log::trace; use std::io; use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::pin::Pin; @@ -46,14 +45,18 @@ impl Stream for UdpFramed { pin.rd.reserve(INITIAL_RD_CAPACITY); - let (n, addr) = unsafe { + let (_n, addr) = unsafe { // Read into the buffer without having to initialize the memory. let res = ready!(Pin::new(&mut pin.socket).poll_recv_from_priv(cx, pin.rd.bytes_mut())); let (n, addr) = res?; pin.rd.advance_mut(n); (n, addr) }; - trace!("received {} bytes, decoding", n); + + let span = trace_span!("decoding", from.addr = %addr, dgram.length = _n); + let _e = span.enter(); + trace!("trying to decode a frame..."); + let frame_res = pin.codec.decode(&mut pin.rd); pin.rd.clear(); let frame = frame_res?; @@ -79,16 +82,18 @@ impl Sink<(C::Item, SocketAddr)> for UdpFramed { } fn start_send(self: Pin<&mut Self>, item: (C::Item, SocketAddr)) -> Result<(), Self::Error> { - trace!("sending frame"); - let (frame, out_addr) = item; + let span = trace_span!("sending", to.addr = %out_addr); + let _e = span.enter(); + trace!("encoding frame..."); + let pin = self.get_mut(); pin.codec.encode(frame, &mut pin.wr)?; pin.out_addr = out_addr; pin.flushed = false; - trace!("frame encoded; length={}", pin.wr.len()); + trace!(message = "frame encoded", frame.length = pin.wr.len()); Ok(()) } @@ -98,8 +103,6 @@ impl Sink<(C::Item, SocketAddr)> for UdpFramed { return Poll::Ready(Ok(())); } - trace!("flushing frame; length={}", self.wr.len()); - let Self { ref mut socket, ref mut out_addr, @@ -107,13 +110,18 @@ impl Sink<(C::Item, SocketAddr)> for UdpFramed { .. } = *self; + let span = trace_span!("flushing", to.addr = %out_addr, frame.length = wr.len()); + let _e = span.enter(); + trace!("flushing frame..."); + let n = ready!(socket.poll_send_to_priv(cx, &wr, &out_addr))?; - trace!("written {}", n); let wrote_all = n == self.wr.len(); self.wr.clear(); self.flushed = true; + trace!(written.length = n, written.complete = wrote_all); + let res = if wrote_all { Ok(()) } else { diff --git a/tokio-net/src/uds/frame.rs b/tokio-net/src/uds/frame.rs index 66b4fff2c..584da8ae7 100644 --- a/tokio-net/src/uds/frame.rs +++ b/tokio-net/src/uds/frame.rs @@ -1,7 +1,6 @@ use super::UnixDatagram; use bytes::{BufMut, BytesMut}; use futures::{try_ready, Async, AsyncSink, Poll, Sink, StartSend, Stream}; -use log::trace; use std::io; use std::os::unix::net::SocketAddr; use std::path::Path; @@ -41,12 +40,16 @@ impl Stream for UnixDatagramFramed { fn poll(&mut self) -> Poll, Self::Error> { self.rd.reserve(INITIAL_RD_CAPACITY); - let (n, addr) = unsafe { + let (_n, addr) = unsafe { let (n, addr) = try_ready!(self.socket.poll_recv_from(self.rd.bytes_mut())); self.rd.advance_mut(n); (n, addr) }; - trace!("received {} bytes, decoding", n); + + let span = trace_span!("decoding", from.addr = %addr, dgram.length = _n); + let _e = span.enter(); + trace!("trying to decode a frame..."); + let frame_res = self.codec.decode(&mut self.rd); self.rd.clear(); let frame = frame_res?; @@ -61,7 +64,10 @@ impl, C: Encoder> Sink for UnixDatagramFramed { type SinkError = C::Error; fn start_send(&mut self, item: Self::SinkItem) -> StartSend { - trace!("sending frame"); + let span = trace_span!("sending", to.addr = %item.0, flushed = self.flushed); + let _e = span.enter(); + + trace!("sending frame..."); if !self.flushed { match self.poll_complete()? { @@ -74,7 +80,7 @@ impl, C: Encoder> Sink for UnixDatagramFramed { self.codec.encode(frame, &mut self.wr)?; self.out_addr = Some(out_addr); self.flushed = false; - trace!("frame encoded; length={}", self.wr.len()); + trace!(message = "frame encoded", frame.length = pin.wr.len()); Ok(AsyncSink::Ready) } @@ -84,6 +90,9 @@ impl, C: Encoder> Sink for UnixDatagramFramed { return Ok(Async::Ready(())); } + let span = trace_span!("flushing", to.addr = %self.out_addr); + let _e = span.enter(); + let n = { let out_path = match self.out_addr { Some(ref out_path) => out_path.as_ref(), @@ -96,16 +105,16 @@ impl, C: Encoder> Sink for UnixDatagramFramed { } }; - trace!("flushing frame; length={}", self.wr.len()); + trace!(message = "flushing frame", frame.length = self.wr.len()); try_ready!(self.socket.poll_send_to(&self.wr, out_path)) }; - trace!("written {}", n); - let wrote_all = n == self.wr.len(); self.wr.clear(); self.flushed = true; + trace!(written.length = n, written.complete = wrote_all); + if wrote_all { self.out_addr = None; Ok(Async::Ready(())) diff --git a/tokio-net/tests/process_stdio.rs b/tokio-net/tests/process_stdio.rs index 8054f5e06..5301ed16a 100644 --- a/tokio-net/tests/process_stdio.rs +++ b/tokio-net/tests/process_stdio.rs @@ -2,7 +2,7 @@ #![warn(rust_2018_idioms)] #[macro_use] -extern crate log; +extern crate tracing; use std::env; use std::io; diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index 9e4b24ac7..d8b585f74 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -75,6 +75,7 @@ tokio-timer = { version = "=0.3.0-alpha.2", optional = true, path = "../tokio-ti tracing-core = { version = "0.1", optional = true } [target.'cfg(feature = "tracing")'.dependencies] +tokio-net = { version = "=0.2.0-alpha.2", optional = true, path = "../tokio-net", features = ["tracing", "async-traits"] } tokio-executor = { version = "=0.2.0-alpha.2", optional = true, path = "../tokio-executor", features = ["tracing"] } [dev-dependencies]