core: Add intrusive linkedlist for callsite registry

tracing-core/src/callsite.rs

@@ -3,7 +3,11 @@
 use crate::stdlib::{
     fmt,
     hash::{Hash, Hasher},
-    sync::Mutex,
+    ptr,
+    sync::{
+        atomic::{AtomicPtr, Ordering},
+        Mutex, MutexGuard,
+    },
     vec::Vec,
 };
 use crate::{
@@ -13,61 +17,18 @@ use crate::{
 };
 
 lazy_static! {
-    static ref REGISTRY: Mutex<Registry> = Mutex::new(Registry {
-        callsites: Vec::new(),
-        dispatchers: Vec::new(),
-    });
-}
-
-struct Registry {
-    callsites: Vec<&'static dyn Callsite>,
-    dispatchers: Vec<dispatcher::Registrar>,
+    static ref REGISTRY: Registry = Registry {
+        callsites: LinkedList::new(),
+        dispatchers: Mutex::new(Vec::new()),
+    };
 }
 
-impl Registry {
-    fn rebuild_callsite_interest(&self, callsite: &'static dyn Callsite) {
-        let meta = callsite.metadata();
-
-        // Iterate over the subscribers in the registry, and — if they are
-        // active — register the callsite with them.
-        let mut interests = self
-            .dispatchers
-            .iter()
-            .filter_map(|registrar| registrar.try_register(meta));
-
-        // Use the first subscriber's `Interest` as the base value.
-        let interest = if let Some(interest) = interests.next() {
-            // Combine all remaining `Interest`s.
-            interests.fold(interest, Interest::and)
-        } else {
-            // If nobody was interested in this thing, just return `never`.
-            Interest::never()
-        };
-
-        callsite.set_interest(interest)
-    }
+type Dispatchers = Vec<dispatcher::Registrar>;
+type Callsites = LinkedList;
 
-    fn rebuild_interest(&mut self) {
-        let mut max_level = LevelFilter::OFF;
-        self.dispatchers.retain(|registrar| {
-            if let Some(dispatch) = registrar.upgrade() {
-                // If the subscriber did not provide a max level hint, assume
-                // that it may enable every level.
-                let level_hint = dispatch.max_level_hint().unwrap_or(LevelFilter::TRACE);
-                if level_hint > max_level {
-                    max_level = level_hint;
-                }
-                true
-            } else {
-                false
-            }
-        });
-
-        self.callsites.iter().for_each(|&callsite| {
-            self.rebuild_callsite_interest(callsite);
-        });
-        LevelFilter::set_max(max_level);
-    }
+struct Registry {
+    callsites: Callsites,
+    dispatchers: Mutex<Dispatchers>,
 }
 
 /// Trait implemented by callsites.
@@ -84,6 +45,15 @@ pub trait Callsite: Sync {
     ///
     /// [metadata]: ../metadata/struct.Metadata.html
     fn metadata(&self) -> &Metadata<'_>;
+
+    /// Returns the callsite's [`Registration`] in the global callsite registry.
+    ///
+    /// Every type implementing `Callsite` must own a single [`Registration`] which
+    /// is constructed with a static reference to that callsite. The `Registration` is
+    /// used to store the callsite in the global list of callsites. `Registration`s must be
+    /// unique per callsite; a callsite's `Registration` method must always return the
+    /// same `Registration`.
+    fn registration(&'static self) -> &'static Registration;
 }
 
 /// Uniquely identifies a [`Callsite`]
@@ -105,6 +75,16 @@ pub struct Identifier(
     pub &'static dyn Callsite,
 );
 
+/// A registration within the callsite cache.
+///
+/// Every callsite implementation must store this type internally to the
+/// callsite and provide a `&'static Registration` reference via the
+/// `Callsite` trait.
+pub struct Registration<T = &'static dyn Callsite> {
+    callsite: T,
+    next: AtomicPtr<Registration<T>>,
+}
+
 /// Clear and reregister interest on every [`Callsite`]
 ///
 /// This function is intended for runtime reconfiguration of filters on traces
@@ -125,24 +105,76 @@ pub struct Identifier(
 /// [`Interest::sometimes()`]: ../subscriber/struct.Interest.html#method.sometimes
 /// [`Subscriber`]: ../subscriber/trait.Subscriber.html
 pub fn rebuild_interest_cache() {
-    let mut registry = REGISTRY.lock().unwrap();
-    registry.rebuild_interest();
+    let mut dispatchers = REGISTRY.dispatchers.lock().unwrap();
+    let callsites = &REGISTRY.callsites;
+    rebuild_interest(callsites, &mut dispatchers);
 }
 
 /// Register a new `Callsite` with the global registry.
 ///
 /// This should be called once per callsite after the callsite has been
 /// constructed.
 pub fn register(callsite: &'static dyn Callsite) {
-    let mut registry = REGISTRY.lock().unwrap();
-    registry.rebuild_callsite_interest(callsite);
-    registry.callsites.push(callsite);
+    let mut dispatchers = REGISTRY.dispatchers.lock().unwrap();
+    rebuild_callsite_interest(&mut dispatchers, callsite);
+    REGISTRY.callsites.push(callsite);
 }
 
 pub(crate) fn register_dispatch(dispatch: &Dispatch) {
-    let mut registry = REGISTRY.lock().unwrap();
-    registry.dispatchers.push(dispatch.registrar());
-    registry.rebuild_interest();
+    let mut dispatchers = REGISTRY.dispatchers.lock().unwrap();
+    let callsites = &REGISTRY.callsites;
+
+    dispatchers.push(dispatch.registrar());
+
+    rebuild_interest(callsites, &mut dispatchers);
+}
+
+fn rebuild_callsite_interest(
+    dispatchers: &mut MutexGuard<'_, Vec<dispatcher::Registrar>>,
+    callsite: &'static dyn Callsite,
+) {
+    let meta = callsite.metadata();
+
+    // Iterate over the subscribers in the registry, and — if they are
+    // active — register the callsite with them.
+    let mut interests = dispatchers
+        .iter()
+        .filter_map(|registrar| registrar.try_register(meta));
+
+    // Use the first subscriber's `Interest` as the base value.
+    let interest = if let Some(interest) = interests.next() {
+        // Combine all remaining `Interest`s.
+        interests.fold(interest, Interest::and)
+    } else {
+        // If nobody was interested in this thing, just return `never`.
+        Interest::never()
+    };
+
+    callsite.set_interest(interest)
+}
+
+fn rebuild_interest(
+    callsites: &Callsites,
+    dispatchers: &mut MutexGuard<'_, Vec<dispatcher::Registrar>>,
+) {
+    let mut max_level = LevelFilter::OFF;
+    dispatchers.retain(|registrar| {
+        if let Some(dispatch) = registrar.upgrade() {
+            // If the subscriber did not provide a max level hint, assume
+            // that it may enable every level.
+            let level_hint = dispatch.max_level_hint().unwrap_or(LevelFilter::TRACE);
+            if level_hint > max_level {
+                max_level = level_hint;
+            }
+            true
+        } else {
+            false
+        }
+    });
+
+    callsites.for_each(|cs| rebuild_callsite_interest(dispatchers, cs));
+
+    LevelFilter::set_max(max_level);
 }
 
 // ===== impl Identifier =====
@@ -169,3 +201,165 @@ impl Hash for Identifier {
         (self.0 as *const dyn Callsite).hash(state)
     }
 }
+
+// ===== impl Registration =====
+
+impl<T> Registration<T> {
+    /// Construct a new `Registration` from some `&'static dyn Callsite`
+    pub const fn new(callsite: T) -> Self {
+        Self {
+            callsite,
+            next: AtomicPtr::new(ptr::null_mut()),
+        }
+    }
+}
+
+impl fmt::Debug for Registration {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Registration")
+            .field("callsite", &format_args!("{:p}", self.callsite))
+            .field(
+                "next",
+                &format_args!("{:p}", self.next.load(Ordering::Acquire)),
+            )
+            .finish()
+    }
+}
+
+// ===== impl LinkedList =====
+
+/// An intrusive atomic push only linked-list.
+struct LinkedList {
+    head: AtomicPtr<Registration>,
+}
+
+impl LinkedList {
+    fn new() -> Self {
+        LinkedList {
+            head: AtomicPtr::new(ptr::null_mut()),
+        }
+    }
+
+    fn for_each(&self, mut f: impl FnMut(&'static dyn Callsite)) {
+        let mut head = self.head.load(Ordering::Acquire);
+
+        while let Some(reg) = unsafe { head.as_ref() } {
+            f(reg.callsite);
+
+            head = reg.next.load(Ordering::Acquire);
+        }
+    }
+
+    fn push(&self, cs: &'static dyn Callsite) {
+        let mut head = self.head.load(Ordering::Acquire);
+
+        loop {
+            let registration = cs.registration() as *const _ as *mut _;
+
+            cs.registration().next.store(head, Ordering::Release);
+
+            assert_ne!(
+                cs.registration() as *const _,
+                head,
+                "Attempting to push a `Callsite` that already exists. \
+                        This will cause an infinite loop when attempting to read from the \
+                        callsite cache. This is likely a bug! You should only need to push a \
+                        `Callsite` once."
+            );
+
+            match self.head.compare_exchange(
+                head,
+                registration,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => {
+                    break;
+                }
+                Err(current) => head = current,
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[derive(Eq, PartialEq)]
+    struct Cs1;
+    static CS1: Cs1 = Cs1;
+    static REG1: Registration = Registration::new(&CS1);
+
+    impl Callsite for Cs1 {
+        fn set_interest(&self, _interest: Interest) {}
+        fn metadata(&self) -> &Metadata<'_> {
+            unimplemented!("not needed for this test")
+        }
+
+        fn registration(&'static self) -> &'static Registration {
+            &REG1
+        }
+    }
+
+    struct Cs2;
+    static CS2: Cs2 = Cs2;
+    static REG2: Registration = Registration::new(&CS2);
+
+    impl Callsite for Cs2 {
+        fn set_interest(&self, _interest: Interest) {}
+        fn metadata(&self) -> &Metadata<'_> {
+            unimplemented!("not needed for this test")
+        }
+
+        fn registration(&'static self) -> &'static Registration {
+            &REG2
+        }
+    }
+
+    #[test]
+    fn linked_list_push() {
+        let linked_list = LinkedList::new();
+
+        linked_list.push(&CS1);
+        linked_list.push(&CS2);
+
+        let mut i = 0;
+
+        linked_list.for_each(|cs| {
+            if i == 0 {
+                assert!(
+                    ptr::eq(cs.registration(), &REG2),
+                    "Registration pointers need to match REG2"
+                );
+            } else {
+                assert!(
+                    ptr::eq(cs.registration(), &REG1),
+                    "Registration pointers need to match REG1"
+                );
+            }
+
+            i += 1;
+        });
+    }
+
+    #[test]
+    #[should_panic]
+    fn linked_list_repeated() {
+        let linked_list = LinkedList::new();
+
+        linked_list.push(&CS1);
+        linked_list.push(&CS1);
+
+        linked_list.for_each(|_| {});
+    }
+
+    #[test]
+    fn linked_list_empty() {
+        let linked_list = LinkedList::new();
+
+        linked_list.for_each(|_| {
+            panic!("List should be empty");
+        });
+    }
+}

tracing-core/src/dispatcher.rs

@@ -783,11 +783,12 @@ impl Drop for DefaultGuard {
 
 #[cfg(test)]
 mod test {
+
     use super::*;
     #[cfg(feature = "std")]
     use crate::stdlib::sync::atomic::{AtomicUsize, Ordering};
     use crate::{
-        callsite::Callsite,
+        callsite::{Callsite, Registration},
         metadata::{Kind, Level, Metadata},
         subscriber::Interest,
     };
@@ -820,6 +821,10 @@ mod test {
         fn metadata(&self) -> &Metadata<'_> {
             &TEST_META
         }
+
+        fn registration(&'static self) -> &'static Registration {
+            unimplemented!("not needed for this test")
+        }
     }
 
     #[test]