src/tracing/agent_perfetto.cc - main - githubmirror/node

src/tracing/agent_perfetto.cc
454 строки15 KB
Chengzhong Wu
src,test: disable trace events tests when perfetto is enabled
02 авг 2026, 17:31
02 авг 2026, 17:31699ac6e
#include "tracing/agent_perfetto.h"

#include <cstdio>
#include <string>
#include "debug_utils-inl.h"
#include "env-inl.h"
#include "node_options.h"
#include "trace_event.h"
#include "trace_event_helper.h"

#include "trace_event_perfetto.h"

namespace node {
namespace tracing {
namespace {

// Perfetto's file writer drains buffers on a fixed period (default 5s) instead
// of continuously; mirror that cadence here.
constexpr uint64_t kReadPeriodMs = 5000;

// Rotate to a new file once the current one reaches this size, so no single
// trace file grows without bound.
constexpr uint64_t kMaxFileSizeBytes = 64 * 1024 * 1024;  // 64 MiB

std::set<std::string> flatten(
    const std::unordered_map<int, std::multiset<std::string>>& map) {
  std::set<std::string> result;
  for (const auto& id_value : map)
    result.insert(id_value.second.begin(), id_value.second.end());
  return result;
}

void perfetto_log_callback(perfetto::LogMessageCallbackArgs args) {
  const char* level_str = "UNKNOWN";
  switch (args.level) {
    case perfetto::base::kLogDebug:
      level_str = "DEBUG";
      break;
    case perfetto::base::kLogInfo:
      level_str = "INFO";
      break;
    case perfetto::base::kLogImportant:
      level_str = "IMPORTANT";
      break;
    case perfetto::base::kLogError:
      level_str = "ERROR";
      break;
  }
  per_process::Debug(DebugCategory::PERFETTO,
                     "[%s] %s:%d: %s\n",
                     level_str,
                     args.filename,
                     args.line,
                     args.message);
}

}  // namespace

// Writes trace chunks to a file, rotating by size. It deliberately uses the
// synchronous uv_fs_* APIs, mirroring Perfetto's internal file writer which
// does blocking writev() on its own service thread.
//
// Synchronous writes are the right fit here, not just simpler:
//   - These writes run on the dedicated tracing loop thread, so blocking on
//     disk stalls neither the application's main thread nor Perfetto's internal
//     thread.
//   - The buffer is drained periodically (see kReadPeriodMs) and each chunk is
//     already contiguous, so there is little I/O to overlap; the thread-pool
//     offload of async uv_fs_* would buy little.
//   - It keeps buffer lifetime, write ordering, flushing, rotation, and
//     shutdown trivial. Async writes would require keeping each buffer alive
//     until its callback, serializing to one in-flight write per fd, and a
//     condition-variable wait for blocking Flush() (see NodeTraceWriter).
class SimpleWriter : public TraceWriter {
 public:
  explicit SimpleWriter(std::string log_file_pattern)
      : log_file_pattern_(std::move(log_file_pattern)) {}

  ~SimpleWriter() override {
    if (fd_ < 0) return;
    uv_fs_t req;
    uv_fs_close(loop_, &req, fd_, nullptr);
    uv_fs_req_cleanup(&req);
  }

  void InitializeOnThread(uv_loop_t* loop) override {
    loop_ = loop;
    OpenNewFileForStreaming();
  }

  // Synchronously writes a chunk, rotating to a new file once it grows too
  // large. See the class comment for why the write is synchronous.
  void AppendTraceChunk(std::vector<char> chunk) override {
    if (fd_ < 0) return;
    uv_buf_t buf =
        uv_buf_init(chunk.data(), static_cast<unsigned int>(chunk.size()));
    uv_fs_t req;
    int written = uv_fs_write(loop_, &req, fd_, &buf, 1, -1, nullptr);
    uv_fs_req_cleanup(&req);
    if (written < 0) return;

    bytes_written_ += written;
    // Each chunk holds only whole trace packets, so rotating on a chunk
    // boundary leaves every file a self-contained, valid trace.
    if (bytes_written_ >= kMaxFileSizeBytes) OpenNewFileForStreaming();
  }

  void Flush(bool blocking) override {
    if (fd_ < 0) return;
    uv_fs_t req;
    uv_fs_fsync(loop_, &req, fd_, nullptr);
    uv_fs_req_cleanup(&req);
  }

 private:
  // Opens the next rotation file, evaluating a JS-style template that accepts
  // ${pid} and ${rotation}, mirroring NodeTraceWriter::OpenNewFileForStreaming.
  void OpenNewFileForStreaming() {
    ++file_num_;
    uv_fs_t req;

    std::string filepath = GetTraceFilePath(log_file_pattern_, file_num_);

    if (fd_ >= 0) {
      uv_fs_close(loop_, &req, fd_, nullptr);
      uv_fs_req_cleanup(&req);
    }

    fd_ = uv_fs_open(loop_,
                     &req,
                     filepath.c_str(),
                     UV_FS_O_CREAT | UV_FS_O_WRONLY | UV_FS_O_TRUNC,
                     0644,
                     nullptr);
    uv_fs_req_cleanup(&req);
    if (fd_ < 0) {
      fprintf(stderr,
              "Could not open trace file %s: %s\n",
              filepath.c_str(),
              uv_strerror(fd_));
      fd_ = -1;
    }
    bytes_written_ = 0;
  }

  std::string log_file_pattern_;
  uv_loop_t* loop_ = nullptr;
  uv_file fd_ = -1;
  int file_num_ = 0;
  uint64_t bytes_written_ = 0;
};

void PerfettoSessionReader::Deleter::operator()(
    PerfettoSessionReader* ptr) const noexcept {
  ptr->tracing_session_->FlushBlocking();
  ptr->Read();
  ptr->tracing_session_->Stop();
}

PerfettoSessionReader::PerfettoSessionReader(
    perfetto::TraceConfig config, std::unique_ptr<TraceWriter> writer)
    : writer_(std::move(writer)) {
  tracing_session_ =
      perfetto::Tracing::NewTrace(perfetto::BackendType::kUnspecifiedBackend);
  tracing_session_->Setup(config);
  tracing_session_->SetOnStopCallback(
      std::bind(&PerfettoSessionReader::SessionStopCallback, this));
  tracing_session_->StartBlocking();
}

PerfettoSessionReader::~PerfettoSessionReader() {}

void PerfettoSessionReader::ChangeTraceConfig(perfetto::TraceConfig config) {
  tracing_session_->ChangeTraceConfig(config);
}

void PerfettoSessionReader::InitializeOnThread(uv_loop_t* loop) {
  tracing_loop_ = loop;
  CHECK_EQ(uv_async_init(tracing_loop_, &read_async_, OnReadAsync), 0);
  read_async_.data = this;
  CHECK_EQ(uv_timer_init(tracing_loop_, &read_timer_), 0);
  read_timer_.data = this;
  writer_->InitializeOnThread(loop);

  // Drain the trace buffer periodically instead of continuously, mirroring
  // Perfetto's file writer which reschedules ReadBuffersIntoFile() every
  // write_period_ms.
  CHECK_EQ(
      uv_timer_start(&read_timer_, OnReadTimer, kReadPeriodMs, kReadPeriodMs),
      0);
}

void PerfettoSessionReader::Read() {
  if (stop_requested_) return;
  // Skip if a previous periodic read is still draining the buffer.
  bool expected = false;
  if (!read_in_progress_.compare_exchange_strong(expected, true)) return;
  tracing_session_->ReadTrace(std::bind(
      &PerfettoSessionReader::ReadTraceCallback, this, std::placeholders::_1));
}

void PerfettoSessionReader::ReadTraceCallback(
    perfetto::TracingSession::ReadTraceCallbackArgs args) {
  // On Perfetto internal thread.
  {
    Mutex::ScopedLock lock(chunks_mutex_);
    if (args.size > 0)
      pending_chunks_.emplace_back(args.data, args.data + args.size);
  }
  // A single ReadTrace() cycle can yield multiple callbacks; the last one has
  // has_more == false, which clears read_in_progress_ so the next timer tick
  // can start a new read.
  read_in_progress_ = args.has_more;
  uv_async_send(&read_async_);
}

void PerfettoSessionReader::SessionStopCallback() {
  stop_requested_ = true;
  uv_async_send(&read_async_);
}

// static
void PerfettoSessionReader::OnReadAsync(uv_async_t* async) {
  PerfettoSessionReader* reader =
      static_cast<PerfettoSessionReader*>(async->data);
  std::list<std::vector<char>> chunks_to_write;
  {
    Mutex::ScopedLock lock(reader->chunks_mutex_);
    std::swap(chunks_to_write, reader->pending_chunks_);
  }

  while (!chunks_to_write.empty()) {
    std::vector<char>& chunk = chunks_to_write.front();
    reader->writer_->AppendTraceChunk(std::move(chunk));
    chunks_to_write.pop_front();
  }

  if (reader->stop_requested_ && reader->handles_pending_close_ == 0) {
    reader->writer_->Flush(true);

    reader->handles_pending_close_ = 2;
    uv_timer_stop(&reader->read_timer_);
    uv_close(reinterpret_cast<uv_handle_t*>(&reader->read_async_),
             OnHandleClose);
    uv_close(reinterpret_cast<uv_handle_t*>(&reader->read_timer_),
             OnHandleClose);
  }
}

// static
void PerfettoSessionReader::OnReadTimer(uv_timer_t* timer) {
  PerfettoSessionReader* reader =
      static_cast<PerfettoSessionReader*>(timer->data);
  reader->Read();
}

// static
void PerfettoSessionReader::OnHandleClose(uv_handle_t* handle) {
  PerfettoSessionReader* reader =
      static_cast<PerfettoSessionReader*>(handle->data);
  if (--reader->handles_pending_close_ == 0) delete reader;
}

PerfettoTracingAgent::PerfettoTracingAgent() {
  CHECK_EQ(uv_loop_init(&tracing_loop_), 0);
  CHECK_EQ(uv_async_init(&tracing_loop_,
                         &initialize_writer_async_,
                         [](uv_async_t* async) {
                           PerfettoTracingAgent* agent = ContainerOf(
                               &PerfettoTracingAgent::initialize_writer_async_,
                               async);
                           agent->InitializeWritersOnThread();
                         }),
           0);

  // Set up the in-process backend that the tracing controller will connect
  // to.
  perfetto::TracingInitArgs init_args;
  init_args.log_message_callback = perfetto_log_callback;
  init_args.backends = perfetto::BackendType::kInProcessBackend;
  perfetto::Tracing::Initialize(init_args);

  node::TrackEvent::Register();
}

void PerfettoTracingAgent::InitializeWritersOnThread() {
  Mutex::ScopedLock lock(initialize_writer_mutex_);
  while (!to_be_initialized_.empty()) {
    auto head = *to_be_initialized_.begin();
    head->InitializeOnThread(&tracing_loop_);
    to_be_initialized_.erase(head);
  }
  initialize_writer_condvar_.Broadcast(lock);

  uv_unref(reinterpret_cast<uv_handle_t*>(&initialize_writer_async_));
}

PerfettoTracingAgent::~PerfettoTracingAgent() {
  StopTracing();

  categories_.clear();
  writers_.clear();

  uv_close(reinterpret_cast<uv_handle_t*>(&initialize_writer_async_), nullptr);
  uv_run(&tracing_loop_, UV_RUN_ONCE);
  CheckedUvLoopClose(&tracing_loop_);
}

void PerfettoTracingAgent::Start() {
  if (started_) return;

  // This thread should be created *after* async handles are created
  // (within NodeTraceWriter and NodeTraceBuffer constructors).
  // Otherwise the thread could shut down prematurely.
  CHECK_EQ(0,
           uv_thread_create(
               &thread_,
               [](void* arg) {
                 uv_thread_setname("TraceEventWorker");
                 PerfettoTracingAgent* agent =
                     static_cast<PerfettoTracingAgent*>(arg);
                 uv_run(&agent->tracing_loop_, UV_RUN_DEFAULT);
               },
               this));

  started_ = true;
}

AgentWriterHandle* PerfettoTracingAgent::GetDefaultWriterHandle() {
  return tracing_file_writer_.has_value() ? &tracing_file_writer_.value()
                                          : nullptr;
}

AgentWriterHandle PerfettoTracingAgent::AddClient(
    const std::set<std::string>& categories,
    std::unique_ptr<TraceWriter> writer,
    enum UseDefaultCategoryMode mode) {
  Start();

  const std::set<std::string>* use_categories = &categories;

  std::set<std::string> categories_with_default;
  if (mode == kUseDefaultCategories) {
    categories_with_default.insert(categories.begin(), categories.end());
    categories_with_default.insert(categories_[kDefaultHandleId].begin(),
                                   categories_[kDefaultHandleId].end());
    use_categories = &categories_with_default;
  }

  int id = next_writer_id_++;
  PerfettoSessionReader::Ptr reader = PerfettoSessionReader::Create(
      CreateTraceConfig({use_categories->begin(), use_categories->end()}),
      std::move(writer));

  auto* raw = reader.get();
  writers_[id] = std::move(reader);
  categories_[id] = {use_categories->begin(), use_categories->end()};

  {
    Mutex::ScopedLock lock(initialize_writer_mutex_);
    to_be_initialized_.insert(raw);
    uv_async_send(&initialize_writer_async_);
    while (to_be_initialized_.count(raw) > 0)
      initialize_writer_condvar_.Wait(lock);
  }

  return AgentWriterHandle(this, id);
}

void PerfettoTracingAgent::StartTracing(const std::string& categories) {
  if (tracing_file_writer_.has_value()) return;

  using std::operator""sv;
  auto parts = std::views::split(categories, ","sv);

  std::set<std::string> categories_set;
  for (const auto& s : parts) {
    categories_set.emplace(std::string(s.data(), s.size()));
  }

  tracing_file_writer_ =
      AddClient(categories_set,
                std::make_unique<SimpleWriter>(
                    per_process::cli_options->trace_event_file_pattern),
                kUseDefaultCategories);
}

void PerfettoTracingAgent::StopTracing() {
  if (!started_) return;
  // Perform final Flush on TraceBuffer. We don't want the tracing controller
  // to flush the buffer again on destruction of the V8::Platform.
  writers_.clear();
  started_ = false;

  // Thread should finish when the tracing loop is stopped.
  uv_thread_join(&thread_);
}

void PerfettoTracingAgent::Disconnect(int client) {
  if (client == kDefaultHandleId) return;
  {
    Mutex::ScopedLock lock(initialize_writer_mutex_);
    to_be_initialized_.erase(writers_[client].get());
  }

  writers_.erase(client);
  categories_.erase(client);
}

void PerfettoTracingAgent::Enable(int id,
                                  const std::set<std::string>& categories) {
  if (categories.empty()) return;

  categories_[id].insert(categories.begin(), categories.end());

  writers_[id]->ChangeTraceConfig(CreateTraceConfig(categories_[id]));
}

void PerfettoTracingAgent::Disable(int id,
                                   const std::set<std::string>& categories) {
  std::multiset<std::string>& writer_categories = categories_[id];
  for (const std::string& category : categories) {
    auto it = writer_categories.find(category);
    if (it != writer_categories.end()) writer_categories.erase(it);
  }

  writers_[id]->ChangeTraceConfig(CreateTraceConfig(writer_categories));
}

std::string PerfettoTracingAgent::GetEnabledCategories() const {
  std::string categories;
  for (const std::string& category : flatten(categories_)) {
    if (!categories.empty()) categories += ',';
    categories += category;
  }
  return categories;
}

perfetto::TraceConfig PerfettoTracingAgent::CreateTraceConfig(
    std::multiset<std::string> categories) const {
  perfetto::TraceConfig perfetto_trace_config;
  perfetto_trace_config.add_buffers()->set_size_kb(4096);
  auto ds_config = perfetto_trace_config.add_data_sources()->mutable_config();
  ds_config->set_name("track_event");
  perfetto::protos::gen::TrackEventConfig te_config;
  te_config.add_disabled_categories("*");
  for (const auto& category : categories)
    te_config.add_enabled_categories(category);
  ds_config->set_track_event_config_raw(te_config.SerializeAsString());
  return perfetto_trace_config;
}

}  // namespace tracing
}  // namespace node