[etw tracing] Refactor EtwSystemDataSource for testing

This is a refactor of EtwSystemDataSource with no intended behavior
change (aside from a small bug fix or two). The motivation is to make
the ETW consumer easier to test and easier to expand with more
functionality in subsequent CLs.

In particular:

- EtwSystemDataSource::Consumer is now EtwConsumer.
- The ProcessEventRecord implementation now dispatches to per-provider
  event handling functions based on the provider guid.
- LostEvent records now emit DLOG messages for diagnostic purposes.
- Perfetto events are only generated when there is valid data to give
  them. Previously, an EtwTraceEvent was created for every ETW event
  originating from the Thread provider, but only populated for valid
  CSwitch events.
- EtwConsumerTest (was EtwSystemDataSource) exercises more of the ETW
  event processing code.

Bug: 41497361
Change-Id: I544533aed9c7974f3ee8bc77bf5d3d384a93f9de
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/6084478
Reviewed-by: Dirk Pranke <dpranke@google.com>
Reviewed-by: Etienne Pierre-Doray <etiennep@chromium.org>
Commit-Queue: Greg Thompson <grt@chromium.org>
Cr-Commit-Position: refs/heads/main@{#1394798}

PRESUBMIT.py

@@ -1086,6 +1086,7 @@ _BANNED_CPP_FUNCTIONS: Sequence[BanRule] = (
             'services/on_device_model/ml/on_device_model_executor\.h',
             # Required to interop with interfaces from the third-party perfetto
             # library.
+            'components/tracing/common/etw_consumer_win_unittest\.cc',
             'services/tracing/public/cpp/perfetto/custom_event_recorder\.cc',
             'services/tracing/public/cpp/perfetto/perfetto_traced_process\.cc',
             'services/tracing/public/cpp/perfetto/perfetto_traced_process\.h',

components/tracing/BUILD.gn

@@ -45,6 +45,8 @@ component("startup_tracing") {
 
   if (is_win) {
     sources += [
+      "common/etw_consumer_win.cc",
+      "common/etw_consumer_win.h",
       "common/etw_export_win.cc",
       "common/etw_export_win.h",
       "common/etw_system_data_source_win.cc",
@@ -119,7 +121,8 @@ source_set("unit_tests") {
   data = [ "$root_gen_dir/third_party/perfetto/protos/perfetto/config/chrome/scenario_config.descriptor" ]
 
   if (is_win) {
-    sources += [ "common/etw_system_data_source_win_unittest.cc" ]
+    sources += [ "common/etw_consumer_win_unittest.cc" ]
+    deps += [ "//third_party/perfetto:libperfetto" ]
   }
   deps += [
     ":background_tracing_utils",

components/tracing/common/etw_consumer_win.cc

@@ -0,0 +1,231 @@
+// Copyright 2024 The Chromium Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "components/tracing/common/etw_consumer_win.h"
+
+#include <windows.h>
+
+#include <functional>
+#include <tuple>
+#include <utility>
+
+#include "base/check_op.h"
+#include "base/compiler_specific.h"
+#include "base/containers/buffer_iterator.h"
+#include "base/containers/fixed_flat_map.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/threading/scoped_blocking_call.h"
+#include "base/time/time.h"
+#include "third_party/perfetto/protos/perfetto/trace/etw/etw.pbzero.h"
+#include "third_party/perfetto/protos/perfetto/trace/etw/etw_event.pbzero.h"
+#include "third_party/perfetto/protos/perfetto/trace/etw/etw_event_bundle.pbzero.h"
+#include "third_party/perfetto/protos/perfetto/trace/trace_packet.pbzero.h"
+
+namespace tracing {
+
+namespace {
+
+// A function object that returns true if one GUID is "less than" another.
+struct IsGuidLess {
+  constexpr bool operator()(const GUID& a, const GUID& b) const {
+    if (auto result = std::tie(a.Data1, a.Data2, a.Data3) <=>
+                      std::tie(b.Data1, b.Data2, b.Data3);
+        result < 0) {
+      return true;
+    } else if (result > 0) {
+      return false;
+    }
+    return base::span(a.Data4) < base::span(b.Data4);
+  }
+};
+
+}  // namespace
+
+EtwConsumer::EtwConsumer(
+    std::unique_ptr<perfetto::TraceWriterBase> trace_writer)
+    : trace_writer_(std::move(trace_writer)) {
+  DETACH_FROM_SEQUENCE(sequence_checker_);
+}
+
+EtwConsumer::~EtwConsumer() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+}
+
+void EtwConsumer::ConsumeEvents() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+  base::ScopedBlockingCall scoped_blocking(FROM_HERE,
+                                           base::BlockingType::MAY_BLOCK);
+  Consume();
+}
+
+// static
+void EtwConsumer::ProcessEventRecord(EVENT_RECORD* event_record) {
+  // https://learn.microsoft.com/en-us/windows/win32/etw/nt-kernel-logger-constants
+  // LostEventGuid, 6a399ae0-4bc6-4de9-870b-3657f8947e7e
+  static constexpr GUID kLostEventGuid = {
+      0x6a399ae0,
+      0x4bc6,
+      0x4de9,
+      {0x87, 0x0b, 0x36, 0x57, 0xf8, 0x94, 0x7e, 0x7e}};
+  // ThreadGuid, 3d6fa8d1-fe05-11d0-9dda-00c04fd7ba7c
+  static constexpr GUID kThreadGuid = {
+      0x3d6fa8d1,
+      0xfe05,
+      0x11d0,
+      {0x9d, 0xda, 0x00, 0xc0, 0x4f, 0xd7, 0xba, 0x7c}};
+
+  // A mapping of provider GUIDs to handler member functions.
+  static constexpr auto kGuidToProvider =
+      base::MakeFixedFlatMap<std::reference_wrapper<const GUID>,
+                             EventHandlerFunction, IsGuidLess>(
+          {{kThreadGuid, &EtwConsumer::HandleThreadEvent},
+           {kLostEventGuid, &EtwConsumer::HandleLostEvent}});
+
+  auto* const self = reinterpret_cast<EtwConsumer*>(event_record->UserContext);
+  DCHECK_CALLED_ON_VALID_SEQUENCE(self->sequence_checker_);
+
+  if (auto iter = kGuidToProvider.find(event_record->EventHeader.ProviderId);
+      iter != kGuidToProvider.end()) {
+    (self->*iter->second)(
+        event_record->EventHeader, event_record->BufferContext,
+        // SAFETY: The pointer and length originate from ETW.
+        UNSAFE_BUFFERS({static_cast<uint8_t*>(event_record->UserData),
+                        event_record->UserDataLength}));
+  }
+  // The following providers are always enabled. There is not yet a need to
+  // handle any events originating from them:
+  // - EventTraceGuid: 68fdd900-4a3e-11d1-84f4-0000f80464e3
+  //   - Opcode 32: EndExtension / Event Trace Header Extension
+  //   - Opcode 5: Extension / Event Trace Header Extension
+  //   - Opcode 8: RDComplete / Event Trace Rundown Complete
+  // - EventTraceConfigGuid: 01853a65-418f-4f36-aefc-dc0f1d2fd235
+  //   - Various hardware configuration events.
+}
+
+// static
+bool EtwConsumer::ProcessBuffer(EVENT_TRACE_LOGFILE* buffer) {
+  auto* const self = reinterpret_cast<EtwConsumer*>(buffer->Context);
+  DCHECK_CALLED_ON_VALID_SEQUENCE(self->sequence_checker_);
+  self->etw_events_ = nullptr;
+  // Release the handle to finalize the previous message.
+  self->packet_handle_ = {};
+  return true;  // Continue processing events.
+}
+
+void EtwConsumer::HandleThreadEvent(const EVENT_HEADER& header,
+                                    const ETW_BUFFER_CONTEXT& buffer_context,
+                                    base::span<const uint8_t> packet_data) {
+  if (header.EventDescriptor.Opcode == 36) {  // CSwitch
+    if (!DecodeCSwitchEvent(header, buffer_context, packet_data)) {
+      DLOG(ERROR) << "Error decoding CSwitch Event";
+    }
+  }
+}
+
+void EtwConsumer::HandleLostEvent(const EVENT_HEADER& header,
+                                  const ETW_BUFFER_CONTEXT& buffer_context,
+                                  base::span<const uint8_t> packet_data) {
+  switch (header.EventDescriptor.Opcode) {
+    case 32:  // RTLostEvent
+      // TODO: Emit a Perfetto event for this?
+      DLOG(ERROR) << "One or more events lost during trace capture";
+      break;
+    case 33:  // RTLostBuffer
+      // TODO: Emit a Perfetto event for this?
+      DLOG(ERROR) << "One or more buffers lost during trace capture";
+      break;
+    default:
+      // 34:  // RTLostFile
+      break;
+  }
+}
+
+bool EtwConsumer::DecodeCSwitchEvent(const EVENT_HEADER& header,
+                                     const ETW_BUFFER_CONTEXT& buffer_context,
+                                     base::span<const uint8_t> packet_data) {
+  using perfetto::protos::pbzero::CSwitchEtwEvent;
+
+  // Size of CSwitch v2 in bytes (4 x 32-bit plus 8 x 8-bit).
+  static constexpr size_t kMinimumCSwitchLength = 4 * 4 + 8;
+  if (packet_data.size() < kMinimumCSwitchLength) {
+    return false;
+  }
+
+  // Read and validate the contents of `packet_data`.
+  base::BufferIterator<const uint8_t> iterator{packet_data};
+  auto new_thread_id = *iterator.CopyObject<uint32_t>();
+  auto old_thread_id = *iterator.CopyObject<uint32_t>();
+  auto new_thread_priority = *iterator.Object<int8_t>();
+  auto old_thread_priority = *iterator.Object<int8_t>();
+  auto previous_c_state = *iterator.Object<uint8_t>();
+  (void)iterator.Object<int8_t>();  // SpareByte
+  auto old_thread_wait_reason = *iterator.Object<int8_t>();
+  if (old_thread_wait_reason < 0 ||
+      old_thread_wait_reason >= CSwitchEtwEvent::MAXIMUM_WAIT_REASON) {
+    return false;
+  }
+  auto old_thread_wait_mode = *iterator.Object<int8_t>();
+  if (old_thread_wait_mode < 0 ||
+      old_thread_wait_mode > CSwitchEtwEvent::USER_MODE) {
+    return false;
+  }
+  auto old_thread_state = *iterator.Object<int8_t>();
+  if (old_thread_state < 0 ||
+      old_thread_state > CSwitchEtwEvent::DEFERRED_READY) {
+    return false;
+  }
+  auto old_thread_wait_ideal_processor = *iterator.Object<int8_t>();
+  auto new_thread_wait_time = *iterator.CopyObject<uint32_t>();
+  (void)iterator.Object<uint32_t>();  // Reserved
+
+  // Generate a CSwitchEtwEvent.
+  auto* c_switch = MakeNextEvent(header, buffer_context)->set_c_switch();
+  c_switch->set_new_thread_id(new_thread_id);
+  c_switch->set_old_thread_id(old_thread_id);
+  c_switch->set_new_thread_priority(new_thread_priority);
+  c_switch->set_old_thread_priority(old_thread_priority);
+  c_switch->set_previous_c_state(previous_c_state);
+  c_switch->set_old_thread_wait_reason(
+      static_cast<CSwitchEtwEvent::OldThreadWaitReason>(
+          old_thread_wait_reason));
+  c_switch->set_old_thread_wait_mode(
+      static_cast<CSwitchEtwEvent::OldThreadWaitMode>(old_thread_wait_mode));
+  c_switch->set_old_thread_state(
+      static_cast<CSwitchEtwEvent::OldThreadState>(old_thread_state));
+  c_switch->set_old_thread_wait_ideal_processor(
+      old_thread_wait_ideal_processor);
+  c_switch->set_new_thread_wait_time(new_thread_wait_time);
+
+  return true;
+}
+
+perfetto::protos::pbzero::EtwTraceEvent* EtwConsumer::MakeNextEvent(
+    const EVENT_HEADER& header,
+    const ETW_BUFFER_CONTEXT& buffer_context) {
+  static const double qpc_ticks_per_second = []() {
+    LARGE_INTEGER perf_counter_frequency = {};
+    ::QueryPerformanceFrequency(&perf_counter_frequency);
+    double frequency = static_cast<double>(perf_counter_frequency.QuadPart);
+    CHECK_GT(frequency, 0.0);
+    return frequency;
+  }();
+
+  uint64_t now = static_cast<uint64_t>(
+      base::Time::kNanosecondsPerSecond *
+      static_cast<double>(header.TimeStamp.QuadPart) / qpc_ticks_per_second);
+  if (!etw_events_) {
+    // Resetting the `packet_handle_` finalizes previous data.
+    packet_handle_ = trace_writer_->NewTracePacket();
+    packet_handle_->set_timestamp(now);
+    etw_events_ = packet_handle_->set_etw_events();
+  }
+
+  auto* event = etw_events_->add_event();
+  event->set_timestamp(now);
+  event->set_cpu(buffer_context.ProcessorIndex);
+  return event;
+}
+
+}  // namespace tracing

components/tracing/common/etw_consumer_win.h

@@ -0,0 +1,101 @@
+// Copyright 2024 The Chromium Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef COMPONENTS_TRACING_COMMON_ETW_CONSUMER_WIN_H_
+#define COMPONENTS_TRACING_COMMON_ETW_CONSUMER_WIN_H_
+
+#include <stdint.h>
+
+#include <memory>
+
+#include "base/containers/span.h"
+#include "base/memory/raw_ptr.h"
+#include "base/sequence_checker.h"
+#include "base/thread_annotations.h"
+#include "base/win/event_trace_consumer.h"
+#include "components/tracing/tracing_export.h"
+#include "third_party/perfetto/include/perfetto/ext/tracing/core/trace_writer.h"
+#include "third_party/perfetto/include/perfetto/tracing/trace_writer_base.h"
+
+namespace perfetto::protos::pbzero {
+class EtwTraceEvent;
+class EtwTraceEventBundle;
+}  // namespace perfetto::protos::pbzero
+
+namespace tracing {
+
+// A consumer of events from the Windows system trace provider that emits
+// corresponding Perfetto trace events. An instance may be constructed on any
+// sequence. Its `ConsumeEvents()` method and its destructor must be called on
+// the same sequence.
+class TRACING_EXPORT EtwConsumer
+    : public base::win::EtwTraceConsumerBase<EtwConsumer> {
+ public:
+  // Receive events in the new EVENT_RECORD format.
+  static constexpr bool kEnableRecordMode = true;
+  // Do not convert timestampts to system time.
+  static constexpr bool kRawTimestamp = true;
+
+  // Constructs an instance that will emit Perfetto events via `trace_writer`.
+  explicit EtwConsumer(std::unique_ptr<perfetto::TraceWriterBase> trace_writer);
+  EtwConsumer(const EtwConsumer&) = delete;
+  EtwConsumer& operator=(const EtwConsumer&) = delete;
+  ~EtwConsumer();
+
+  // Consumes ETW events; blocking the calling thread. Returns when the ETW
+  // trace session is stopped.
+  void ConsumeEvents();
+
+  // base::win::EtwTraceConsumerBase<>:
+  static void ProcessEventRecord(EVENT_RECORD* event_record);
+  static bool ProcessBuffer(EVENT_TRACE_LOGFILE* buffer);
+
+ private:
+  // The type of a member function that handles an event originating from a
+  // specific provider.
+  using EventHandlerFunction =
+      void (EtwConsumer::*)(const EVENT_HEADER& header,
+                            const ETW_BUFFER_CONTEXT& buffer_context,
+                            base::span<const uint8_t> packet_data);
+
+  // Per-provider event handlers. `ProcessEventRecord` dispatches to these based
+  // on the ProviderId in the record's EventHeader.
+  void HandleThreadEvent(const EVENT_HEADER& header,
+                         const ETW_BUFFER_CONTEXT& buffer_context,
+                         base::span<const uint8_t> packet_data)
+      VALID_CONTEXT_REQUIRED(sequence_checker_);
+  void HandleLostEvent(const EVENT_HEADER& header,
+                       const ETW_BUFFER_CONTEXT& buffer_context,
+                       base::span<const uint8_t> packet_data)
+      VALID_CONTEXT_REQUIRED(sequence_checker_);
+
+  // Decodes a CSwitch Event and emits a Perfetto trace event; see
+  // https://learn.microsoft.com/en-us/windows/win32/etw/cswitch.
+  // Returns true on success, or false if `packet_data` is invalid.
+  bool DecodeCSwitchEvent(const EVENT_HEADER& header,
+                          const ETW_BUFFER_CONTEXT& buffer_context,
+                          base::span<const uint8_t> packet_data)
+      VALID_CONTEXT_REQUIRED(sequence_checker_);
+
+  // Returns a new perfetto trace event to be emitted for an ETW event with a
+  // given event header. The timestamp and cpu fields of the returned event are
+  // prepopulated.
+  perfetto::protos::pbzero::EtwTraceEvent* MakeNextEvent(
+      const EVENT_HEADER& header,
+      const ETW_BUFFER_CONTEXT& buffer_context)
+      VALID_CONTEXT_REQUIRED(sequence_checker_);
+
+  std::unique_ptr<perfetto::TraceWriterBase> trace_writer_
+      GUARDED_BY_CONTEXT(sequence_checker_);
+  perfetto::TraceWriter::TracePacketHandle packet_handle_
+      GUARDED_BY_CONTEXT(sequence_checker_);
+  raw_ptr<perfetto::protos::pbzero::EtwTraceEventBundle> etw_events_
+      GUARDED_BY_CONTEXT(sequence_checker_);
+
+  SEQUENCE_CHECKER(sequence_checker_);
+};
+
+}  // namespace tracing
+
+#endif  // COMPONENTS_TRACING_COMMON_ETW_CONSUMER_WIN_H_

components/tracing/common/etw_consumer_win_unittest.cc

@@ -0,0 +1,221 @@
+// Copyright 2024 The Chromium Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "components/tracing/common/etw_consumer_win.h"
+
+#include <windows.h>
+
+#include <stdint.h>
+
+#include <optional>
+#include <queue>
+#include <utility>
+#include <vector>
+
+#include "base/containers/span.h"
+#include "base/functional/bind.h"
+#include "base/functional/callback.h"
+#include "base/notreached.h"
+#include "base/numerics/safe_conversions.h"
+#include "testing/gtest/include/gtest/gtest.h"
+#include "third_party/perfetto/include/perfetto/protozero/message.h"
+#include "third_party/perfetto/include/perfetto/protozero/message_handle.h"
+#include "third_party/perfetto/include/perfetto/protozero/scattered_heap_buffer.h"
+#include "third_party/perfetto/include/perfetto/tracing/trace_writer_base.h"
+#include "third_party/perfetto/protos/perfetto/trace/etw/etw.pbzero.h"
+#include "third_party/perfetto/protos/perfetto/trace/etw/etw_event.pbzero.h"
+#include "third_party/perfetto/protos/perfetto/trace/etw/etw_event_bundle.pbzero.h"
+#include "third_party/perfetto/protos/perfetto/trace/trace_packet.pbzero.h"
+
+namespace tracing {
+
+namespace {
+
+// A trace writer that creates TracePacket messages on the heap and sends their
+// serialized form to an owner-provided callback.
+class FakeTraceWriter : public perfetto::TraceWriterBase,
+                        public protozero::MessageFinalizationListener {
+ public:
+  using TracePacketHandle =
+      protozero::MessageHandle<perfetto::protos::pbzero::TracePacket>;
+
+  explicit FakeTraceWriter(
+      base::RepeatingCallback<void(std::vector<uint8_t>)> on_packet)
+      : on_packet_(std::move(on_packet)) {}
+
+  // perfetto::TraceWriterBase:
+  TracePacketHandle NewTracePacket() override {
+    packet_ = std::make_unique<
+        protozero::HeapBuffered<perfetto::protos::pbzero::TracePacket>>();
+    TracePacketHandle handle(packet_->get());
+    handle.set_finalization_listener(this);
+    return handle;
+  }
+  void FinishTracePacket() override { NOTREACHED(); }
+  void Flush(std::function<void()> callback = {}) override {}
+  uint64_t written() const override { return 0u; }
+
+  // protozero::MessageFinalizationListener:
+  void OnMessageFinalized(protozero::Message* message) override {
+    on_packet_.Run(packet_->SerializeAsArray());
+    packet_.reset();
+  }
+
+ private:
+  base::RepeatingCallback<void(std::vector<uint8_t>)> on_packet_;
+  std::unique_ptr<
+      protozero::HeapBuffered<perfetto::protos::pbzero::TracePacket>>
+      packet_;
+};
+
+}  // namespace
+
+class EtwConsumerTest : public testing::Test {
+ protected:
+  // Generates an ETW CSwitch event with `packet_data` as its payload and sends
+  // it to the EtwConsumer for processing. If the EtwConsumer generates a
+  // TracePacket containing a `CSwitchEtwEvent`, `cswitch_decoder` is
+  // constructed from it.
+  void ProcessCSwitchEvent(
+      base::span<const uint8_t> packet_data,
+      std::optional<perfetto::protos::pbzero::CSwitchEtwEvent::Decoder>&
+          cswitch_decoder) {
+    std::optional<perfetto::protos::pbzero::TracePacket::Decoder> decoder;
+    ASSERT_NO_FATAL_FAILURE(ProcessThreadEvent(/*version=*/2u, /*opcode=*/36u,
+                                               packet_data, decoder));
+    if (!decoder.has_value()) {
+      return;
+    }
+    ASSERT_TRUE(decoder->has_timestamp());
+    ASSERT_NE(decoder->timestamp(), 0u);
+    ASSERT_TRUE(decoder->has_etw_events());
+    perfetto::protos::pbzero::EtwTraceEventBundle::Decoder bundle(
+        decoder->etw_events());
+    ASSERT_TRUE(bundle.has_event());
+    perfetto::protos::pbzero::EtwTraceEvent::Decoder event(*bundle.event());
+    ASSERT_TRUE(event.has_timestamp());
+    ASSERT_TRUE(event.has_cpu());
+    ASSERT_EQ(event.cpu(), kTestProcessorIndex);
+    ASSERT_TRUE(event.has_c_switch());
+    cswitch_decoder.emplace(event.c_switch());
+  }
+
+ private:
+  static constexpr uint16_t kTestProcessorIndex = 47;
+
+  // Generates an ETW EVENT_RECORD from the Thread provider of a particular
+  // version and opcode with `packet_data` as its payload and sends it to the
+  // EtwConsumer for processing. If the EtwConsumer generates a TracePacket,
+  // `decoder` is constructed from it.
+  void ProcessThreadEvent(
+      uint8_t version,
+      uint8_t opcode,
+      base::span<const uint8_t> packet_data,
+      std::optional<perfetto::protos::pbzero::TracePacket::Decoder>& decoder) {
+    ProcessEvent({0x3d6fa8d1,
+                  0xfe05,
+                  0x11d0,
+                  {0x9d, 0xda, 0x00, 0xc0, 0x4f, 0xd7, 0xba, 0x7c}},
+                 version, opcode, packet_data, decoder);
+  }
+
+  // Generates an ETW EVENT_RECORD for a given trace provider of a particular
+  // version and opcode with `packet_data` as its payload and sends it to the
+  // EtwConsumer for processing. If the EtwConsumer generates a TracePacket,
+  // `decoder` is constructed from it.
+  void ProcessEvent(
+      const GUID& provider,
+      uint8_t version,
+      uint8_t opcode,
+      base::span<const uint8_t> packet_data,
+      std::optional<perfetto::protos::pbzero::TracePacket::Decoder>& decoder) {
+    EVENT_RECORD event_record = {
+        .EventHeader = {.TimeStamp = {},
+                        .ProviderId = provider,
+                        .EventDescriptor = {.Version = version,
+                                            .Opcode = opcode}},
+        .BufferContext = {.ProcessorIndex = kTestProcessorIndex},
+        .UserDataLength = base::checked_cast<uint16_t>(packet_data.size()),
+        .UserData = const_cast<uint8_t*>(packet_data.data()),
+        .UserContext = &etw_consumer_};
+    ::QueryPerformanceCounter(&event_record.EventHeader.TimeStamp);
+    etw_consumer_.ProcessEventRecord(&event_record);
+
+    EVENT_TRACE_LOGFILE event_trace_logfile = {.Context = &etw_consumer_};
+    ASSERT_TRUE(etw_consumer_.ProcessBuffer(&event_trace_logfile));
+
+    if (messages_.empty()) {
+      return;
+    }
+    auto& message = messages_.front();
+    decoder.emplace(message.data(), message.size());
+    // The decoder references the memory owned by `message`, so the item in
+    // `messages_` must stay alive.
+  }
+
+  // Called by FakeTraceWriter to process the message for a TracePacket.
+  void OnPacket(std::vector<uint8_t> message) {
+    messages_.push(std::move(message));
+  }
+
+  EtwConsumer etw_consumer_{std::make_unique<FakeTraceWriter>(
+      base::BindRepeating(&EtwConsumerTest::OnPacket, base::Unretained(this)))};
+  std::queue<std::vector<uint8_t>> messages_;
+};
+
+// Tests that no CSwitchEtwEvent is emitted for an empty CSwitch ETW event.
+TEST_F(EtwConsumerTest, CSwitchEventIsEmpty) {
+  std::optional<perfetto::protos::pbzero::CSwitchEtwEvent::Decoder>
+      cswitch_decoder;
+  ASSERT_NO_FATAL_FAILURE(ProcessCSwitchEvent({}, cswitch_decoder));
+  ASSERT_FALSE(cswitch_decoder.has_value());
+}
+
+// Tests that no CSwitchEtwEvent is emitted for a small CSwitch ETW event.
+TEST_F(EtwConsumerTest, CSwitchEventIsTooShort) {
+  static constexpr uint8_t kData[] = {0x00, 23};
+  std::optional<perfetto::protos::pbzero::CSwitchEtwEvent::Decoder>
+      cswitch_decoder;
+  ASSERT_NO_FATAL_FAILURE(ProcessCSwitchEvent({kData}, cswitch_decoder));
+  ASSERT_FALSE(cswitch_decoder.has_value());
+}
+
+// Tests that CSwitchEtwEvent is emitted for a CSwitch ETW event.
+TEST_F(EtwConsumerTest, CSwitchEvent) {
+  static constexpr uint8_t kData[] = {
+      0x55, 0x00, 0x00, 0x00,  // new_thread_id
+      0x44, 0x00, 0x00, 0x00,  // old_thread_id
+      0x01,                    // new_thread_priority
+      0x02,                    // old_thread_priority
+      0x03,                    // previous_c_state
+      0x42,                    // SpareByte
+      36,                      // old_thread_wait_reason = WR_RUNDOWN
+      1,                       // old_thread_wait_mode = USER_MODE
+      7,                       // old_thread_state = DEFERRED_READY
+      0x04,                    // old_thread_wait_ideal_processor
+      0x05, 0x00, 0x00, 0x00,  // new_thread_wait_time
+      0x42, 0x42, 0x42, 0x42,  // Reserved
+  };
+  std::optional<perfetto::protos::pbzero::CSwitchEtwEvent::Decoder>
+      cswitch_decoder;
+  ASSERT_NO_FATAL_FAILURE(ProcessCSwitchEvent({kData}, cswitch_decoder));
+  ASSERT_TRUE(cswitch_decoder.has_value());
+  auto& c_switch = *cswitch_decoder;
+
+  EXPECT_EQ(0x55u, c_switch.new_thread_id());
+  EXPECT_EQ(0x44u, c_switch.old_thread_id());
+  EXPECT_EQ(0x01, c_switch.new_thread_priority());
+  EXPECT_EQ(0x02, c_switch.old_thread_priority());
+  EXPECT_EQ(0x03u, c_switch.previous_c_state());
+  EXPECT_EQ(perfetto::protos::pbzero::CSwitchEtwEvent::WR_RUNDOWN,
+            c_switch.old_thread_wait_reason());
+  EXPECT_EQ(perfetto::protos::pbzero::CSwitchEtwEvent::USER_MODE,
+            c_switch.old_thread_wait_mode());
+  EXPECT_EQ(perfetto::protos::pbzero::CSwitchEtwEvent::DEFERRED_READY,
+            c_switch.old_thread_state());
+  EXPECT_EQ(0x04, c_switch.old_thread_wait_ideal_processor());
+  EXPECT_EQ(0x05u, c_switch.new_thread_wait_time());
+}
+
+}  // namespace tracing

components/tracing/common/etw_system_data_source_win.cc

@@ -14,29 +14,20 @@
 #include <evntcons.h>
 #include <evntrace.h>
 
-#include <utility>
+#include <ios>
 
 #include "base/check.h"
-#include "base/check_op.h"
-#include "base/containers/buffer_iterator.h"
-#include "base/containers/span.h"
 #include "base/logging.h"
 #include "base/task/thread_pool.h"
-#include "base/win/event_trace_consumer.h"
+#include "components/tracing/common/etw_consumer_win.h"
+#include "third_party/perfetto/include/perfetto/ext/tracing/core/trace_writer.h"
 #include "third_party/perfetto/include/perfetto/tracing/core/data_source_descriptor.h"
-#include "third_party/perfetto/protos/perfetto/config/data_source_config.gen.h"
 #include "third_party/perfetto/protos/perfetto/config/etw/etw_config.gen.h"
-#include "third_party/perfetto/protos/perfetto/trace/etw/etw.pbzero.h"
-#include "third_party/perfetto/protos/perfetto/trace/etw/etw_event.pbzero.h"
-#include "third_party/perfetto/protos/perfetto/trace/etw/etw_event_bundle.pbzero.h"
 
 namespace tracing {
 
 namespace {
 
-constexpr wchar_t kEtwSystemSessionName[] = L"org.chromium.etw_system";
-constexpr uint8_t kCSwitchEventOpcode = 36;
-
 ULONG EtwSystemFlagsFromEnum(
     perfetto::protos::gen::EtwConfig::KernelFlag flag) {
   switch (flag) {
@@ -49,36 +40,7 @@ ULONG EtwSystemFlagsFromEnum(
 
 }  // namespace
 
-class EtwSystemDataSource::Consumer
-    : public base::win::EtwTraceConsumerBase<Consumer> {
- public:
-  static constexpr bool kEnableRecordMode = true;
-  static constexpr bool kRawTimestamp = true;
-
-  explicit Consumer(std::unique_ptr<perfetto::TraceWriterBase>);
-  ~Consumer();
-
-  static void ProcessEventRecord(EVENT_RECORD* event_record);
-  static bool ProcessBuffer(EVENT_TRACE_LOGFILE* buffer);
-
-  void ConsumeEvents();
-
- private:
-  void ProcessEventRecordImpl(const EVENT_HEADER* header,
-                              const ETW_BUFFER_CONTEXT* buffer_context,
-                              base::span<uint8_t> packet_data)
-      VALID_CONTEXT_REQUIRED(sequence_checker_);
-
-  std::unique_ptr<perfetto::TraceWriterBase> trace_writer_
-      GUARDED_BY_CONTEXT(sequence_checker_);
-  perfetto::TraceWriter::TracePacketHandle packet_handle_
-      GUARDED_BY_CONTEXT(sequence_checker_);
-  raw_ptr<perfetto::protos::pbzero::EtwTraceEventBundle> etw_events_
-      GUARDED_BY_CONTEXT(sequence_checker_);
-
-  SEQUENCE_CHECKER(sequence_checker_);
-};
-
+// static
 void EtwSystemDataSource::Register() {
   perfetto::DataSourceDescriptor desc;
   desc.set_name("org.chromium.etw_system");
@@ -93,7 +55,9 @@ EtwSystemDataSource::EtwSystemDataSource()
   DETACH_FROM_SEQUENCE(sequence_checker_);
 }
 
-EtwSystemDataSource::~EtwSystemDataSource() = default;
+EtwSystemDataSource::~EtwSystemDataSource() {
+  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
+}
 
 void EtwSystemDataSource::OnSetup(const SetupArgs& args) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
@@ -101,6 +65,8 @@ void EtwSystemDataSource::OnSetup(const SetupArgs& args) {
 }
 
 void EtwSystemDataSource::OnStart(const StartArgs&) {
+  static constexpr wchar_t kEtwSystemSessionName[] = L"org.chromium.etw_system";
+
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
 
   if (data_source_config_.etw_config_raw().empty()) {
@@ -156,16 +122,17 @@ void EtwSystemDataSource::OnStart(const StartArgs&) {
     return;
   }
 
-  consumer_ = {new Consumer(CreateTraceWriter()),
+  consumer_ = {new EtwConsumer(CreateTraceWriter()),
                base::OnTaskRunnerDeleter(consume_task_runner_)};
   hr = consumer_->OpenRealtimeSession(kEtwSystemSessionName);
   if (FAILED(hr)) {
     etw_controller_.Stop(nullptr);
+    consumer_.reset();
     DLOG(ERROR) << "Failed to open system trace session: 0x" << std::hex << hr;
     return;
   }
   consume_task_runner_->PostTask(
-      FROM_HERE, base::BindOnce(&EtwSystemDataSource::Consumer::ConsumeEvents,
+      FROM_HERE, base::BindOnce(&EtwConsumer::ConsumeEvents,
                                 base::Unretained(consumer_.get())));
 }
 
@@ -175,145 +142,10 @@ void EtwSystemDataSource::OnStop(const StopArgs&) {
   consumer_.reset();
 }
 
-EtwSystemDataSource::Consumer::Consumer(
-    std::unique_ptr<perfetto::TraceWriterBase> trace_writer)
-    : trace_writer_(std::move(trace_writer)) {
-  DETACH_FROM_SEQUENCE(sequence_checker_);
-}
-
-// Decodes a CSwitch Event.
-// https://learn.microsoft.com/en-us/windows/win32/etw/cswitch
-// Returns true on success, or false if `packet_data` is invalid.
-bool EtwSystemDataSource::DecodeCSwitchEvent(
-    base::span<uint8_t> packet_data,
-    perfetto::protos::pbzero::EtwTraceEvent& event) {
-  // Size of CSwitch v2 in bytes.
-  static constexpr size_t kMinimumCSwitchLength = 24;
-  if (packet_data.size() < kMinimumCSwitchLength) {
-    return false;
-  }
-  base::BufferIterator<uint8_t> iterator{packet_data};
-  auto* c_switch = event.set_c_switch();
-  c_switch->set_new_thread_id(*iterator.Object<uint32_t>());
-  c_switch->set_old_thread_id(*iterator.Object<uint32_t>());
-  c_switch->set_new_thread_priority(*iterator.Object<int8_t>());
-  c_switch->set_old_thread_priority(*iterator.Object<int8_t>());
-  c_switch->set_previous_c_state(*iterator.Object<uint8_t>());
-
-  // SpareByte
-  std::ignore = iterator.Object<uint8_t>();
-
-  const int8_t* old_thread_wait_reason = iterator.Object<int8_t>();
-  if (!(*old_thread_wait_reason >= 0 &&
-        *old_thread_wait_reason <
-            perfetto::protos::pbzero::CSwitchEtwEvent::MAXIMUM_WAIT_REASON)) {
-    return false;
-  }
-  c_switch->set_old_thread_wait_reason(
-      static_cast<
-          perfetto::protos::pbzero::CSwitchEtwEvent::OldThreadWaitReason>(
-          *old_thread_wait_reason));
-
-  const int8_t* old_thread_wait_mode = iterator.Object<int8_t>();
-  if (!(*old_thread_wait_mode >= 0 &&
-        *old_thread_wait_mode <=
-            perfetto::protos::pbzero::CSwitchEtwEvent::USER_MODE)) {
-    return false;
-  }
-  c_switch->set_old_thread_wait_mode(
-      static_cast<perfetto::protos::pbzero::CSwitchEtwEvent::OldThreadWaitMode>(
-          *old_thread_wait_mode));
-
-  const int8_t* old_thread_state = iterator.Object<int8_t>();
-  if (!(*old_thread_state >= 0 &&
-        *old_thread_state <=
-            perfetto::protos::pbzero::CSwitchEtwEvent::DEFERRED_READY)) {
-    return false;
-  }
-  c_switch->set_old_thread_state(
-      static_cast<perfetto::protos::pbzero::CSwitchEtwEvent::OldThreadState>(
-          *old_thread_state));
-
-  c_switch->set_old_thread_wait_ideal_processor(*iterator.Object<int8_t>());
-  c_switch->set_new_thread_wait_time(*iterator.Object<uint32_t>());
-  return true;
-}
-
-EtwSystemDataSource::Consumer::~Consumer() = default;
-
-void EtwSystemDataSource::Consumer::ConsumeEvents() {
-  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-  base::ScopedBlockingCall scoped_blocking(FROM_HERE,
-                                           base::BlockingType::MAY_BLOCK);
-  Consume();
-}
-
-void EtwSystemDataSource::Consumer::ProcessEventRecord(
-    EVENT_RECORD* event_record) {
-  // ThreadGuid, 3d6fa8d1-fe05-11d0-9dda-00c04fd7ba7c
-  // https://learn.microsoft.com/en-us/windows/win32/etw/nt-kernel-logger-constants
-  static constexpr GUID kThreadGuid = {
-      0x3d6fa8d1,
-      0xfe05,
-      0x11d0,
-      {0x9d, 0xda, 0x00, 0xc0, 0x4f, 0xd7, 0xba, 0x7c}};
-
-  Consumer* self = reinterpret_cast<Consumer*>(event_record->UserContext);
-  DCHECK_CALLED_ON_VALID_SEQUENCE(self->sequence_checker_);
-
-  if (IsEqualGUID(event_record->EventHeader.ProviderId, kThreadGuid)) {
-    self->ProcessEventRecordImpl(&event_record->EventHeader,
-                                 &event_record->BufferContext,
-                                 {static_cast<uint8_t*>(event_record->UserData),
-                                  event_record->UserDataLength});
-  }
-}
-
-void EtwSystemDataSource::Consumer::ProcessEventRecordImpl(
-    const EVENT_HEADER* header,
-    const ETW_BUFFER_CONTEXT* buffer_context,
-    base::span<uint8_t> packet_data) {
-  static const int64_t qpc_ticks_per_second = []() {
-    LARGE_INTEGER perf_counter_frequency = {};
-    ::QueryPerformanceFrequency(&perf_counter_frequency);
-    CHECK_GT(perf_counter_frequency.QuadPart, 0);
-    return perf_counter_frequency.QuadPart;
-  }();
-
-  uint64_t now =
-      static_cast<uint64_t>(base::Time::kNanosecondsPerSecond *
-                            static_cast<double>(header->TimeStamp.QuadPart) /
-                            static_cast<double>(qpc_ticks_per_second));
-  if (!etw_events_) {
-    // Resetting the `packet_handle_` finalizes previous data.
-    packet_handle_ = trace_writer_->NewTracePacket();
-    packet_handle_->set_timestamp(now);
-    etw_events_ = packet_handle_->set_etw_events();
-  }
-
-  auto* event = etw_events_->add_event();
-  event->set_timestamp(now);
-  event->set_cpu(buffer_context->ProcessorIndex);
-  if (header->EventDescriptor.Opcode == kCSwitchEventOpcode) {
-    bool result = DecodeCSwitchEvent(packet_data, *event);
-    if (!result) {
-      DLOG(ERROR) << "Error decoding CSwitch Event";
-    }
-  }
-}
-
-bool EtwSystemDataSource::Consumer::ProcessBuffer(EVENT_TRACE_LOGFILE* buffer) {
-  Consumer* self = reinterpret_cast<Consumer*>(buffer->Context);
-  DCHECK_CALLED_ON_VALID_SEQUENCE(self->sequence_checker_);
-  self->etw_events_ = nullptr;
-  self->packet_handle_ = {};
-  return true;
-}
-
 }  // namespace tracing
 
 PERFETTO_DEFINE_DATA_SOURCE_STATIC_MEMBERS_WITH_ATTRS(
-    BASE_EXPORT,
+    TRACING_EXPORT,
     tracing::EtwSystemDataSource);
 
 // This should go after PERFETTO_DEFINE_DATA_SOURCE_STATIC_MEMBERS_WITH_ATTRS

components/tracing/common/etw_system_data_source_win.h

@@ -7,21 +7,22 @@
 
 #include <memory>
 
-#include "base/containers/span.h"
 #include "base/memory/scoped_refptr.h"
 #include "base/sequence_checker.h"
 #include "base/task/sequenced_task_runner.h"
 #include "base/thread_annotations.h"
 #include "base/win/event_trace_controller.h"
 #include "components/tracing/tracing_export.h"
-#include "third_party/perfetto/include/perfetto/ext/tracing/core/trace_writer.h"
-#include "third_party/perfetto/include/perfetto/tracing/core/forward_decls.h"
 #include "third_party/perfetto/include/perfetto/tracing/data_source.h"
-#include "third_party/perfetto/include/perfetto/tracing/tracing.h"
-#include "third_party/perfetto/protos/perfetto/trace/etw/etw_event.pbzero.h"
+
+namespace perfetto {
+class TraceWriterBase;
+}
 
 namespace tracing {
 
+class EtwConsumer;
+
 class TRACING_EXPORT EtwSystemDataSource
     : public perfetto::DataSource<EtwSystemDataSource> {
  public:
@@ -30,26 +31,23 @@ class TRACING_EXPORT EtwSystemDataSource
   static void Register();
 
   EtwSystemDataSource();
+  EtwSystemDataSource(const EtwSystemDataSource&) = delete;
+  EtwSystemDataSource& operator=(const EtwSystemDataSource&) = delete;
   ~EtwSystemDataSource() override;
 
+  // perfetto::DataSource:
   void OnSetup(const SetupArgs&) override;
   void OnStart(const StartArgs&) override;
   void OnStop(const StopArgs&) override;
 
-  static bool DecodeCSwitchEvent(
-      base::span<uint8_t> packet_data,
-      perfetto::protos::pbzero::EtwTraceEvent& event);
-
  private:
-  class Consumer;
-
   std::unique_ptr<perfetto::TraceWriterBase> CreateTraceWriter();
 
   base::win::EtwTraceController etw_controller_
       GUARDED_BY_CONTEXT(sequence_checker_);
   scoped_refptr<base::SequencedTaskRunner> consume_task_runner_
       GUARDED_BY_CONTEXT(sequence_checker_);
-  std::unique_ptr<Consumer, base::OnTaskRunnerDeleter> consumer_
+  std::unique_ptr<EtwConsumer, base::OnTaskRunnerDeleter> consumer_
       GUARDED_BY_CONTEXT(sequence_checker_);
 
   perfetto::DataSourceConfig data_source_config_;

components/tracing/common/etw_system_data_source_win_unittest.cc

@@ -1,77 +0,0 @@
-// Copyright 2024 The Chromium Authors
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#ifdef UNSAFE_BUFFERS_BUILD
-// TODO(crbug.com/40285824): Remove this and convert code to safer constructs.
-#pragma allow_unsafe_buffers
-#endif
-
-#include "components/tracing/common/etw_system_data_source_win.h"
-
-#include <vector>
-
-#include "base/containers/span.h"
-#include "base/logging.h"
-#include "testing/gtest/include/gtest/gtest.h"
-#include "third_party/perfetto/include/perfetto/protozero/scattered_heap_buffer.h"
-#include "third_party/perfetto/protos/perfetto/trace/etw/etw.pbzero.h"
-#include "third_party/perfetto/protos/perfetto/trace/etw/etw_event.pbzero.h"
-
-namespace tracing {
-
-TEST(EtwSystemDataSource, DecodeCSwitchEvent_Empty) {
-  std::vector<uint8_t> data = {};
-  protozero::HeapBuffered<perfetto::protos::pbzero::EtwTraceEvent> event;
-  EXPECT_FALSE(
-      EtwSystemDataSource::DecodeCSwitchEvent({data.data(), 0U}, *event.get()));
-}
-
-TEST(EtwSystemDataSource, DecodeCSwitchEvent_TooShort) {
-  std::vector<uint8_t> data(0x00, 23);
-  protozero::HeapBuffered<perfetto::protos::pbzero::EtwTraceEvent> event;
-  EXPECT_FALSE(EtwSystemDataSource::DecodeCSwitchEvent(
-      {data.data(), data.size()}, *event.get()));
-}
-
-TEST(EtwSystemDataSource, DecodeCSwitchEvent) {
-  std::vector<uint8_t> data = {
-      0x55, 0x00, 0x00, 0x00,  // new_thread_id
-      0x44, 0x00, 0x00, 0x00,  // old_thread_id
-      0x01,                    // new_thread_priority
-      0x02,                    // old_thread_priority
-      0x03,                    // previous_c_state
-      0x42,                    // SpareByte
-      36,                      // old_thread_wait_reason = WR_RUNDOWN
-      1,                       // old_thread_wait_mode = USER_MODE
-      7,                       // old_thread_state = DEFERRED_READY
-      0x04,                    // old_thread_wait_ideal_processor
-      0x05, 0x00, 0x00, 0x00,  // new_thread_wait_time
-      0x42, 0x42, 0x42, 0x42,  // Reserved
-  };
-  protozero::HeapBuffered<perfetto::protos::pbzero::EtwTraceEvent> event;
-  ASSERT_TRUE(EtwSystemDataSource::DecodeCSwitchEvent(
-      {data.data(), data.size()}, *event.get()));
-
-  auto serialized = event.SerializeAsString();
-  perfetto::protos::pbzero::EtwTraceEvent::Decoder decoded_event(serialized);
-
-  ASSERT_TRUE(decoded_event.has_c_switch());
-  perfetto::protos::pbzero::CSwitchEtwEvent::Decoder c_switch(
-      decoded_event.c_switch());
-  EXPECT_EQ(0x55u, c_switch.new_thread_id());
-  EXPECT_EQ(0x44u, c_switch.old_thread_id());
-  EXPECT_EQ(0x01, c_switch.new_thread_priority());
-  EXPECT_EQ(0x02, c_switch.old_thread_priority());
-  EXPECT_EQ(0x03u, c_switch.previous_c_state());
-  EXPECT_EQ(perfetto::protos::pbzero::CSwitchEtwEvent::WR_RUNDOWN,
-            c_switch.old_thread_wait_reason());
-  EXPECT_EQ(perfetto::protos::pbzero::CSwitchEtwEvent::USER_MODE,
-            c_switch.old_thread_wait_mode());
-  EXPECT_EQ(perfetto::protos::pbzero::CSwitchEtwEvent::DEFERRED_READY,
-            c_switch.old_thread_state());
-  EXPECT_EQ(0x04, c_switch.old_thread_wait_ideal_processor());
-  EXPECT_EQ(0x05u, c_switch.new_thread_wait_time());
-}
-
-}  // namespace tracing