thread filter - add a free list

Author: r1viollet

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

@@ -128,6 +128,8 @@ class Recording {
   static char *_java_command;
 
   RecordingBuffer _buf[CONCURRENCY_LEVEL];
+  // we have several sets to avoid lock contention
+  // we have a second dimension to allow a switch in the active set
   std::unordered_set<int> _thread_ids[CONCURRENCY_LEVEL][2];
   std::atomic<int> _active_index{0};  // 0 or 1 globally
 

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

@@ -17,12 +17,24 @@ ThreadFilter::~ThreadFilter() {
     std::unique_lock<std::mutex> lock(_slot_mutex);
     _slots.clear();
 }
-
 ThreadFilter::SlotID ThreadFilter::registerThread() {
+    int top = _free_list_top.load(std::memory_order_acquire);
+    for (int i = 0; i < top; ++i) {
+        int value = _free_list[i].load(std::memory_order_relaxed);
+        if (value >= 0) {
+            int expected = value;
+            if (_free_list[i].compare_exchange_strong(expected, -1, std::memory_order_acq_rel)) {
+                return value; // Successfully claimed a free slot
+            }
+            // If CAS fails, someone else claimed it, continue scanning
+        }
+    }
+
     SlotID index = _next_index.fetch_add(1, std::memory_order_relaxed);
     if (index >= kMaxThreads) {
         return -1;
     }
+
     const int outer_idx = index >> kChunkShift;
     {
         if (outer_idx < static_cast<int>(_slots.size())) {
@@ -46,6 +58,10 @@ void ThreadFilter::clear() {
             slot.value.store(-1, std::memory_order_relaxed);
         }
     }
+    for (int i = 0; i < kFreeListSize; ++i) {
+        _free_list[i].store(-1, std::memory_order_relaxed);
+    }
+    _free_list_top.store(0, std::memory_order_relaxed);
 }
 
 bool ThreadFilter::accept(SlotID slot_id) const {
@@ -73,9 +89,35 @@ void ThreadFilter::remove(SlotID slot_id) {
 
 void ThreadFilter::unregisterThread(SlotID slot_id) {
     if (slot_id < 0) return;
+
     int outer_idx = slot_id >> kChunkShift;
     int inner_idx = slot_id & kChunkMask;
     _slots[outer_idx][inner_idx].value.store(-1, std::memory_order_relaxed);
+
+    constexpr int try_limit = 16;
+
+    int top = _free_list_top.load(std::memory_order_acquire);
+    int limit = top < kFreeListSize ? top : kFreeListSize;
+    int tries = 0;
+
+    for (int i = 0; i < limit && tries < try_limit; ++i) {
+        int value = _free_list[i].load(std::memory_order_relaxed);
+        if (value == -1) {
+            int expected = -1;
+            if (_free_list[i].compare_exchange_strong(expected, slot_id, std::memory_order_acq_rel)) {
+                return; // Successfully claimed empty spot
+            }
+            ++tries; // Only count actual CAS attempts
+        }
+    }
+
+    // Fallback: append if no empty slot found
+    int pos = _free_list_top.fetch_add(1, std::memory_order_acq_rel);
+    if (pos < kFreeListSize) {
+        _free_list[pos].store(slot_id, std::memory_order_release);
+    } else {
+        // Free list overflow: ignore
+    }
 }
 
 void ThreadFilter::collect(std::vector<int>& tids) const {

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

@@ -43,6 +43,10 @@ class ThreadFilter {
     std::vector<std::array<Slot, kChunkSize>> _slots;
     std::atomic<SlotID> _next_index;
 
+    static constexpr int kFreeListSize = 128;
+    std::atomic<SlotID> _free_list[kFreeListSize];
+    std::atomic<int> _free_list_top;  // Points to next free slot
+
     mutable std::mutex _slot_mutex;
 };