fix(subscriber): only send *new* tasks/resources/etc over the event channel (#238)

## Motivation

Currently, there are some rather bad issues that occur when the event
buffer is at capacity and events are dropped.

Completely *losing* data due to buffer capacity is relatively okay: if
we set a bound on how much memory the console can use, and we don't
record new things that occur when we've reached that limit, this is
correct and acceptable behavior. However, the current design can result
in *incorrect* data when events are lost due to the buffer being at
capacity.

This is because we currently record things like starting to poll a
task/resource, ending a poll, dropping a task/resource/async op, and
waker operations, as individual events in the buffer. This means that we
can have a situation where the creation of a task was recorded, because
there was buffer capacity at the time, but then when the task ended, the
buffer was full, so we never recorded its termination. That results in
the tasks appearing to run forever and never terminate --- see issue
#230. Similarly, if we record the beginning of a poll, but drop the
end-of-poll event because we're at capacity, this means we will
(incorrectly) record a poll that goes on forever, which is obviously
incorrect. I think this may also be the cause of the false positives
with the lost waker lint (#149), if we record a waker drop but missed a
waker clone that previously occurred.

The change in #212 fixed one category of issue that occurs due to event
buffer capacity --- when a task, resource, or async op's _creation_
event is dropped due to buffer capacity, we skip any subsequent events
related to that task/resource/op. However, this doesn't fix issues where
the subsequent events are the ones that are dropped.

## Solution

This branch proposes a solution to this whole category of event buffer
capacity related issues. Unfortunately, this requires rewriting a *lot*
of `console-subscriber`'s internals.

In the new approach, we now _only_ send events over the channel when
creating a new task, resource, or async op. Those events now contain an
`Arc` holding the stats for that entity. Another clone of the `Arc` is
stored in the `tracing_subscriber::Registry`'s [span extensions] for the
span corresponding to that entity. When the `ConsoleLayer` records
subsequent events for a particular entity, such as starting/ending a
poll, it looks up the span by ID, and updates the stats type stored in
its extensions. The aggregator stores its clone of the `Arc` in a map of
entities, just like it does currently, but no longer handles actually
updating the stats; just building wire format updates from any tracked
entities whose data was updated by the layer.

This should fix all issues where dropping something due to event buffer
capacity results in incorrect data. Once we have successfully recorded
the *creation* of a task, resource, or async op, any subsequent updates
to its stats are *guaranteed* to be reliable. If the channel is at
capacity and we fail to record a new resource/task/op, we never create a
stats extension for it, and we won't record anything for it at all.
Otherwise, it will always have correct data recorded.

When possible, the stats in the `Arc`ed stats are updated atomically. In
some cases, this isn't easily possible, and some fields of the stats
types are stored in a mutex. In particualr, this is required for storing
timestamps. I don't really love that, but these mutices should be
contented very infrequently. Stats aren't marked as having unset updates
until after the stats inside the mutices have been updated, so the
aggregator will not try to lock the mutex if the layer is currently
updating it; instead, it will simply be included in the next update once
the layer is no longer touching it. Mutices here will only be contended
when multiple threads are updating a task's stats at the same time,
which should occur very rarely...and in most cases, they *still* won't
have to contend a mutex, since access to most of the mutices are guarded
by an atomic variable for e.g. determining which thread actually was the
last to complete a concurrent poll. The biggest performance downside of
the mutices is probably not actually contention, but the additional heap
allocation required when using `std::sync::Mutex`. However, since we
have conditional `parking_lot` support, parking_lot can be used to avoid
requiring additional allocations.

In the future, it's probably possible to make more of this atomic by
converting timestamps into integers and storing them in atomic
variables. I haven't done this yet because both the protobuf timestamps
and `std::time` timestamps are larger than a single 64-bit number and it
might take a little extra work to ensure we can nicely fit them in an
`AtomicUsize`...but we can probably do that later.

[span extensions]: https://docs.rs/tracing-subscriber/latest/tracing_subscriber/registry/struct.SpanRef.html#method.extensions

Signed-off-by: Eliza Weisman <[email protected]>

Author: hawkw

Cargo.lock

@@ -40,13 +40,16 @@ tracing = "0.1.26"
 tracing-subscriber = { version = "0.3.0", default-features = false, features = ["fmt", "registry"] }
 futures = { version = "0.3", default-features = false }
 hdrhistogram = { version = "7.3.0", default-features = false, features = ["serialization"] }
-serde = { version = "1", features = ["derive"] }
-serde_json = "1"
 # The parking_lot dependency is renamed, because we want our `parking_lot`
 # feature to also enable `tracing-subscriber`'s parking_lot feature flag.
 parking_lot_crate = { package = "parking_lot", version = "0.11", optional = true }
 humantime = "2.1.0"
 
+# Required for recording:
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+crossbeam-channel = "0.5"
+
 [dev-dependencies]
 tokio = { version = "^1.7", features = ["full", "rt-multi-thread"] }
 futures = "0.3"

console-subscriber/Cargo.toml

@@ -1,14 +1,12 @@
-use super::{shrink::ShrinkMap, DroppedAt, Id, ToProto};
+use super::{shrink::ShrinkMap, Id, ToProto};
+use crate::stats::{DroppedAt, Unsent};
 use std::collections::HashMap;
-use std::ops::{Deref, DerefMut};
 use std::time::{Duration, SystemTime};
 
 pub(crate) struct IdData<T> {
-    data: ShrinkMap<Id, (T, bool)>,
+    data: ShrinkMap<Id, T>,
 }
 
-pub(crate) struct Updating<'a, T>(&'a mut (T, bool));
-
 pub(crate) enum Include {
     All,
     UpdatedOnly,
@@ -19,31 +17,19 @@ pub(crate) enum Include {
 impl<T> Default for IdData<T> {
     fn default() -> Self {
         IdData {
-            data: ShrinkMap::<Id, (T, bool)>::new(),
+            data: ShrinkMap::<Id, T>::new(),
         }
     }
 }
 
-impl<T> IdData<T> {
-    pub(crate) fn update_or_default(&mut self, id: Id) -> Updating<'_, T>
-    where
-        T: Default,
-    {
-        Updating(self.data.entry(id).or_default())
-    }
-
-    pub(crate) fn update(&mut self, id: &Id) -> Option<Updating<'_, T>> {
-        self.data.get_mut(id).map(Updating)
-    }
-
+impl<T: Unsent> IdData<T> {
     pub(crate) fn insert(&mut self, id: Id, data: T) {
-        self.data.insert(id, (data, true));
+        self.data.insert(id, data);
     }
 
     pub(crate) fn since_last_update(&mut self) -> impl Iterator<Item = (&Id, &mut T)> {
-        self.data.iter_mut().filter_map(|(id, (data, dirty))| {
-            if *dirty {
-                *dirty = false;
+        self.data.iter_mut().filter_map(|(id, data)| {
+            if data.take_unsent() {
                 Some((id, data))
             } else {
                 None
@@ -52,11 +38,11 @@ impl<T> IdData<T> {
     }
 
     pub(crate) fn all(&self) -> impl Iterator<Item = (&Id, &T)> {
-        self.data.iter().map(|(id, (data, _))| (id, data))
+        self.data.iter()
     }
 
     pub(crate) fn get(&self, id: &Id) -> Option<&T> {
-        self.data.get(id).map(|(data, _)| data)
+        self.data.get(id)
     }
 
     pub(crate) fn as_proto(&mut self, include: Include) -> HashMap<u64, T::Output>
@@ -75,7 +61,7 @@ impl<T> IdData<T> {
         }
     }
 
-    pub(crate) fn drop_closed<R: DroppedAt>(
+    pub(crate) fn drop_closed<R: DroppedAt + Unsent>(
         &mut self,
         stats: &mut IdData<R>,
         now: SystemTime,
@@ -92,18 +78,19 @@ impl<T> IdData<T> {
         // drop closed entities
         tracing::trace!(?retention, has_watchers, "dropping closed");
 
-        stats.data.retain_and_shrink(|id, (stats, dirty)| {
+        stats.data.retain_and_shrink(|id, stats| {
             if let Some(dropped_at) = stats.dropped_at() {
                 let dropped_for = now.duration_since(dropped_at).unwrap_or_default();
+                let dirty = stats.is_unsent();
                 let should_drop =
                         // if there are any clients watching, retain all dirty tasks regardless of age
-                        (*dirty && has_watchers)
+                        (dirty && has_watchers)
                         || dropped_for > retention;
                 tracing::trace!(
                     stats.id = ?id,
                     stats.dropped_at = ?dropped_at,
                     stats.dropped_for = ?dropped_for,
-                    stats.dirty = *dirty,
+                    stats.dirty = dirty,
                     should_drop,
                 );
                 return !should_drop;
@@ -114,27 +101,6 @@ impl<T> IdData<T> {
 
         // drop closed entities which no longer have stats.
         self.data
-            .retain_and_shrink(|id, (_, _)| stats.data.contains_key(id));
-    }
-}
-
-// === impl Updating ===
-
-impl<'a, T> Deref for Updating<'a, T> {
-    type Target = T;
-    fn deref(&self) -> &Self::Target {
-        &self.0 .0
-    }
-}
-
-impl<'a, T> DerefMut for Updating<'a, T> {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.0 .0
-    }
-}
-
-impl<'a, T> Drop for Updating<'a, T> {
-    fn drop(&mut self) {
-        self.0 .1 = true;
+            .retain_and_shrink(|id, _| stats.data.contains_key(id));
     }
 }