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 <[email protected]>

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" }

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<Duration>) -> 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();
 

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()),

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;

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<T: Wait> OrphanQueue<T> for AtomicOrphanQueue<T> {
             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

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 => {

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 {}
+    }
+}

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<C: Decoder + Unpin> Stream for UdpFramed<C> {
 
         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<C: Encoder + Unpin> Sink<(C::Item, SocketAddr)> for UdpFramed<C> {
     }
 
     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<C: Encoder + Unpin> Sink<(C::Item, SocketAddr)> for UdpFramed<C> {
             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<C: Encoder + Unpin> Sink<(C::Item, SocketAddr)> for UdpFramed<C> {
             ..
         } = *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 {

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<A, C: Decoder> Stream for UnixDatagramFramed<A, C> {
     fn poll(&mut self) -> Poll<Option<Self::Item>, 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<A: AsRef<Path>, C: Encoder> Sink for UnixDatagramFramed<A, C> {
     type SinkError = C::Error;
 
     fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
-        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<A: AsRef<Path>, C: Encoder> Sink for UnixDatagramFramed<A, C> {
         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<A: AsRef<Path>, C: Encoder> Sink for UnixDatagramFramed<A, C> {
             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<A: AsRef<Path>, C: Encoder> Sink for UnixDatagramFramed<A, C> {
                 }
             };
 
-            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(()))

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;

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]