fix(profiling): add TSAN annotations to ExporterManager channel operations

When C++ tests are compiled with -fsanitize=thread but the Rust library
is not instrumented, TSAN cannot observe the atomic synchronization
inside crossbeam-channel, causing false positive data race reports.

Add __tsan_acquire/__tsan_release annotations around channel send/recv
operations using dlsym to dynamically detect TSAN at runtime. This
establishes the happens-before relationship that TSAN needs without
changing runtime behavior when sanitizers are not active.

Author: KowalskiThomas

Cargo.lock

@@ -33,6 +33,7 @@ byteorder = { version = "1.5", features = ["std"] }
 bytes = "1.11.1"
 chrono = {version = "0.4", default-features = false, features = ["std", "clock"]}
 crossbeam-channel = "0.5.15"
+libc = "0.2"
 crossbeam-utils = { version = "0.8.21" }
 cxx = { version = "1.0", optional = true }
 futures = { version = "0.3", default-features = false }

libdd-profiling/Cargo.toml

@@ -6,9 +6,65 @@ use crate::{exporter::File, internal::EncodedProfile};
 use crossbeam_channel::{Receiver, Sender};
 use libdd_common::tag::Tag;
 use reqwest::RequestBuilder;
+use std::sync::Arc;
 use std::thread::JoinHandle;
 use tokio_util::sync::CancellationToken;
 
+/// TSAN annotations for crossbeam-channel synchronization.
+///
+/// When C++ tests are compiled with `-fsanitize=thread` but the Rust library is
+/// not, TSAN cannot see the atomic synchronization inside crossbeam-channel.
+/// These annotations tell TSAN about the happens-before relationship between
+/// channel send and receive operations via `__tsan_release`/`__tsan_acquire`.
+mod tsan {
+    use std::ffi::c_void;
+    use std::sync::OnceLock;
+
+    type TsanAnnotateFn = unsafe extern "C" fn(*mut c_void);
+
+    struct TsanFunctions {
+        acquire: Option<TsanAnnotateFn>,
+        release: Option<TsanAnnotateFn>,
+    }
+
+    static FUNCTIONS: OnceLock<TsanFunctions> = OnceLock::new();
+
+    fn get() -> &'static TsanFunctions {
+        FUNCTIONS.get_or_init(|| unsafe {
+            let acquire =
+                libc::dlsym(libc::RTLD_DEFAULT, b"__tsan_acquire\0".as_ptr().cast());
+            let release =
+                libc::dlsym(libc::RTLD_DEFAULT, b"__tsan_release\0".as_ptr().cast());
+            TsanFunctions {
+                acquire: if acquire.is_null() {
+                    None
+                } else {
+                    Some(std::mem::transmute(acquire))
+                },
+                release: if release.is_null() {
+                    None
+                } else {
+                    Some(std::mem::transmute(release))
+                },
+            }
+        })
+    }
+
+    /// Signal that data associated with `addr` has been published (sender side).
+    pub fn release(addr: *const u8) {
+        if let Some(f) = get().release {
+            unsafe { f(addr as *mut c_void) };
+        }
+    }
+
+    /// Signal that data associated with `addr` has been consumed (receiver side).
+    pub fn acquire(addr: *const u8) {
+        if let Some(f) = get().acquire {
+            unsafe { f(addr as *mut c_void) };
+        }
+    }
+}
+
 #[derive(Debug)]
 pub enum ExporterManager {
     Active {
@@ -17,6 +73,8 @@ pub enum ExporterManager {
         handle: JoinHandle<()>,
         sender: Sender<RequestBuilder>,
         receiver: Receiver<RequestBuilder>,
+        /// Stable heap address used for TSAN release/acquire annotations.
+        tsan_sync: Arc<u8>,
     },
     Suspended {
         exporter: ProfileExporter,
@@ -31,8 +89,10 @@ impl ExporterManager {
     pub fn new(exporter: ProfileExporter) -> anyhow::Result<Self> {
         let (sender, receiver) = crossbeam_channel::bounded(2);
         let cancel = CancellationToken::new();
+        let tsan_sync = Arc::new(0u8);
         let cloned_receiver: Receiver<RequestBuilder> = receiver.clone();
         let cloned_cancel = cancel.clone();
+        let cloned_tsan_sync = tsan_sync.clone();
         let handle = std::thread::spawn(move || {
             let Ok(runtime) = tokio::runtime::Builder::new_current_thread()
                 .enable_all()
@@ -46,6 +106,7 @@ impl ExporterManager {
                     let Ok(msg) = cloned_receiver.recv() else {
                         return;
                     };
+                    tsan::acquire(Arc::as_ptr(&cloned_tsan_sync));
                     if cloned_cancel
                         .run_until_cancelled(msg.send())
                         .await
@@ -63,6 +124,7 @@ impl ExporterManager {
             handle,
             sender,
             receiver,
+            tsan_sync,
         })
     }
 
@@ -75,6 +137,7 @@ impl ExporterManager {
             handle,
             sender,
             receiver,
+            tsan_sync: _,
         } = old
         else {
             *self = old;
@@ -147,7 +210,10 @@ impl ExporterManager {
         process_tags: Option<&str>,
     ) -> anyhow::Result<()> {
         let Self::Active {
-            exporter, sender, ..
+            exporter,
+            sender,
+            tsan_sync,
+            ..
         } = self
         else {
             anyhow::bail!("Cannot queue on manager in state: {:?}", self);
@@ -163,6 +229,7 @@ impl ExporterManager {
         )?;
         // TODO, use thiserror and get back the actual error if one.
         sender.try_send(msg)?;
+        tsan::release(Arc::as_ptr(tsan_sync));
         Ok(())
     }