Skip to content

Refactor thread filter mechanisms #209

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
r1viollet wants to merge 12 commits into from
Closed

Refactor thread filter mechanisms #209

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
3b91530
Bench - Add a benchmark for the thread filter mechanism
r1viollet Mar 21, 2025
913d2b6
Bench - Add a benchmark for the thread filter mechanism
r1viollet Mar 21, 2025
873fcb2
thread filter - reverse lower bits
r1viollet Mar 21, 2025
be0752b
Merge branch 'r1viollet/bench_thread_filter' of github.com:DataDog/ja…
r1viollet Mar 21, 2025
0f92917
thread filter - adjust implementation
r1viollet Apr 3, 2025
258f23e
thread filter
r1viollet Apr 28, 2025
1d4efc0
thread filter - refactor recording implementation
r1viollet Apr 28, 2025
115a60a
thread filter - add a free list
r1viollet Apr 28, 2025
efa094e
Thread filter - add a table with linear probing
r1viollet Jun 25, 2025
1be1c3a
Thread filter bench - Add parameters to run different configurations …
r1viollet Jun 25, 2025
4772169
Merge branch 'main' of github.com:DataDog/java-profiler into r1violle…
r1viollet Jun 25, 2025
60ef060
Minor thread filter adjustement
r1viollet Jun 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ddprof-lib/src/main/cpp/flightRecorder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
44 changes: 43 additions & 1 deletion ddprof-lib/src/main/cpp/threadFilter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
}

Comment on lines +21 to +33

@jbachorik jbachorik Apr 29, 2025

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This looks good. Have you experimented with a custom stride or thread-local random starting offset to improve the contended situation? Basically, to prevent all contending threads starting at 0 and then competing for all subsequent slots until everyone is satisfied - forcing the most unlucky thread to do N checks, N being the number of competing threads.

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())) {
Expand All @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down
4 changes: 4 additions & 0 deletions ddprof-lib/src/main/cpp/threadFilter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
};

Expand Down