trace: Improve support for batching field values (#880)

This branch reworks `tokio-trace`'s field recording APIs to improve
performance in cases where field values are provided in batches. 

`tokio-trace` supports recording values for span fields either when the
span is constructed, or at any point during its lifetime. In order to
support this, the current implementation represents each field value
recorded with a separate call to a `Subscriber::record_$TYPE` function.
This can incur significant overhead --- for example, if the span data is
stored behind a mutex, every call to `record` must reacquire the lock. 

Providing most or all of the fields on a span up-front as it is
constructed is likely to be the common case, but in the current design,
this use-case has to pay performance costs in order to support the less
common case of adding values post-initialization. Even in situations
where some field values are unknown, if a majority are recorded at
construction-time, each individual value takes a separate trip through
the `record` pipeline.

This branch rewrites this API so that all field recording is done by a
`ValueSet` type, which groups together multiple values. A subscriber may
pass an implementation of `Record` to the `ValueSet`'s `record`
function, which will record all the provided values with that recorder.
When a span is constructed, the `Subscriber::new_span` method is called
with a `ValueSet` containing all the fields provided to the `span!`
macro, and additional fields may be recorded either individually or in
batches.

In addition, the previous implementation of `Event`s on top of `Span`s
has been removed and replaced with a new `Event` type which goes through
a separate `Subscriber::event` function. Events are always known to have
all their values at construction time, since the event cannot live long
enough to have more values added. Thus, an `Event` also contains a
`ValueSet`.

I've added new test-support code for expecting certain fields to be
recorded by spans and events, and written new tests to cover this
behaviour. I've also updated the example code, and my experience doing
so suggests that the new API is also somewhat more ergonomic to use.

These changes result in significant improvement on the benchmark that
simulates recording multiple fields on a span 
(`bench_span_with_fields_record`).

Before:
```
     Running target/release/deps/no_subscriber-a6af9f5aa4934184

running 5 tests
test bench_1_atomic_load              ... bench:           0 ns/iter (+/- 0)
test bench_costly_field_no_subscriber ... bench:           0 ns/iter (+/- 0)
test bench_log_no_logger              ... bench:           0 ns/iter (+/- 0)
test bench_no_span_no_subscriber      ... bench:           0 ns/iter (+/- 0)
test bench_span_no_subscriber         ... bench:           0 ns/iter (+/- 0)

test result: ok. 0 passed; 0 failed; 0 ignored; 5 measured; 0 filtered out

     Running target/release/deps/subscriber-ba753dc1d87e480d

running 4 tests
test span_no_fields          ... bench:          41 ns/iter (+/- 5)
test span_repeatedly         ... bench:       5,418 ns/iter (+/- 658)
test span_with_fields        ... bench:         110 ns/iter (+/- 42)
test span_with_fields_record ... bench:       1,047 ns/iter (+/- 240)

test result: ok. 0 passed; 0 failed; 0 ignored; 4 measured; 0 filtered out
```

After:
```
     Running target/release/deps/no_subscriber-611d60b262fbbcfb

running 5 tests
test bench_1_atomic_load              ... bench:           0 ns/iter (+/- 0)
test bench_costly_field_no_subscriber ... bench:           1 ns/iter (+/- 0)
test bench_log_no_logger              ... bench:           0 ns/iter (+/- 0)
test bench_no_span_no_subscriber      ... bench:           0 ns/iter (+/- 0)
test bench_span_no_subscriber         ... bench:           0 ns/iter (+/- 0)

test result: ok. 0 passed; 0 failed; 0 ignored; 5 measured; 0 filtered out

     Running target/release/deps/subscriber-7a1279c84c0d50ff

running 4 tests
test span_no_fields          ... bench:          37 ns/iter (+/- 9)
test span_repeatedly         ... bench:       4,502 ns/iter (+/- 712)
test span_with_fields        ... bench:          54 ns/iter (+/- 17)
test span_with_fields_record ... bench:         364 ns/iter (+/- 38)

test result: ok. 0 passed; 0 failed; 0 ignored; 4 measured; 0 filtered out
```

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

tokio-trace/benches/subscriber.rs

@@ -5,23 +5,30 @@ extern crate tokio_trace;
 extern crate test;
 use test::Bencher;
 
-use std::sync::Mutex;
-use tokio_trace::{field, span, Id, Metadata};
+use std::{
+    fmt,
+    sync::{Mutex, MutexGuard},
+};
+use tokio_trace::{field, span, Event, Id, Metadata};
 
 /// A subscriber that is enabled but otherwise does nothing.
 struct EnabledSubscriber;
 
 impl tokio_trace::Subscriber for EnabledSubscriber {
-    fn new_span(&self, span: &Metadata) -> Id {
-        let _ = span;
+    fn new_span(&self, span: &Metadata, values: &field::ValueSet) -> Id {
+        let _ = (span, values);
         Id::from_u64(0)
     }
 
-    fn record_debug(&self, span: &Id, field: &field::Field, value: &::std::fmt::Debug) {
-        let _ = (span, field, value);
+    fn event(&self, event: &Event) {
+        let _ = event;
+    }
+
+    fn record(&self, span: &Id, values: &field::ValueSet) {
+        let _ = (span, values);
     }
 
-    fn add_follows_from(&self, span: &Id, follows: Id) {
+    fn record_follows_from(&self, span: &Id, follows: &Id) {
         let _ = (span, follows);
     }
 
@@ -40,19 +47,35 @@ impl tokio_trace::Subscriber for EnabledSubscriber {
 }
 
 /// Simulates a subscriber that records span data.
-struct Record(Mutex<Option<String>>);
+struct RecordingSubscriber(Mutex<String>);
 
-impl tokio_trace::Subscriber for Record {
-    fn new_span(&self, span: &Metadata) -> Id {
-        *self.0.lock().unwrap() = Some(span.name().to_string());
+struct Recorder<'a>(MutexGuard<'a, String>);
+
+impl<'a> field::Record for Recorder<'a> {
+    fn record_debug(&mut self, _field: &field::Field, value: &fmt::Debug) {
+        use std::fmt::Write;
+        let _ = write!(&mut *self.0, "{:?}", value);
+    }
+}
+
+impl tokio_trace::Subscriber for RecordingSubscriber {
+    fn new_span(&self, _span: &Metadata, values: &field::ValueSet) -> Id {
+        let mut recorder = Recorder(self.0.lock().unwrap());
+        values.record(&mut recorder);
         Id::from_u64(0)
     }
 
-    fn record_debug(&self, _span: &Id, _field: &field::Field, value: &::std::fmt::Debug) {
-        *self.0.lock().unwrap() = Some(format!("{:?}", value));
+    fn record(&self, _span: &Id, values: &field::ValueSet) {
+        let mut recorder = Recorder(self.0.lock().unwrap());
+        values.record(&mut recorder);
+    }
+
+    fn event(&self, event: &Event) {
+        let mut recorder = Recorder(self.0.lock().unwrap());
+        event.record(&mut recorder);
     }
 
-    fn add_follows_from(&self, span: &Id, follows: Id) {
+    fn record_follows_from(&self, span: &Id, follows: &Id) {
         let _ = (span, follows);
     }
 
@@ -107,7 +130,8 @@ fn span_with_fields(b: &mut Bencher) {
 
 #[bench]
 fn span_with_fields_record(b: &mut Bencher) {
-    tokio_trace::subscriber::with_default(Record(Mutex::new(None)), || {
+    let subscriber = RecordingSubscriber(Mutex::new(String::from("")));
+    tokio_trace::subscriber::with_default(subscriber, || {
         b.iter(|| {
             span!(
                 "span",

tokio-trace/examples/counters.rs

@@ -2,16 +2,18 @@
 extern crate tokio_trace;
 
 use tokio_trace::{
-    field, span,
+    field::{self, Field, Record},
+    span,
     subscriber::{self, Subscriber},
-    Id, Metadata,
+    Event, Id, Metadata,
 };
 
 use std::{
     collections::HashMap,
+    fmt,
     sync::{
         atomic::{AtomicUsize, Ordering},
-        Arc, RwLock,
+        Arc, RwLock, RwLockReadGuard,
     },
 };
 
@@ -23,8 +25,42 @@ struct CounterSubscriber {
     counters: Counters,
 }
 
+struct Count<'a> {
+    counters: RwLockReadGuard<'a, HashMap<String, AtomicUsize>>,
+}
+
+impl<'a> Record for Count<'a> {
+    fn record_i64(&mut self, field: &Field, value: i64) {
+        if let Some(counter) = self.counters.get(field.name()) {
+            if value > 0 {
+                counter.fetch_add(value as usize, Ordering::Release);
+            } else {
+                counter.fetch_sub((value * -1) as usize, Ordering::Release);
+            }
+        };
+    }
+
+    fn record_u64(&mut self, field: &Field, value: u64) {
+        if let Some(counter) = self.counters.get(field.name()) {
+            counter.fetch_add(value as usize, Ordering::Release);
+        };
+    }
+
+    fn record_bool(&mut self, _: &Field, _: bool) {}
+    fn record_str(&mut self, _: &Field, _: &str) {}
+    fn record_debug(&mut self, _: &Field, _: &fmt::Debug) {}
+}
+
+impl CounterSubscriber {
+    fn recorder(&self) -> Count {
+        Count {
+            counters: self.counters.0.read().unwrap(),
+        }
+    }
+}
+
 impl Subscriber for CounterSubscriber {
-    fn register_callsite(&self, meta: &tokio_trace::Metadata) -> subscriber::Interest {
+    fn register_callsite(&self, meta: &Metadata) -> subscriber::Interest {
         let mut interest = subscriber::Interest::never();
         for key in meta.fields() {
             let name = key.name();
@@ -41,35 +77,24 @@ impl Subscriber for CounterSubscriber {
         interest
     }
 
-    fn new_span(&self, _new_span: &Metadata) -> Id {
+    fn new_span(&self, _new_span: &Metadata, values: &field::ValueSet) -> Id {
+        values.record(&mut self.recorder());
         let id = self.ids.fetch_add(1, Ordering::SeqCst);
         Id::from_u64(id as u64)
     }
 
-    fn add_follows_from(&self, _span: &Id, _follows: Id) {
+    fn record_follows_from(&self, _span: &Id, _follows: &Id) {
         // unimplemented
     }
 
-    fn record_i64(&self, _id: &Id, field: &field::Field, value: i64) {
-        let registry = self.counters.0.read().unwrap();
-        if let Some(counter) = registry.get(field.name()) {
-            if value > 0 {
-                counter.fetch_add(value as usize, Ordering::Release);
-            } else {
-                counter.fetch_sub(value as usize, Ordering::Release);
-            }
-        };
+    fn record(&self, _: &Id, values: &field::ValueSet) {
+        values.record(&mut self.recorder())
     }
 
-    fn record_u64(&self, _id: &Id, field: &field::Field, value: u64) {
-        let registry = self.counters.0.read().unwrap();
-        if let Some(counter) = registry.get(field.name()) {
-            counter.fetch_add(value as usize, Ordering::Release);
-        };
+    fn event(&self, event: &Event) {
+        event.record(&mut self.recorder())
     }
 
-    fn record_debug(&self, _id: &Id, _field: &field::Field, _value: &::std::fmt::Debug) {}
-
     fn enabled(&self, metadata: &Metadata) -> bool {
         metadata.fields().iter().any(|f| f.name().contains("count"))
     }