Fix calltrace_storage counters accumulating unboundedly across rotations (#427)

Author: jbachorik

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

@@ -102,9 +102,48 @@ 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.
+  // 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.
+  const size_t header_size = sizeof(CallTrace) - sizeof(ASGCT_CallFrame);
+  long long freed_bytes = 0;
+  long long freed_traces = 0;
+  size_t estimated_entries = 0;
+  for (LongHashTable *t = _table; t != nullptr; t = t->prev()) {
+    estimated_entries += t->size();
+  }
+  std::unordered_set<CallTrace*> seen;
+  seen.reserve(estimated_entries);
+  for (LongHashTable *t = _table; t != nullptr; t = t->prev()) {
+    u64 *keys = t->keys();
+    CallTraceSample *values = t->values();
+    u32 capacity = t->capacity();
+    for (u32 slot = 0; slot < capacity; slot++) {
+      if (keys[slot] != 0) {
+        CallTrace *trace = values[slot].acquireTrace();
+        if (trace != nullptr && trace != CallTraceSample::PREPARING) {
+          if (seen.insert(trace).second) {
+            freed_bytes += header_size + trace->num_frames * sizeof(ASGCT_CallFrame);
+            freed_traces++;
+          }
+        }
+      }
+    }
+  }
+  Counters::increment(CALLTRACE_STORAGE_BYTES, -freed_bytes);
+  Counters::increment(CALLTRACE_STORAGE_TRACES, -freed_traces);
+#endif // COUNTERS
+}
+
 ChunkList CallTraceHashTable::clearTableOnly() {
   // Wait for all refcount guards to clear before detaching chunks
   RefCountGuard::waitForAllRefCountsToClear();
+  decrementCounters();
 
   // Clear previous chain pointers to prevent traversal during deallocation
   for (LongHashTable *table = _table; table != nullptr; table = table->prev()) {
@@ -424,7 +463,7 @@ void CallTraceHashTable::putWithExistingId(CallTrace* source_trace, u64 weight)
         table->values()[slot].setTrace(copied_trace);
         Counters::increment(CALLTRACE_STORAGE_BYTES, total_size);
         Counters::increment(CALLTRACE_STORAGE_TRACES);
-        
+
         // Increment table size
         u32 new_size = table->incSize();
         probe.updateCapacity(new_size);

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

@@ -72,6 +72,7 @@ class CallTraceHashTable {
   CallTrace *storeCallTrace(int num_frames, ASGCT_CallFrame *frames,
                             bool truncated, u64 trace_id);
   CallTrace *findCallTrace(LongHashTable *table, u64 hash);
+  void decrementCounters();
 
 
 public:

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

@@ -453,7 +453,9 @@ u64 Profiler::recordJVMTISample(u64 counter, int tid, jthread thread, jint event
   }
   u64 call_trace_id = 0;
   if (!_omit_stacktraces) {
+#ifdef COUNTERS
     u64 startTime = TSC::ticks();
+#endif // COUNTERS
     ASGCT_CallFrame *frames = _calltrace_buffer[lock_index]->_asgct_frames;
     jvmtiFrameInfo *jvmti_frames = _calltrace_buffer[lock_index]->_jvmti_frames;
 
@@ -473,10 +475,12 @@ u64 Profiler::recordJVMTISample(u64 counter, int tid, jthread thread, jint event
     }
 
     call_trace_id = _call_trace_storage.put(num_frames, frames, false, counter);
+#ifdef COUNTERS
     u64 duration = TSC::ticks() - startTime;
     if (duration > 0) {
-      Counters::increment(UNWINDING_TIME_JVMTI, duration); // increment the JVMTI specific counter
+      Counters::increment(UNWINDING_TIME_JVMTI, duration);
     }
+#endif // COUNTERS
   }
   if (!deferred) {
     _jfr.recordEvent(lock_index, tid, call_trace_id, event_type, event);
@@ -529,7 +533,9 @@ void Profiler::recordSample(void *ucontext, u64 counter, int tid,
   // call_trace_id determined to be reusable at a higher level
 
   if (!_omit_stacktraces && call_trace_id == 0) {
+#ifdef COUNTERS
     u64 startTime = TSC::ticks();
+#endif // COUNTERS
     ASGCT_CallFrame *frames = _calltrace_buffer[lock_index]->_asgct_frames;
 
     int num_frames = 0;
@@ -558,10 +564,12 @@ void Profiler::recordSample(void *ucontext, u64 counter, int tid,
     if (thread != nullptr) {
       thread->recordCallTraceId(call_trace_id);
     }
+#ifdef COUNTERS
     u64 duration = TSC::ticks() - startTime;
     if (duration > 0) {
-      Counters::increment(UNWINDING_TIME_ASYNC, duration); // increment the async specific counter
+      Counters::increment(UNWINDING_TIME_ASYNC, duration);
     }
+#endif // COUNTERS
   }
   _jfr.recordEvent(lock_index, tid, call_trace_id, event_type, event);
 

ddprof-test/src/test/java/com/datadoghq/profiler/AbstractProfilerTest.java

@@ -475,4 +475,26 @@ protected void verifyStackTraces(Path recording, String eventType, String... pat
     assertNotEquals(0, cumulatedEvents, "no events found for " + eventType);
     assertTrue(unmatched.isEmpty(), "couldn't find " + eventType + " with " + unmatched);
   }
+
+  /**
+   * Returns the value of a named counter from {@code datadog.ProfilerCounter} events in the JFR
+   * recording. These events are written before the final cleanup ({@code processTraces}), so they
+   * capture the pre-cleanup state.
+   *
+   * @return the counter value, or -1 if no matching event is found
+   */
+  public long getRecordedCounterValue(String counterName) {
+    IItemCollection events = verifyEvents("datadog.ProfilerCounter", false);
+    for (IItemIterable iterable : events) {
+      IMemberAccessor<String, IItem> nameAccessor = NAME.getAccessor(iterable.getType());
+      IMemberAccessor<IQuantity, IItem> countAccessor = COUNT.getAccessor(iterable.getType());
+      if (nameAccessor == null || countAccessor == null) continue;
+      for (IItem item : iterable) {
+        if (counterName.equals(nameAccessor.getMember(item))) {
+          return countAccessor.getMember(item).longValue();
+        }
+      }
+    }
+    return -1;
+  }
 }

ddprof-test/src/test/java/com/datadoghq/profiler/cpu/ContextCpuTest.java

@@ -106,11 +106,14 @@ public void test(@CStack String cstack) throws ExecutionException, InterruptedEx
             assertInRange(method2Weight / (double) totalWeight, 0.1, 0.6);
             assertInRange(method3Weight / (double) totalWeight, 0.05, 0.6);
         }
+        // The recording captures counter values before the final cleanup (before processTraces
+        // runs and frees all traces). Verify the recording contains meaningful data.
+        assertInRange(getRecordedCounterValue("calltrace_storage_traces"), 1, 100);
+        assertInRange(getRecordedCounterValue("calltrace_storage_bytes"), 1024, 8 * 1024 * 1024);
+        // live counters are 0 after stop (all traces freed - correct, non-leaking behaviour)
         Map<String, Long> debugCounters = profiler.getDebugCounters();
-        // these are here to verify these counters produce reasonable values so they can be used for memory leak detection
-        assertInRange(debugCounters.get("calltrace_storage_traces"), 1, 100);
-        assertInRange(debugCounters.get("calltrace_storage_bytes"), 1024, 8 * 1024 * 1024);
-        // this allocator is only used for calltrace storage and eagerly allocates chunks of 8MiB
+        assertEquals(0, debugCounters.get("calltrace_storage_traces"));
+        assertEquals(0, debugCounters.get("calltrace_storage_bytes"));
         assertEquals(0, debugCounters.get("linear_allocator_bytes"));
         assertEquals(0, debugCounters.get("linear_allocator_chunks"));
     }

ddprof-test/src/test/java/com/datadoghq/profiler/wallclock/BaseContextWallClockTest.java

@@ -196,11 +196,14 @@ void test(AbstractProfilerTest test, boolean assertContext, String cstack) throw
             assertWeight("method2Impl", totalWeight, method2Weight, 0.33, allowedError);
             // method3 has as much self time as method1, and should account for half the executor's thread's time
             assertWeight("method3Impl", totalWeight, method3Weight, 0.33, allowedError);
+            // The recording captures counter values before the final cleanup (before processTraces
+            // runs and frees all traces). Verify the recording contains meaningful data.
+            assertInRange(test.getRecordedCounterValue("calltrace_storage_traces"), 1, 100);
+            assertInRange(test.getRecordedCounterValue("calltrace_storage_bytes"), 1024, 8 * 1024 * 1024);
+            // live counters are 0 after stop (all traces freed - correct, non-leaking behaviour)
             Map<String, Long> debugCounters = profiler.getDebugCounters();
-            // these are here to verify these counters produce reasonable values so they can be used for memory leak detection
-            assertInRange(debugCounters.get("calltrace_storage_traces"), 1, 100);
-            assertInRange(debugCounters.get("calltrace_storage_bytes"), 1024, 8 * 1024 * 1024);
-            // this allocator is only used for calltrace storage and eagerly allocates chunks of 8MiB
+            assertEquals(0, debugCounters.get("calltrace_storage_traces"));
+            assertEquals(0, debugCounters.get("calltrace_storage_bytes"));
             assertEquals(0, debugCounters.get("linear_allocator_bytes"));
             assertEquals(0, debugCounters.get("linear_allocator_chunks"));
             assertInRange(debugCounters.get("thread_ids_count"), 1, 100);