perfetto: Switch to clearing incremental state periodically

The perfetto service now provides a signal for clearing of incremental
state. This patch plumbs the signal into our data sources down to the
event sink.

Bug: 928738
Change-Id: If4bfb5329eb50e8222cd670ed4a0ece0ad25081e
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1632213Reviewed-by: oysteine <oysteine@chromium.org>
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Commit-Queue: Eric Seckler <eseckler@chromium.org>
Cr-Commit-Position: refs/heads/master@{#664154}

services/tracing/perfetto/producer_host.cc

@@ -132,7 +132,10 @@ void ProducerHost::Flush(
 }
 
 void ProducerHost::ClearIncrementalState(const perfetto::DataSourceInstanceID*,
-                                         size_t) {}
+                                         size_t) {
+  DCHECK(producer_client_);
+  producer_client_->ClearIncrementalState();
+}
 
 // This data can come from a malicious child process. We don't do any
 // sanitization here because ProducerEndpoint::CommitData() (And any other

services/tracing/public/cpp/perfetto/perfetto_config.cc

@@ -51,6 +51,10 @@ perfetto::TraceConfig GetDefaultPerfettoConfig(
   builtin_data_sources->set_disable_trace_config(privacy_filtering_enabled);
   builtin_data_sources->set_disable_system_info(privacy_filtering_enabled);
 
+  // Clear incremental state every 5 seconds, so that we lose at most the first
+  // 5 seconds of the trace (if we wrap around perfetto's central buffer).
+  perfetto_config.mutable_incremental_state_config()->set_clear_period_ms(5000);
+
   // We strip the process filter from the config string we send to Perfetto,
   // so perfetto doesn't reject it from a future
   // TracingService::ChangeTraceConfig call due to being an unsupported

services/tracing/public/cpp/perfetto/perfetto_traced_process.h

@@ -48,6 +48,8 @@ class COMPONENT_EXPORT(TRACING_CPP) PerfettoTracedProcess final {
     // Flush the data source.
     virtual void Flush(base::RepeatingClosure flush_complete_callback) = 0;
 
+    virtual void ClearIncrementalState() {}
+
     const std::string& name() const { return name_; }
     uint64_t data_source_id() const { return data_source_id_; }
 

services/tracing/public/cpp/perfetto/producer_client.cc

@@ -92,6 +92,7 @@ void ProducerClient::NewDataSourceAdded(
   new_registration.set_name(data_source->name());
   new_registration.set_will_notify_on_start(true);
   new_registration.set_will_notify_on_stop(true);
+  new_registration.set_handles_incremental_state_clear(true);
   producer_host_->RegisterDataSource(std::move(new_registration));
 }
 
@@ -175,6 +176,12 @@ void ProducerClient::Flush(uint64_t flush_request_id,
   }
 }
 
+void ProducerClient::ClearIncrementalState() {
+  for (auto* data_source : PerfettoTracedProcess::Get()->data_sources()) {
+    data_source->ClearIncrementalState();
+  }
+}
+
 void ProducerClient::RegisterDataSource(const perfetto::DataSourceDescriptor&) {
   NOTREACHED();
 }

services/tracing/public/cpp/perfetto/producer_client.h

@@ -61,6 +61,7 @@ class COMPONENT_EXPORT(TRACING_CPP) ProducerClient
   void StopDataSource(uint64_t id, StopDataSourceCallback callback) override;
   void Flush(uint64_t flush_request_id,
              const std::vector<uint64_t>& data_source_ids) override;
+  void ClearIncrementalState() override;
 
   // perfetto::TracingService::ProducerEndpoint implementation.
   // Used by the TraceWriters

services/tracing/public/cpp/perfetto/thread_local_event_sink.cc

@@ -25,6 +25,4 @@ ThreadLocalEventSink::~ThreadLocalEventSink() {
       std::move(trace_writer_));
 }
 
-void ThreadLocalEventSink::ResetIncrementalState() {}
-
 }  // namespace tracing

services/tracing/public/cpp/perfetto/thread_local_event_sink.h

@@ -44,8 +44,6 @@ class COMPONENT_EXPORT(TRACING_CPP) ThreadLocalEventSink {
 
   virtual void Flush() = 0;
 
-  virtual void ResetIncrementalState();
-
   uint32_t session_id() const { return session_id_; }
 
  protected:

services/tracing/public/cpp/perfetto/trace_event_data_source.cc

@@ -368,12 +368,8 @@ void TraceEventDataSource::Flush(
       std::move(flush_complete_callback)));
 }
 
-void TraceEventDataSource::ResetIncrementalStateForTesting() {
-  auto* thread_local_event_sink =
-      static_cast<ThreadLocalEventSink*>(ThreadLocalEventSinkSlot()->Get());
-  if (thread_local_event_sink) {
-    thread_local_event_sink->ResetIncrementalState();
-  }
+void TraceEventDataSource::ClearIncrementalState() {
+  TrackEventThreadLocalEventSink::ClearIncrementalState();
 }
 
 ThreadLocalEventSink* TraceEventDataSource::CreateThreadLocalEventSink(

services/tracing/public/cpp/perfetto/trace_event_data_source.h

@@ -112,11 +112,7 @@ class COMPONENT_EXPORT(TRACING_CPP) TraceEventDataSource
   // Called from the PerfettoProducer.
   void StopTracing(base::OnceClosure stop_complete_callback) override;
   void Flush(base::RepeatingClosure flush_complete_callback) override;
-
-  // Resets emitted incremental state on the current thread and causes
-  // incremental data (e.g. interning index entries and a ThreadDescriptor) to
-  // be emitted again.
-  void ResetIncrementalStateForTesting();
+  void ClearIncrementalState() override;
 
   // Deletes TraceWriter safely on behalf of a ThreadLocalEventSink.
   void ReturnTraceWriter(

services/tracing/public/cpp/perfetto/trace_event_data_source_unittest.cc

@@ -915,7 +915,7 @@ TEST_F(TraceEventDataSourceTest, InternedStrings) {
 
     // Resetting the interning state causes ThreadDescriptor and interning
     // entries to be emitted again, with the same interning IDs.
-    TraceEventDataSource::GetInstance()->ResetIncrementalStateForTesting();
+    TraceEventDataSource::GetInstance()->ClearIncrementalState();
   }
 }
 

services/tracing/public/cpp/perfetto/track_event_thread_local_event_sink.cc

@@ -34,12 +34,6 @@ namespace {
 constexpr uint32_t kMagicChunkIndex =
     base::trace_event::TraceBufferChunk::kMaxChunkIndex;
 
-// Force an incremental state reset every 1000 events on each thread. This
-// limits the maximum number of events we lose when trace buffers wrap.
-// TODO(eseckler): Tune this value experimentally and/or replace it with a
-// signal by the service.
-constexpr int kMaxEventsBeforeIncrementalStateReset = 1000;
-
 // Replacement string for names of events with TRACE_EVENT_FLAG_COPY.
 const char* const kPrivacyFiltered = "PRIVACY_FILTERED";
 
@@ -119,6 +113,10 @@ void WriteDebugAnnotations(
 // static
 constexpr size_t TrackEventThreadLocalEventSink::kMaxCompleteEventDepth;
 
+// static
+std::atomic<uint32_t>
+    TrackEventThreadLocalEventSink::incremental_state_reset_id_{0};
+
 TrackEventThreadLocalEventSink::TrackEventThreadLocalEventSink(
     std::unique_ptr<perfetto::StartupTraceWriter> trace_writer,
     uint32_t session_id,
@@ -138,11 +136,9 @@ TrackEventThreadLocalEventSink::TrackEventThreadLocalEventSink(
 
 TrackEventThreadLocalEventSink::~TrackEventThreadLocalEventSink() {}
 
-// TODO(eseckler): Trigger this upon a signal from the perfetto, once perfetto
-// supports this.
-void TrackEventThreadLocalEventSink::ResetIncrementalState() {
-  reset_incremental_state_ = true;
-  events_since_last_incremental_state_reset_ = 0;
+// static
+void TrackEventThreadLocalEventSink::ClearIncrementalState() {
+  incremental_state_reset_id_.fetch_add(1u, std::memory_order_relaxed);
 }
 
 void TrackEventThreadLocalEventSink::AddTraceEvent(
@@ -176,6 +172,15 @@ void TrackEventThreadLocalEventSink::AddTraceEvent(
   bool copy_strings = flags & TRACE_EVENT_FLAG_COPY;
   bool explicit_timestamp = flags & TRACE_EVENT_FLAG_EXPLICIT_TIMESTAMP;
 
+  // We access |incremental_state_reset_id_| atomically but racily. It's OK if
+  // we don't notice the reset request immediately, as long as we will notice
+  // and service it eventually.
+  auto reset_id = incremental_state_reset_id_.load(std::memory_order_relaxed);
+  if (reset_id != last_incremental_state_reset_id_) {
+    reset_incremental_state_ = true;
+    last_incremental_state_reset_id_ = reset_id;
+  }
+
   if (reset_incremental_state_) {
     interned_event_categories_.ResetEmittedState();
     interned_event_names_.ResetEmittedState();
@@ -456,12 +461,6 @@ void TrackEventThreadLocalEventSink::AddTraceEvent(
     interned_annotation_names_.Clear();
     interned_source_locations_.Clear();
   }
-
-  events_since_last_incremental_state_reset_++;
-  if (events_since_last_incremental_state_reset_ >=
-      kMaxEventsBeforeIncrementalStateReset) {
-    ResetIncrementalState();
-  }
 }
 
 void TrackEventThreadLocalEventSink::UpdateDuration(

services/tracing/public/cpp/perfetto/track_event_thread_local_event_sink.h

@@ -34,8 +34,11 @@ class COMPONENT_EXPORT(TRACING_CPP) TrackEventThreadLocalEventSink
       bool proto_writer_filtering_enabled);
   ~TrackEventThreadLocalEventSink() override;
 
+  // Resets emitted incremental state on all threads and causes incremental data
+  // (e.g. interning index entries and a ThreadDescriptor) to be emitted again.
+  static void ClearIncrementalState();
+
   // ThreadLocalEventSink implementation:
-  void ResetIncrementalState() override;
   void AddTraceEvent(base::trace_event::TraceEvent* trace_event,
                      base::trace_event::TraceEventHandle* handle) override;
   void UpdateDuration(base::trace_event::TraceEventHandle handle,
@@ -54,12 +57,14 @@ class COMPONENT_EXPORT(TRACING_CPP) TrackEventThreadLocalEventSink
   InterningIndex<std::pair<const char*, const char*>>
       interned_source_locations_;
 
+  static std::atomic<uint32_t> incremental_state_reset_id_;
+
   bool reset_incremental_state_ = true;
+  uint32_t last_incremental_state_reset_id_ = 0;
   base::TimeTicks last_timestamp_;
   base::ThreadTicks last_thread_time_;
   int process_id_;
   int thread_id_;
-  int events_since_last_incremental_state_reset_ = 0;
 
   base::trace_event::TraceEvent complete_event_stack_[kMaxCompleteEventDepth];
   uint32_t current_stack_depth_ = 0;

services/tracing/public/mojom/data_source_descriptor_mojom_traits.cc

@@ -18,6 +18,8 @@ bool StructTraits<tracing::mojom::DataSourceRegistrationDataView,
   out->set_name(name);
   out->set_will_notify_on_start(data.will_notify_on_start());
   out->set_will_notify_on_stop(data.will_notify_on_stop());
+  out->set_handles_incremental_state_clear(
+      data.handles_incremental_state_clear());
   return true;
 }
 }  // namespace mojo

services/tracing/public/mojom/data_source_descriptor_mojom_traits.h

@@ -28,6 +28,10 @@ class StructTraits<tracing::mojom::DataSourceRegistrationDataView,
   static bool will_notify_on_stop(const perfetto::DataSourceDescriptor& src) {
     return src.will_notify_on_stop();
   }
+  static bool handles_incremental_state_clear(
+      const perfetto::DataSourceDescriptor& src) {
+    return src.handles_incremental_state_clear();
+  }
 
   static bool Read(tracing::mojom::DataSourceRegistrationDataView data,
                    perfetto::DataSourceDescriptor* out);

services/tracing/public/mojom/perfetto_service.mojom

@@ -90,6 +90,7 @@ struct DataSourceRegistration {
   string name;
   bool will_notify_on_start;
   bool will_notify_on_stop;
+  bool handles_incremental_state_clear;
 };
 
 // This is implemented by the tracing service and represents the service-
@@ -127,6 +128,7 @@ interface ProducerClient {
   // sent in the StartDataSource call.
   StopDataSource(uint64 id) => ();
   Flush(uint64 flush_request_id, array<uint64> data_source_ids);
+  ClearIncrementalState();
 };
 
 // This is implemented by the tracing service, and is essentially a singleton