bench(profiling): measure live heap tracking paths

Author: realFlowControl

profiling/Cargo.toml

@@ -61,6 +61,10 @@ perfcnt = "0.8.0"
 name = "stack_walking"
 harness = false
 
+[[bench]]
+name = "heap_live_tracking"
+harness = false
+
 [features]
 default = ["io_profiling"]
 debug_stats = []

profiling/benches/heap_live_tracking.rs

@@ -0,0 +1,91 @@
+use core::cell::Cell;
+use criterion::{black_box, criterion_group, criterion_main, BatchSize, Criterion};
+
+// Compile the production tracker without linking the PHP extension executable.
+#[allow(dead_code)]
+#[path = "../src/profiling/live_heap.rs"]
+mod live_heap;
+use live_heap::LiveHeapTracker;
+
+const BASE_ADDRESS: usize = 0x1000_0000;
+const TRACKED_ALLOCATIONS: usize = 2048;
+const ADDRESS_STRIDE: usize = 64;
+
+fn address(index: usize) -> usize {
+    BASE_ADDRESS + index % TRACKED_ALLOCATIONS * ADDRESS_STRIDE
+}
+
+fn untracked_address(index: usize) -> usize {
+    address(index) + 8
+}
+
+fn populated_tracker() -> LiveHeapTracker<[usize; 4]> {
+    let tracker = LiveHeapTracker::new();
+    for index in 0..TRACKED_ALLOCATIONS {
+        assert!(tracker.track(address(index), [0; 4]));
+    }
+    tracker
+}
+
+fn benchmark(c: &mut Criterion) {
+    let mut group = c.benchmark_group("heap_live_tracking");
+
+    {
+        let tracker = populated_tracker();
+        let next = Cell::new(0);
+        group.bench_function("allocate_tracked", |b| {
+            b.iter_batched(
+                || {
+                    let index = next.get();
+                    next.set(index + 1);
+                    let ptr = untracked_address(index);
+                    let _ = tracker.untrack(ptr);
+                    (ptr, [0; 4])
+                },
+                |(ptr, sample)| black_box(tracker.track(ptr, sample)),
+                BatchSize::PerIteration,
+            )
+        });
+    }
+
+    {
+        let tracker = populated_tracker();
+        let next = Cell::new(0);
+        group.bench_function("free_tracked", |b| {
+            b.iter_batched(
+                || {
+                    let index = next.get();
+                    next.set(index + 1);
+                    let ptr = untracked_address(index);
+                    assert!(tracker.track(ptr, [0; 4]));
+                    ptr
+                },
+                |ptr| black_box(tracker.untrack(ptr)),
+                BatchSize::PerIteration,
+            )
+        });
+    }
+
+    {
+        let tracker = populated_tracker();
+        let next = Cell::new(0);
+        group.bench_function("free_untracked", |b| {
+            b.iter_batched(
+                || {
+                    let index = next.get();
+                    next.set(index + 1);
+                    let ptr = untracked_address(index);
+                    let _ = tracker.untrack(ptr);
+                    ptr
+                },
+                |ptr| black_box(tracker.untrack(ptr)),
+                BatchSize::PerIteration,
+            )
+        });
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, benchmark);
+criterion_main!(benches);

profiling/src/profiling/live_heap.rs

@@ -0,0 +1,68 @@
+use dashmap::DashMap;
+use rustc_hash::FxBuildHasher;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+/// Maximum number of allocations to track for live heap profiling.
+const MAX_SIZE: usize = 4096;
+
+/// Tracks live heap samples by allocation address. FxHasher spreads sequential
+/// ZendMM addresses across DashMap's shards without hashing already-randomized
+/// pointer bytes with SipHash.
+pub(super) struct LiveHeapTracker<T> {
+    allocations: DashMap<usize, T, FxBuildHasher>,
+    count: AtomicUsize,
+}
+
+impl<T> LiveHeapTracker<T> {
+    pub(super) fn new() -> Self {
+        Self {
+            allocations: DashMap::with_hasher(FxBuildHasher),
+            count: AtomicUsize::new(0),
+        }
+    }
+
+    pub(super) fn len(&self) -> usize {
+        self.count.load(Ordering::Relaxed)
+    }
+
+    pub(super) fn clear(&self) {
+        self.allocations.clear();
+        self.count.store(0, Ordering::Relaxed);
+    }
+
+    pub(super) fn track(&self, ptr: usize, sample: T) -> bool {
+        // Best-effort cap: in ZTS the count check and insert still race, so
+        // the map can briefly exceed MAX_SIZE.
+        if self.len() >= MAX_SIZE {
+            return false;
+        }
+
+        if self.allocations.insert(ptr, sample).is_none() {
+            self.count.fetch_add(1, Ordering::Relaxed);
+        }
+        true
+    }
+
+    pub(super) fn untrack(&self, ptr: usize) -> Option<T> {
+        let result = self.allocations.remove(&ptr).map(|(_, sample)| sample);
+        if result.is_some() {
+            self.count.fetch_sub(1, Ordering::Relaxed);
+        }
+        result
+    }
+}
+
+impl<T: Clone> LiveHeapTracker<T> {
+    pub(super) fn snapshot(&self) -> Vec<T> {
+        self.allocations
+            .iter()
+            .map(|entry| entry.value().clone())
+            .collect()
+    }
+}
+
+impl<T> Default for LiveHeapTracker<T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}

profiling/src/profiling/mod.rs

@@ -1,5 +1,6 @@
 mod backtrace;
 mod interrupts;
+mod live_heap;
 mod sample_type_filter;
 pub mod stack_walking;
 mod thread_utils;
@@ -30,20 +31,18 @@ use core::mem::forget;
 use core::{ptr, str};
 use cpu_time::ThreadTime;
 use crossbeam_channel::{Receiver, Sender, TrySendError};
-use dashmap::DashMap;
 use libdd_common::tag::Tag;
 use libdd_profiling::api::{
     Function, Label as ApiLabel, Location, Period, Sample, SampleType as ApiSampleType,
     UpscalingInfo, ValueType as ApiValueType,
 };
 use libdd_profiling::internal::Profile as InternalProfile;
 use log::{debug, info, trace, warn};
-use rustc_hash::FxBuildHasher;
 use std::borrow::Cow;
 use std::collections::HashMap;
 use std::hash::Hash;
 use std::num::NonZeroI64;
-use std::sync::atomic::{AtomicBool, AtomicPtr, AtomicU64, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicPtr, AtomicU64, Ordering};
 use std::sync::{Arc, Barrier, OnceLock};
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
@@ -67,13 +66,6 @@ pub const NO_TIMESTAMP: i64 = 0;
 // magnitude for the capacity.
 const UPLOAD_CHANNEL_CAPACITY: usize = 8;
 
-/// HeapTracker uses FxHasher (rustc-hash) instead of the default SipHash.
-/// FxHasher's multiply-rotate mix fully avalanches bits, spreading sequential
-/// ZendMM bump-allocator addresses evenly across DashMap's 16 shards and
-/// avoiding lock hot-spots under concurrent ZTS workloads.
-/// FxBuildHasher satisfies Clone, which DashMap requires.
-type HeapTracker = DashMap<usize, LiveHeapSample, FxBuildHasher>;
-
 /// The global profiler. Profiler gets made during the first rinit after an
 /// minit, and is destroyed on mshutdown.
 static mut PROFILER: OnceLock<Profiler> = OnceLock::new();
@@ -273,9 +265,7 @@ pub(crate) struct LiveHeapSample {
     pub allocation_size: i64,
 }
 
-/// Maximum number of allocations to track for live heap profiling.
-/// This bounds memory usage. When full, new allocations are not tracked.
-pub(crate) const LIVE_HEAP_TRACKER_MAX_SIZE: usize = 4096;
+use live_heap::LiveHeapTracker;
 
 pub struct Profiler {
     fork_barrier: Arc<Barrier>,
@@ -292,14 +282,8 @@ pub struct Profiler {
     /// through this pointer.
     system_settings: AtomicPtr<SystemSettings>,
 
-    /// Tracks sampled allocations for live heap profiling.
-    /// Maps allocation pointer -> sample data for batched emission at export time.
-    /// Wrapped in Arc to share with TimeCollector for batched sample emission.
-    /// Uses a fast pointer hasher since addresses are already well-distributed.
-    live_heap_tracker: Arc<HeapTracker>,
-    /// Cached entry count for live_heap_tracker. A single Relaxed load replaces
-    /// the 16 shard read-locks that DashMap::len() acquires per sampled allocation.
-    live_heap_tracker_count: Arc<AtomicUsize>,
+    /// Shared with TimeCollector for batched heap-live sample emission.
+    live_heap_tracker: Arc<LiveHeapTracker<LiveHeapSample>>,
 }
 
 struct TimeCollector {
@@ -309,9 +293,7 @@ struct TimeCollector {
     upload_sender: Sender<UploadMessage>,
     upload_period: Duration,
     /// Shared tracker for batched heap-live sample emission at export time.
-    live_heap_tracker: Arc<HeapTracker>,
-    /// See Profiler::live_heap_tracker_count.
-    live_heap_tracker_count: Arc<AtomicUsize>,
+    live_heap_tracker: Arc<LiveHeapTracker<LiveHeapSample>>,
     /// Used to build correctly-positioned sample_values for heap-live samples
     /// without duplicating the type-string → index mapping.
     sample_types_filter: SampleTypeFilter,
@@ -325,7 +307,7 @@ impl TimeCollector {
         profiles: &mut HashMap<Arc<ProfileIndex>, InternalProfile>,
         started_at: &WallTime,
     ) {
-        let tracker_len = self.live_heap_tracker_count.load(Ordering::Relaxed);
+        let tracker_len = self.live_heap_tracker.len();
         if tracker_len == 0 {
             return;
         }
@@ -336,11 +318,7 @@ impl TimeCollector {
         // for the duration of the Arc::clone calls, not for handle_sample_message.
         // This prevents concurrent efree calls on PHP threads from stalling on
         // shards that the TimeCollector is reading during a full export iteration.
-        let snapshot: Vec<LiveHeapSample> = self
-            .live_heap_tracker
-            .iter()
-            .map(|entry| entry.value().clone())
-            .collect();
+        let snapshot = self.live_heap_tracker.snapshot();
 
         for tracked in snapshot {
             let sample_values = self.sample_types_filter.filter(SampleValues {
@@ -837,8 +815,7 @@ impl Profiler {
         let interrupt_manager = Arc::new(InterruptManager::new());
         let (message_sender, message_receiver) = crossbeam_channel::bounded(100);
         let (upload_sender, upload_receiver) = crossbeam_channel::bounded(UPLOAD_CHANNEL_CAPACITY);
-        let live_heap_tracker = Arc::new(DashMap::with_hasher(FxBuildHasher));
-        let live_heap_tracker_count = Arc::new(AtomicUsize::new(0));
+        let live_heap_tracker = Arc::new(LiveHeapTracker::new());
         let sample_types_filter = SampleTypeFilter::new(system_settings);
         let time_collector = TimeCollector {
             fork_barrier: fork_barrier.clone(),
@@ -847,7 +824,6 @@ impl Profiler {
             upload_sender: upload_sender.clone(),
             upload_period: UPLOAD_PERIOD,
             live_heap_tracker: live_heap_tracker.clone(),
-            live_heap_tracker_count: live_heap_tracker_count.clone(),
             sample_types_filter: sample_types_filter.clone(),
         };
 
@@ -888,7 +864,6 @@ impl Profiler {
             sample_types_filter,
             system_settings: AtomicPtr::new(system_settings as *const _ as *mut _),
             live_heap_tracker,
-            live_heap_tracker_count,
         }
     }
 
@@ -945,31 +920,12 @@ impl Profiler {
     /// Track an allocation for live heap profiling.
     /// Returns true if tracked, false if tracking is disabled or limit reached.
     pub(crate) fn track_allocation(&self, ptr: usize, sample: LiveHeapSample) -> bool {
-        // Best-effort cap: in ZTS the count check and insert still race, so
-        // the map can briefly exceed LIVE_HEAP_TRACKER_MAX_SIZE. A single
-        // Relaxed load is equivalent correctness-wise to the former
-        // DashMap::len() but avoids 16 shard read-locks per sampled alloc.
-        if self.live_heap_tracker_count.load(Ordering::Relaxed) >= LIVE_HEAP_TRACKER_MAX_SIZE {
-            return false;
-        }
-
-        let old = self.live_heap_tracker.insert(ptr, sample);
-        if old.is_none() {
-            self.live_heap_tracker_count.fetch_add(1, Ordering::Relaxed);
-        }
-        true
+        self.live_heap_tracker.track(ptr, sample)
     }
 
     /// Untrack an allocation. Returns the sample if it was tracked.
     pub(crate) fn untrack_allocation(&self, ptr: usize) -> Option<LiveHeapSample> {
-        let result = self
-            .live_heap_tracker
-            .remove(&ptr)
-            .map(|(_, sample)| sample);
-        if result.is_some() {
-            self.live_heap_tracker_count.fetch_sub(1, Ordering::Relaxed);
-        }
-        result
+        self.live_heap_tracker.untrack(ptr)
     }
 
     pub fn send_local_root_span_resource(
@@ -1087,7 +1043,6 @@ impl Profiler {
 
             // Clear live heap tracker to avoid stale entries from parent process
             profiler.live_heap_tracker.clear();
-            profiler.live_heap_tracker_count.store(0, Ordering::Relaxed);
 
             // But we're not 100% sure everything is safe to drop, notably the
             // join handles, so we leak the rest.