Skip to content

wall profiler: use thread-local for the active profiler #322

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

Draft
szegedi wants to merge 1 commit into
base: main
Choose a base branch
from
+57 −135
Draft

wall profiler: use thread-local for the active profiler #322

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
190 changes: 56 additions & 134 deletions bindings/profilers/wall.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include <memory>
#include <mutex>
#include <type_traits>
#include <unordered_set>
#include <vector>

#include "map-get.hh"
Expand Down Expand Up @@ -179,134 +180,46 @@ int detectV8Bug(const v8::CpuProfile* profile) {
return 0;
}

class ProtectedProfilerMap {
public:
WallProfiler* GetProfiler(const Isolate* isolate) const {
// Prevent updates to profiler map by atomically setting g_profilers to null
auto prof_map = profilers_.exchange(nullptr, std::memory_order_acq_rel);
if (!prof_map) {
return nullptr;
}
auto prof_it = prof_map->find(isolate);
WallProfiler* profiler = nullptr;
if (prof_it != prof_map->end()) {
profiler = prof_it->second;
}
// Allow updates
profilers_.store(prof_map, std::memory_order_release);
return profiler;
}

WallProfiler* RemoveProfilerForIsolate(const v8::Isolate* isolate) {
return UpdateProfilers([isolate](auto map) {
auto it = map->find(isolate);
if (it != map->end()) {
auto profiler = it->second;
map->erase(it);
return profiler;
}
return static_cast<WallProfiler*>(nullptr);
});
}

bool RemoveProfiler(const v8::Isolate* isolate, WallProfiler* profiler) {
return UpdateProfilers([isolate, profiler, this](auto map) {
terminatedWorkersCpu_ += profiler->GetAndResetThreadCpu();
// Per-thread active profiler, used by the SIGPROF handler. In Node's threading
// model each V8 isolate is pinned to a single thread, so a thread-local pointer
// is equivalent to "the profiler for the isolate that received this signal".
thread_local WallProfiler* t_active_profiler = nullptr;

if (isolate != nullptr) {
auto it = map->find(isolate);
if (it != map->end() && it->second == profiler) {
map->erase(it);
return true;
}
} else {
auto it = std::find_if(map->begin(), map->end(), [profiler](auto& x) {
return x.second == profiler;
});
if (it != map->end()) {
map->erase(it);
return true;
}
}
return false;
});
// Registry of all live profilers across threads. Used only by the main thread
// to gather CPU consumed by JS worker threads while building a profile.
class ActiveProfilers {
public:
void Add(WallProfiler* profiler) {
std::lock_guard<std::mutex> lock(mutex_);
profilers_.insert(profiler);
}

bool AddProfiler(const v8::Isolate* isolate, WallProfiler* profiler) {
return UpdateProfilers([isolate, profiler](auto map) {
return map->emplace(isolate, profiler).second;
});
bool Remove(WallProfiler* profiler) {
std::lock_guard<std::mutex> lock(mutex_);
auto it = profilers_.find(profiler);
if (it == profilers_.end()) return false;
terminatedWorkersCpu_ += profiler->GetAndResetThreadCpu();
profilers_.erase(it);
return true;
}

ThreadCpuClock::duration GatherTotalWorkerCpuAndReset() {
std::lock_guard<std::mutex> lock(update_mutex_);

// Retrieve CPU of workers that have terminated during the last period
std::lock_guard<std::mutex> lock(mutex_);
ThreadCpuClock::duration totalWorkerCpu = terminatedWorkersCpu_;

// Reset terminated workers cpu to 0
terminatedWorkersCpu_ = ThreadCpuClock::duration::zero();

if (!init_) {
return totalWorkerCpu;
}

auto currProfilers = profilers_.load(std::memory_order_acquire);
// Wait until sighandler is done using the map
while (!currProfilers) {
currProfilers = profilers_.load(std::memory_order_relaxed);
}

// Gather CPU of workers that are still running
for (auto& profiler : *currProfilers) {
totalWorkerCpu += profiler.second->GetAndResetThreadCpu();
for (auto* profiler : profilers_) {
totalWorkerCpu += profiler->GetAndResetThreadCpu();
}

return totalWorkerCpu;
}

private:
using ProfilerMap = std::unordered_map<const Isolate*, WallProfiler*>;

template <typename F>
std::invoke_result_t<F, ProfilerMap*> UpdateProfilers(F updateFn) {
// use mutex to prevent two isolates of updating profilers concurrently
std::lock_guard<std::mutex> lock(update_mutex_);

if (!init_) {
profilers_.store(new ProfilerMap(), std::memory_order_release);
init_ = true;
}

auto currProfilers = profilers_.load(std::memory_order_acquire);
// Wait until sighandler is done using the map
while (!currProfilers) {
currProfilers = profilers_.load(std::memory_order_relaxed);
}
auto newProfilers = new ProfilerMap(*currProfilers);
auto res = updateFn(newProfilers);
// Wait until sighandler is done using the map before installing a new map.
// The value in profilers is either nullptr or currProfilers.
for (;;) {
ProfilerMap* currProfilers2 = currProfilers;
if (profilers_.compare_exchange_weak(
currProfilers2, newProfilers, std::memory_order_acq_rel)) {
break;
}
}
delete currProfilers;
return res;
}

mutable std::atomic<ProfilerMap*> profilers_;
std::mutex update_mutex_;
bool init_ = false;
std::chrono::nanoseconds terminatedWorkersCpu_{};
std::mutex mutex_;
std::unordered_set<WallProfiler*> profilers_;
ThreadCpuClock::duration terminatedWorkersCpu_{};
};

using ProfilerMap = std::unordered_map<Isolate*, WallProfiler*>;

static ProtectedProfilerMap g_profilers;
static ActiveProfilers g_profilers;

namespace {

Expand Down Expand Up @@ -380,15 +293,9 @@ void SignalHandler::HandleProfilerSignal(int sig,
if (!old_handler) {
return;
}
auto isolate = Isolate::GetCurrent();
if (!isolate) {
return;
}
WallProfiler* prof = g_profilers.GetProfiler(isolate);

WallProfiler* prof = t_active_profiler;
if (!prof) {
// no profiler found for current isolate, just pass the signal to old
// handler
// no profiler active on this thread, just pass the signal to old handler
old_handler(sig, info, context);
return;
}
Expand All @@ -408,6 +315,10 @@ void SignalHandler::HandleProfilerSignal(int sig,
auto time_from = Now();
old_handler(sig, info, context);
auto time_to = Now();
auto isolate = Isolate::GetCurrent();
if (!isolate) {
return;
}
prof->PushContext(time_from, time_to, cpu_time, isolate);
}
#else
Expand Down Expand Up @@ -456,9 +367,12 @@ static int64_t GetV8ToEpochOffset() {
}

void WallProfiler::CleanupHook(void* data) {
auto isolate = static_cast<Isolate*>(data);
auto prof = g_profilers.RemoveProfilerForIsolate(isolate);
// Environment cleanup hooks run on the isolate's own thread, so the
// thread-local pointer for this thread (if any) refers to the profiler that
// belongs to this isolate.
auto prof = t_active_profiler;
if (prof) {
auto isolate = static_cast<Isolate*>(data);
prof->Cleanup(isolate);
delete prof;
}
Expand All @@ -472,7 +386,7 @@ void WallProfiler::Cleanup(Isolate* isolate) {
if (interceptSignal()) {
SignalHandler::DecreaseUseCount();
}
Dispose(isolate, false);
Dispose(isolate);
}
}

Expand Down Expand Up @@ -669,6 +583,14 @@ WallProfiler::WallProfiler(std::chrono::microseconds samplingPeriod,
}

WallProfiler::~WallProfiler() {
// Defensive: under normal use Dispose() has already cleared both of these,
// but if a profiler is destroyed without going through Dispose we still must
// not leave dangling pointers.
if (t_active_profiler == this) {
t_active_profiler = nullptr;
}
g_profilers.Remove(this);

// Delete all live contexts
for (auto ptr : liveContextPtrs_) {
ptr->Detach(); // so it doesn't invalidate our iterator
Expand All @@ -677,13 +599,14 @@ WallProfiler::~WallProfiler() {
liveContextPtrs_.clear();
}

void WallProfiler::Dispose(Isolate* isolate, bool removeFromMap) {
void WallProfiler::Dispose(Isolate* isolate) {
if (cpuProfiler_ != nullptr) {
cpuProfiler_->Dispose();
cpuProfiler_ = nullptr;

if (removeFromMap) {
g_profilers.RemoveProfiler(isolate, this);
g_profilers.Remove(this);
if (t_active_profiler == this) {
t_active_profiler = nullptr;
}

if (collectAsyncId_ || useCPED()) {
Expand Down Expand Up @@ -1065,7 +988,7 @@ Result WallProfiler::StopCore(bool restart, ProfileBuilder&& buildProfile) {
v8_profile->Delete();

if (!restart) {
Dispose(v8::Isolate::GetCurrent(), true);
Dispose(v8::Isolate::GetCurrent());
} else if (workaroundV8Bug_) {
waitForSignal(callCount + 1);
std::atomic_signal_fence(std::memory_order_release);
Expand Down Expand Up @@ -1151,14 +1074,13 @@ NAN_MODULE_INIT(WallProfiler::Init) {

v8::CpuProfiler* WallProfiler::CreateV8CpuProfiler() {
if (cpuProfiler_ == nullptr) {
v8::Isolate* isolate = v8::Isolate::GetCurrent();

bool inserted = g_profilers.AddProfiler(isolate, this);

if (!inserted) {
// refuse to create a new profiler if one is already active
if (t_active_profiler != nullptr) {
// refuse to create a new profiler if one is already active on this thread
return nullptr;
}
v8::Isolate* isolate = v8::Isolate::GetCurrent();
g_profilers.Add(this);
t_active_profiler = this;
cpuProfiler_ = v8::CpuProfiler::New(isolate);
cpuProfiler_->SetSamplingInterval(
std::chrono::microseconds(samplingPeriod_).count());
Expand Down
2 changes: 1 addition & 1 deletion bindings/profilers/wall.hh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ class WallProfiler : public Nan::ObjectWrap {
ContextBuffer contexts_;

~WallProfiler();
void Dispose(v8::Isolate* isolate, bool removeFromMap);
void Dispose(v8::Isolate* isolate);

// A new CPU profiler object will be created each time profiling is started
// to work around https://bugs.chromium.org/p/v8/issues/detail?id=11051.
Expand Down
Loading