Merge branch 'main' into jb/sframe

Author: jbachorik

build-logic/conventions/src/main/kotlin/com/datadoghq/native/fuzz/FuzzTargetsPlugin.kt

@@ -74,6 +74,13 @@ class FuzzTargetsPlugin : Plugin<Project> {
         }
 
         if (!hasFuzzer) {
+            val msg = if (PlatformUtils.currentPlatform == Platform.MACOS) {
+                "WARNING: libFuzzer not available on macOS — skipping fuzz targets. " +
+                "Install LLVM via Homebrew and ensure 'clang' resolves to the Homebrew clang."
+            } else {
+                "WARNING: libFuzzer not available — skipping fuzz targets (requires clang with -fsanitize=fuzzer)."
+            }
+            project.logger.lifecycle(msg)
             createListFuzzTargetsTask(project, extension)
             return
         }

build-logic/conventions/src/main/kotlin/com/datadoghq/native/gtest/GtestTaskBuilder.kt

@@ -192,7 +192,9 @@ class GtestTaskBuilder(
             // causes "incompatible ASan runtimes" → immediate abort before any test runs.
             config.testEnvironment.get()
                 .filter { (key, _) -> key != "LD_PRELOAD" }
-                .forEach { (key, value) -> environment(key, value) }
+                .forEach { (key, value) ->
+                    environment(key, hardenSanitizerOptions(key, value))
+                }
 
             inputs.files(binary)
 
@@ -257,4 +259,17 @@ class GtestTaskBuilder(
 
         return args
     }
+
+    // Gtest binaries have no JVM, so halt_on_error=0 / abort_on_error=0 (which exists
+    // to avoid conflicts with JVM signal handlers in Java integration tests) is wrong:
+    // it silently swallows ASan/TSan findings and lets the test exit 0 despite errors.
+    // Promote both flags to =1 so any sanitizer finding fails the gtest task immediately.
+    private fun hardenSanitizerOptions(key: String, value: String): String {
+        if (key != "ASAN_OPTIONS" && key != "TSAN_OPTIONS" && key != "UBSAN_OPTIONS") {
+            return value
+        }
+        return value
+            .replace("halt_on_error=0", "halt_on_error=1")
+            .replace("abort_on_error=0", "abort_on_error=1")
+    }
 }

build-logic/conventions/src/main/kotlin/com/datadoghq/native/util/PlatformUtils.kt

@@ -166,23 +166,34 @@ object PlatformUtils {
     fun checkFuzzerSupport(): Boolean {
         return try {
             val testFile = createTempFile("fuzzer_check", ".cpp")
+            val outFile = createTempFile("fuzzer_check", "")
+            // Remove the pre-created output file so the compiler writes its own binary
+            outFile.deleteIfExists()
             try {
                 testFile.writeText("extern \"C\" int LLVMFuzzerTestOneInput(const char*, long) { return 0; }")
 
+                // Link (not just compile) to catch missing libclang_rt.fuzzer_osx.a on Apple clang
                 val process = ProcessBuilder(
                     "clang++",
                     "-fsanitize=fuzzer",
-                    "-c",
                     testFile.toAbsolutePath().toString(),
                     "-o",
-                    "/dev/null"
+                    outFile.toAbsolutePath().toString()
                 ).redirectErrorStream(true).start()
 
-                process.waitFor()
+                if (!process.waitFor(30, TimeUnit.SECONDS)) {
+                    process.destroyForcibly()
+                    process.waitFor(5, TimeUnit.SECONDS)
+                    return false
+                }
                 process.exitValue() == 0
             } finally {
                 testFile.deleteIfExists()
+                outFile.deleteIfExists()
             }
+        } catch (e: InterruptedException) {
+            Thread.currentThread().interrupt()
+            false
         } catch (e: Exception) {
             false
         }

ddprof-lib/src/main/cpp/callTraceHashTable.cpp

@@ -105,8 +105,11 @@ CallTraceHashTable::~CallTraceHashTable() {
 void CallTraceHashTable::decrementCounters() {
 #ifdef COUNTERS
   // Compute and decrement the global counters for everything in this table.
-  // Must only be called after waitForAllRefCountsToClear() so there are no
-  // concurrent writers and plain iteration is safe.
+  // Safe to call when (a) this is a standby/scratch table (never _active_storage,
+  // so no signal-handler put() can target it), or (b) the active-table path is
+  // guarded by lockAll() — both conditions are enforced by the only caller,
+  // clearTableOnly().  The _prev traversal is safe because waitForRefCountToClear(this)
+  // in clearTableOnly() has already drained any in-flight put() operations.
   // Use a set to deduplicate: put() may store the same CallTrace* pointer in
   // both a newer and an older table (when findCallTrace finds it in prev()),
   // but the counter was only incremented once, so we must only count it once.
@@ -141,16 +144,27 @@ void CallTraceHashTable::decrementCounters() {
 }
 
 ChunkList CallTraceHashTable::clearTableOnly() {
-  // Wait for all refcount guards to clear before detaching chunks
-  RefCountGuard::waitForAllRefCountsToClear();
+  // Wait only for in-flight put() operations that hold a RefCountGuard on THIS
+  // table.  Waiting globally (waitForAllRefCountsToClear) would block on
+  // unrelated puts to the currently-active table, causing 500 ms timeouts under
+  // sustained wall-clock profiling and leaving collect() racing with a still-
+  // running put().  Since standby and scratch tables never appear as the
+  // _active_storage, this wait returns instantly for them; for the active table
+  // (called from clear() -> clearTableOnly()) the protection comes from the caller
+  // holding lockAll() (which blocks signal-handler puts) and from this in-function
+  // targeted wait — there is no prior caller-side drain.
+  RefCountGuard::waitForRefCountToClear(this);
   decrementCounters();
 
-  // Clear previous chain pointers to prevent traversal during deallocation
-  for (LongHashTable *table = _table; table != nullptr; table = table->prev()) {
-    LongHashTable *prev_table = table->prev();
-    if (prev_table != nullptr) {
-      table->setPrev(nullptr);  // Clear link before deallocation
-    }
+  // Disconnect the full _prev chain before freeing chunks.  The advance step
+  // must use a pre-saved pointer because setPrev(nullptr) clears the link that
+  // the original loop used for advancement, causing early termination after only
+  // the first node on an expanded (multi-node) table.
+  for (LongHashTable *table = __atomic_load_n(&_table, __ATOMIC_ACQUIRE);
+       table != nullptr; ) {
+    LongHashTable *next = table->prev();
+    table->setPrev(nullptr);
+    table = next;
   }
 
   // Detach chunks for deferred deallocation - keeps trace memory alive
@@ -161,7 +175,11 @@ ChunkList CallTraceHashTable::clearTableOnly() {
   // _tail will be nullptr. LongHashTable::allocate will try to allocate,
   // which will call LinearAllocator::alloc(), which needs to handle nullptr _tail.
   // This is already handled in alloc() by checking _tail before use.
-  _table = LongHashTable::allocate(nullptr, INITIAL_CAPACITY, &_allocator);
+  // RELEASE: pairs with ACQUIRE loads in collect() and put() to ensure the
+  // freshly-initialised table is visible on weakly-ordered architectures (aarch64).
+  __atomic_store_n(&_table,
+      LongHashTable::allocate(nullptr, INITIAL_CAPACITY, &_allocator),
+      __ATOMIC_RELEASE);
   _overflow = 0;
 
   return detached_chunks;
@@ -392,11 +410,12 @@ u64 CallTraceHashTable::put(int num_frames, ASGCT_CallFrame *frames,
 }
 
 void CallTraceHashTable::collect(std::unordered_set<CallTrace *> &traces, std::function<void(CallTrace*)> trace_hook) {
-  // Lock-free collection for read-only tables
-  // No new put() operations can occur, so no synchronization needed
-  
-  // Collect from all tables in the chain (current and previous tables)
-  for (LongHashTable *table = _table; table != nullptr; table = table->prev()) {
+  // Lock-free collection for read-only tables.
+  // Use ACQUIRE to pair with the ACQ_REL CAS in put()'s expansion path and the
+  // RELEASE store in clearTableOnly(); ensures we see the fully-initialised table
+  // on weakly-ordered architectures (aarch64).
+  for (LongHashTable *table = __atomic_load_n(&_table, __ATOMIC_ACQUIRE);
+       table != nullptr; table = table->prev()) {
     u64 *keys = table->keys();
     CallTraceSample *values = table->values();
     u32 capacity = table->capacity();
@@ -429,16 +448,19 @@ void CallTraceHashTable::putWithExistingId(CallTrace* source_trace, u64 weight)
 
   u64 hash = calcHash(source_trace->num_frames, source_trace->frames, source_trace->truncated);
 
-  // First check if trace already exists in any table in the chain
-  for (LongHashTable *search_table = _table; search_table != nullptr; search_table = search_table->prev()) {
+  // First check if trace already exists in any table in the chain.
+  // Use ACQUIRE to match the RELEASE store in clearTableOnly(); putWithExistingId()
+  // is only called on scratch/standby tables with no concurrent writers, so the
+  // load is safe, but consistent ordering prevents latent issues if callers change.
+  for (LongHashTable *search_table = __atomic_load_n(&_table, __ATOMIC_ACQUIRE);
+       search_table != nullptr; search_table = search_table->prev()) {
     CallTrace *existing_trace = findCallTrace(search_table, hash);
     if (existing_trace != nullptr) {
-      // Trace already exists in the chain
       return;
     }
   }
-  
-  LongHashTable *table = _table;
+
+  LongHashTable *table = __atomic_load_n(&_table, __ATOMIC_ACQUIRE);
   if (table == nullptr) {
     return; // Table allocation failed
   }

ddprof-lib/src/main/cpp/callTraceHashTable.h

@@ -63,8 +63,13 @@ class CallTraceHashTable {
 
   LinearAllocator _allocator;
 
-  // Single large pre-allocated table - no expansion needed!
-  LongHashTable* _table;  // Simple pointer, no atomics needed
+  // Expandable hash table; put() doubles capacity when fill reaches 75%.
+  // Memory-ordering protocol:
+  //   - ACQ_REL CAS in put() when installing the expanded table
+  //   - RELEASE store in clearTableOnly() when resetting to a fresh table
+  //   - ACQUIRE loads in collect(), put(), and putWithExistingId()
+  // Required for correct visibility on weakly-ordered architectures (aarch64).
+  LongHashTable* _table;
 
   volatile u64 _overflow;
 

ddprof-lib/src/main/cpp/linearAllocator.cpp

@@ -36,6 +36,10 @@
   #include <sanitizer/asan_interface.h>
 #endif
 
+#ifdef __SANITIZE_THREAD__
+  #include <sanitizer/tsan_interface.h>
+#endif
+
 LinearAllocator::LinearAllocator(size_t chunk_size) {
   _chunk_size = chunk_size;
   _reserve = _tail = allocateChunk(NULL);
@@ -47,9 +51,21 @@ LinearAllocator::~LinearAllocator() {
 }
 
 void LinearAllocator::clear() {
+  // OS::safeAlloc/safeFree use raw syscalls not intercepted by TSan, so TSan
+  // never clears shadow memory on munmap.  Add explicit acquire/release around
+  // every plain prev-field read so the happens-before chain from freeChunk's
+  // __tsan_release reaches any thread that later reuses the same VA.
+  #ifdef __SANITIZE_THREAD__
+  __tsan_acquire(_reserve);
+  #endif
   if (_reserve->prev == _tail) {
-    freeChunk(_reserve);
+    freeChunk(_reserve);  // __tsan_release inside
   }
+  #ifdef __SANITIZE_THREAD__
+  else {
+    __tsan_release(_reserve);  // not freed here; release for future VA-reuse acquirers
+  }
+  #endif
 
   // ASAN POISONING: Mark all allocated memory as poisoned BEFORE freeing chunks
   // This catches use-after-free even when memory isn't munmap'd (kept in _tail)
@@ -71,12 +87,26 @@ void LinearAllocator::clear() {
   }
   #endif
 
-  while (_tail->prev != NULL) {
+  // Walk the chain freeing all chunks except the last (prev==NULL).
+  // Acquire each chunk BEFORE reading its prev field so the TSan happens-before
+  // chain covers the condition check, not just the assignment inside the body.
+  {
     Chunk *current = _tail;
-    _tail = _tail->prev;
-    freeChunk(current);
+    #ifdef __SANITIZE_THREAD__
+    __tsan_acquire(current);  // before the first current->prev read (loop condition)
+    #endif
+    while (current->prev != NULL) {
+      Chunk *next = current->prev;
+      freeChunk(current);  // __tsan_release(current) + safeFree inside
+      current = next;
+      #ifdef __SANITIZE_THREAD__
+      __tsan_acquire(current);  // before the next iteration's current->prev read
+      #endif
+    }
+    // current is the last chunk (prev==NULL); keep it as the new allocator base.
+    _reserve = current;
+    _tail = current;
   }
-  _reserve = _tail;
   _tail->offs = sizeof(Chunk);
 
   // DON'T UNPOISON HERE - let alloc() do it on-demand!
@@ -88,11 +118,20 @@ ChunkList LinearAllocator::detachChunks() {
   // Capture current state before detaching
   ChunkList result(_tail, _chunk_size);
 
-  // Handle reserve chunk: if it's ahead of tail, it needs special handling
-  if (_reserve->prev == _tail) {
-    // Reserve is a separate chunk ahead of tail - it becomes part of detached list
-    // We need to include it in the chain by making it the new head
-    result.head = _reserve;
+  // Handle reserve chunk: if it's ahead of tail, it becomes part of detached list.
+  // Acquire TSan ownership before reading _reserve->prev: the reserve chunk may
+  // have been allocated by another thread via reserveChunk() → allocateChunk(),
+  // which released ownership with __tsan_release after writing chunk->prev.
+  if (_reserve != _tail) {
+    #ifdef __SANITIZE_THREAD__
+    __tsan_acquire(_reserve);
+    #endif
+    if (_reserve->prev == _tail) {
+      result.head = _reserve;
+    }
+    #ifdef __SANITIZE_THREAD__
+    __tsan_release(_reserve);
+    #endif
   }
 
   // Allocate a fresh chunk for new allocations
@@ -118,17 +157,25 @@ void LinearAllocator::freeChunks(ChunkList& chunks) {
     return;
   }
 
-  // Walk the chain and free each chunk
   Chunk* current = chunks.head;
   while (current != nullptr) {
+    // Acquire TSan ownership before reading chunk->prev: pairs with the
+    // __tsan_release in allocateChunk() that published the initialized chunk.
+    // Without this, TSan cannot connect the writer's (e.g. reserveChunk thread)
+    // initialization of chunk->prev to this read, and reports a false data race.
+    #ifdef __SANITIZE_THREAD__
+    __tsan_acquire(current);
+    #endif
     Chunk* prev = current->prev;
+    #ifdef __SANITIZE_THREAD__
+    __tsan_release(current);
+    #endif
     OS::safeFree(current, chunks.chunk_size);
     Counters::decrement(LINEAR_ALLOCATOR_BYTES, chunks.chunk_size);
     Counters::decrement(LINEAR_ALLOCATOR_CHUNKS);
     current = prev;
   }
 
-  // Mark as freed to prevent double-free
   chunks.head = nullptr;
   chunks.chunk_size = 0;
 }
@@ -172,16 +219,28 @@ void *LinearAllocator::alloc(size_t size) {
 Chunk *LinearAllocator::allocateChunk(Chunk *current) {
   Chunk *chunk = (Chunk *)OS::safeAlloc(_chunk_size);
   if (chunk != NULL) {
+    // OS::safeAlloc uses a raw mmap syscall that bypasses ASan and TSan
+    // interceptors by design (to avoid profiling self-instrumentation).
+    // When the OS reuses a VA that had stale sanitizer state from a previous
+    // allocation at that address, writing to the chunk header triggers:
+    //   ASan: use-after-poison (f7 shadow from prior ASAN_POISON_MEMORY_REGION)
+    //   TSan: data-race (prior access history for the same VA not cleared by munmap)
+    // Fix: unpoison the entire chunk and acquire TSan ownership BEFORE the first
+    // write, establishing a clean sanitizer baseline for this logical allocation.
+    #ifdef ASAN_ENABLED
+    ASAN_UNPOISON_MEMORY_REGION(chunk, _chunk_size);
+    #endif
+    #ifdef __SANITIZE_THREAD__
+    __tsan_acquire(chunk);
+    #endif
+
     chunk->prev = current;
     chunk->offs = sizeof(Chunk);
 
-    // ASAN UNPOISONING: New chunks from mmap are clean, unpoison them for use
-    // mmap returns memory that ASan may track as unallocated, so we need to
-    // explicitly unpoison it to allow allocations
-    #ifdef ASAN_ENABLED
-    size_t usable_size = _chunk_size - sizeof(Chunk);
-    void* data_start = (char*)chunk + sizeof(Chunk);
-    ASAN_UNPOISON_MEMORY_REGION(data_start, usable_size);
+    // Publish the initialized chunk: release TSan ownership so that any thread
+    // which later acquires this chunk (via __tsan_acquire) will see these writes.
+    #ifdef __SANITIZE_THREAD__
+    __tsan_release(chunk);
     #endif
 
     Counters::increment(LINEAR_ALLOCATOR_BYTES, _chunk_size);
@@ -191,6 +250,13 @@ Chunk *LinearAllocator::allocateChunk(Chunk *current) {
 }
 
 void LinearAllocator::freeChunk(Chunk *current) {
+  // Release TSan ownership before munmap so the sanitizer knows this thread is
+  // done with the memory.  The paired __tsan_acquire in allocateChunk() ensures
+  // the next thread to receive this VA (after OS VA reuse) starts with a clean
+  // happens-before baseline rather than seeing stale access history.
+  #ifdef __SANITIZE_THREAD__
+  __tsan_release(current);
+  #endif
   OS::safeFree(current, _chunk_size);
   Counters::decrement(LINEAR_ALLOCATOR_BYTES, _chunk_size);
   Counters::decrement(LINEAR_ALLOCATOR_CHUNKS);
@@ -206,7 +272,9 @@ void LinearAllocator::reserveChunk(Chunk *current) {
 }
 
 Chunk *LinearAllocator::getNextChunk(Chunk *current) {
-  Chunk *reserve = _reserve;
+  // _reserve is written via CAS in reserveChunk(); load it atomically so TSan
+  // sees the acquire-release relationship with the CAS store.
+  Chunk *reserve = __atomic_load_n(&_reserve, __ATOMIC_ACQUIRE);
 
   if (reserve == current) {
     // Unlikely case: no reserve yet.