trace_processor: introduce better support for sorting proto traces

Our hardcoded number for the proto window sizes has caused no end of
issues with various people running into issues with very long traces
which have very little data in them (so the whole trace could be
loaded into memory)

Use the observation that long traces which also consume a lot of space
(i.e. long running sched capture) also is almost certainly writing
to file. These traces should also be using flush (if not then they
will have to accept trace processor being unable to work with them
for now - see the bug for info on how we will fix this in the future)

For these traces, use 2 * max{file period, flush period} as the
window size for sorting. For all other traces, move to having an
infinite window size.

Bug: 132620032
Change-Id: Ib16cb820beeb11598106459ce419e47a15c7ac4d

10 files changed, 109 insertions, 48 deletions

diff --git a/include/perfetto/trace_processor/basic_types.h b/include/perfetto/trace_processor/basic_types.h
index 3e06ea691..0dafc35d7 100644
--- a/include/perfetto/trace_processor/basic_types.h
+++ b/include/perfetto/trace_processor/basic_types.h
@@ -28,7 +28,6 @@ namespace perfetto {
 namespace trace_processor {
 
 struct Config {
-  uint64_t window_size_ns = 180 * 1000 * 1000 * 1000ULL;  // 3 minutes.
 };
 
 // Represents a dynamically typed value returned by SQL.
diff --git a/src/trace_processor/BUILD.gn b/src/trace_processor/BUILD.gn
index 5e2182779..6c0e0465b 100644
--- a/src/trace_processor/BUILD.gn
+++ b/src/trace_processor/BUILD.gn
@@ -159,6 +159,7 @@ source_set("lib") {
     "../../gn:default_deps",
     "../../include/perfetto/traced:sys_stats_counters",
     "../../protos/perfetto/common:zero",
+    "../../protos/perfetto/config:zero",
     "../../protos/perfetto/metrics:zero",
     "../../protos/perfetto/metrics/android:zero",
     "../../protos/perfetto/trace:zero",
diff --git a/src/trace_processor/proto_trace_tokenizer.cc b/src/trace_processor/proto_trace_tokenizer.cc
index 9b7133199..9279e58c9 100644
--- a/src/trace_processor/proto_trace_tokenizer.cc
+++ b/src/trace_processor/proto_trace_tokenizer.cc
@@ -29,6 +29,7 @@
 #include "src/trace_processor/trace_sorter.h"
 #include "src/trace_processor/trace_storage.h"
 
+#include "perfetto/config/trace_config.pbzero.h"
 #include "perfetto/trace/ftrace/ftrace_event.pbzero.h"
 #include "perfetto/trace/ftrace/ftrace_event_bundle.pbzero.h"
 #include "perfetto/trace/interned_data/interned_data.pbzero.h"
@@ -81,24 +82,6 @@ void InternMessage(TraceProcessorContext* context,
 
 }  // namespace
 
-// static
-TraceType ProtoTraceTokenizer::GuessProtoTraceType(const uint8_t* data,
-                                                   size_t size) {
-  // Scan at most the first 128MB for a track event packet.
-  constexpr size_t kMaxScanSize = 128 * 1024 * 1024;
-  protos::pbzero::Trace::Decoder decoder(data, std::min(size, kMaxScanSize));
-  if (!decoder.has_packet())
-    return TraceType::kUnknownTraceType;
-  for (auto it = decoder.packet(); it; ++it) {
-    ProtoDecoder packet_decoder(it->data(), it->size());
-    if (PERFETTO_UNLIKELY(packet_decoder.FindField(
-            protos::pbzero::TracePacket::kTrackEventFieldNumber))) {
-      return TraceType::kProtoWithTrackEventsTraceType;
-    }
-  }
-  return TraceType::kProtoTraceType;
-}
-
 ProtoTraceTokenizer::ProtoTraceTokenizer(TraceProcessorContext* ctx)
     : context_(ctx) {}
 ProtoTraceTokenizer::~ProtoTraceTokenizer() = default;
@@ -236,6 +219,30 @@ util::Status ProtoTraceTokenizer::ParsePacket(TraceBlobView packet) {
     return util::OkStatus();
   }
 
+  if (decoder.has_trace_config()) {
+    auto config = decoder.trace_config();
+    protos::pbzero::TraceConfig::Decoder trace_config(config.data, config.size);
+
+    if (trace_config.write_into_file()) {
+      int64_t window_size_ns;
+      if (trace_config.has_flush_period_ms() &&
+          trace_config.flush_period_ms() > 0) {
+        // We use 2x the flush period as a margin of error to allow for any
+        // late flush responses to still be sorted correctly.
+        window_size_ns = trace_config.has_flush_period_ms() * 2 * 1000 * 1000;
+      } else {
+        constexpr uint64_t kDefaultWindowNs =
+            180 * 1000 * 1000 * 1000ULL;  // 3 minutes.
+        PERFETTO_ELOG(
+            "It is strongly recommended to have flush_period_ms set when "
+            "write_into_file is turned on. You will likely have many dropped "
+            "events because of inability to sort the events correctly.");
+        window_size_ns = static_cast<int64_t>(kDefaultWindowNs);
+      }
+      context_->sorter->SetWindowSizeNs(window_size_ns);
+    }
+  }
+
   // Use parent data and length because we want to parse this again
   // later to get the exact type of the packet.
   context_->sorter->PushTracePacket(timestamp, std::move(packet));
diff --git a/src/trace_processor/proto_trace_tokenizer.h b/src/trace_processor/proto_trace_tokenizer.h
index 918c25a2e..e227ce15b 100644
--- a/src/trace_processor/proto_trace_tokenizer.h
+++ b/src/trace_processor/proto_trace_tokenizer.h
@@ -40,15 +40,6 @@ class TraceStorage;
 // (or subfields, for the case of ftrace) with their timestamps.
 class ProtoTraceTokenizer : public ChunkedTraceReader {
  public:
-  // Scans the beginning of the trace for valid TracePackets to determine if the
-  // trace contains TrackEvents.
-  //
-  // TODO(eseckler): This is a pretty bad hack to enable us to choose a
-  // different sorting window size for traces with TrackEvents. We should
-  // reconsider and redesign our sorting strategy, so that we don't need to
-  // change global trace processor options if TrackEvents are present.
-  static TraceType GuessProtoTraceType(const uint8_t* data, size_t size);
-
   // |reader| is the abstract method of getting chunks of size |chunk_size_b|
   // from a trace file with these chunks parsed into |trace|.
   explicit ProtoTraceTokenizer(TraceProcessorContext*);
diff --git a/src/trace_processor/trace_processor_context.h b/src/trace_processor/trace_processor_context.h
index 5a4f815df..bff471e49 100644
--- a/src/trace_processor/trace_processor_context.h
+++ b/src/trace_processor/trace_processor_context.h
@@ -19,6 +19,8 @@
 
 #include <memory>
 
+#include "perfetto/trace_processor/basic_types.h"
+
 namespace perfetto {
 namespace trace_processor {
 
@@ -39,6 +41,7 @@ class TraceProcessorContext {
   TraceProcessorContext();
   ~TraceProcessorContext();
 
+  Config config;
   std::unique_ptr<ArgsTracker> args_tracker;
   std::unique_ptr<SliceTracker> slice_tracker;
   std::unique_ptr<ProcessTracker> process_tracker;
diff --git a/src/trace_processor/trace_processor_impl.cc b/src/trace_processor/trace_processor_impl.cc
index dff845103..19ad94a56 100644
--- a/src/trace_processor/trace_processor_impl.cc
+++ b/src/trace_processor/trace_processor_impl.cc
@@ -246,10 +246,10 @@ TraceType GuessTraceType(const uint8_t* data, size_t size) {
     if (first_word == kFuchsiaMagicNumber)
       return kFuchsiaTraceType;
   }
-  return ProtoTraceTokenizer::GuessProtoTraceType(data, size);
+  return kProtoTraceType;
 }
 
-TraceProcessorImpl::TraceProcessorImpl(const Config& cfg) : cfg_(cfg) {
+TraceProcessorImpl::TraceProcessorImpl(const Config& cfg) {
   sqlite3* db = nullptr;
   PERFETTO_CHECK(sqlite3_open(":memory:", &db) == SQLITE_OK);
   InitializeSqlite(db);
@@ -257,6 +257,7 @@ TraceProcessorImpl::TraceProcessorImpl(const Config& cfg) : cfg_(cfg) {
   CreateBuiltinViews(db);
   db_.reset(std::move(db));
 
+  context_.config = cfg;
   context_.storage.reset(new TraceStorage());
   context_.args_tracker.reset(new ArgsTracker(&context_));
   context_.slice_tracker.reset(new SliceTracker(&context_));
@@ -313,36 +314,37 @@ util::Status TraceProcessorImpl::Parse(std::unique_ptr<uint8_t[]> data,
           stats::guess_trace_type_duration_ns);
       trace_type = GuessTraceType(data.get(), size);
     }
-    int64_t window_size_ns = static_cast<int64_t>(cfg_.window_size_ns);
     switch (trace_type) {
-      case kJsonTraceType:
+      case kJsonTraceType: {
         PERFETTO_DLOG("Legacy JSON trace detected");
 #if PERFETTO_BUILDFLAG(PERFETTO_STANDALONE_BUILD)
         context_.chunk_reader.reset(new JsonTraceTokenizer(&context_));
         // JSON traces have no guarantees about the order of events in them.
-        window_size_ns = std::numeric_limits<int64_t>::max();
+        int64_t window_size_ns = std::numeric_limits<int64_t>::max();
         context_.sorter.reset(new TraceSorter(&context_, window_size_ns));
         context_.parser.reset(new JsonTraceParser(&context_));
 #else
         PERFETTO_FATAL("JSON traces only supported in standalone mode.");
 #endif
         break;
-      case kProtoWithTrackEventsTraceType:
-      case kProtoTraceType:
-        if (trace_type == kProtoWithTrackEventsTraceType) {
-          // TrackEvents can be ordered arbitrarily due to out-of-order absolute
-          // timestamps and cross-packet-sequence events (e.g. async events).
-          window_size_ns = std::numeric_limits<int64_t>::max();
-        }
+      }
+      case kProtoTraceType: {
+        // This will be reduced once we read the trace config and we see flush
+        // period being set.
+        int64_t window_size_ns = std::numeric_limits<int64_t>::max();
         context_.chunk_reader.reset(new ProtoTraceTokenizer(&context_));
         context_.sorter.reset(new TraceSorter(&context_, window_size_ns));
         context_.parser.reset(new ProtoTraceParser(&context_));
         break;
-      case kFuchsiaTraceType:
+      }
+      case kFuchsiaTraceType: {
+        constexpr uint64_t kDefaultWindowNs =
+            180 * 1000 * 1000 * 1000ULL;  // 3 minutes.
         context_.chunk_reader.reset(new FuchsiaTraceTokenizer(&context_));
-        context_.sorter.reset(new TraceSorter(&context_, window_size_ns));
+        context_.sorter.reset(new TraceSorter(&context_, kDefaultWindowNs));
         context_.parser.reset(new FuchsiaTraceParser(&context_));
         break;
+      }
       case kUnknownTraceType:
         return util::ErrStatus("Unknown trace type provided");
     }
diff --git a/src/trace_processor/trace_processor_impl.h b/src/trace_processor/trace_processor_impl.h
index 9d4c3234d..ebf41a7a2 100644
--- a/src/trace_processor/trace_processor_impl.h
+++ b/src/trace_processor/trace_processor_impl.h
@@ -37,7 +37,6 @@ namespace trace_processor {
 enum TraceType {
   kUnknownTraceType,
   kProtoTraceType,
-  kProtoWithTrackEventsTraceType,
   kJsonTraceType,
   kFuchsiaTraceType,
 };
@@ -83,8 +82,6 @@ class TraceProcessorImpl : public TraceProcessor {
   // This is atomic because it is set by the CTRL-C signal handler and we need
   // to prevent single-flow compiler optimizations in ExecuteQuery().
   std::atomic<bool> query_interrupted_{false};
-
-  const Config cfg_;
 };
 
 // The pointer implementation of TraceProcessor::Iterator.
diff --git a/src/trace_processor/trace_sorter.cc b/src/trace_processor/trace_sorter.cc
index b5428d933..315460c71 100644
--- a/src/trace_processor/trace_sorter.cc
+++ b/src/trace_processor/trace_sorter.cc
@@ -76,6 +76,7 @@ void TraceSorter::Queue::Sort() {
 // time in a profiler.
 void TraceSorter::SortAndExtractEventsBeyondWindow(int64_t window_size_ns) {
   DCHECK_ftrace_batch_cpu(kNoBatch);
+
   constexpr int64_t kTsMax = std::numeric_limits<int64_t>::max();
   const bool was_empty = global_min_ts_ == kTsMax && global_max_ts_ == 0;
   int64_t extract_end_ts = global_max_ts_ - window_size_ns;
diff --git a/src/trace_processor/trace_sorter.h b/src/trace_processor/trace_sorter.h
index 4178f3d77..839453b8e 100644
--- a/src/trace_processor/trace_sorter.h
+++ b/src/trace_processor/trace_sorter.h
@@ -221,8 +221,21 @@ class TraceSorter {
     SortAndExtractEventsBeyondWindow(/*window_size_ns=*/0);
   }
 
-  void set_window_ns_for_testing(int64_t window_size_ns) {
+  // Sets the window size to be the size specified (which should be lower than
+  // any previous window size specified) and flushes any data beyond
+  // this window size.
+  // It is undefined to call this function with a window size greater than than
+  // the current size.
+  void SetWindowSizeNs(int64_t window_size_ns) {
+    PERFETTO_DCHECK(window_size_ns <= window_size_ns_);
+
+    PERFETTO_DLOG("Setting window size to be %" PRId64 " ns", window_size_ns);
     window_size_ns_ = window_size_ns;
+
+    // Fast path: if, globally, we are within the window size, then just exit.
+    if (global_max_ts_ - global_min_ts_ < window_size_ns)
+      return;
+    SortAndExtractEventsBeyondWindow(window_size_ns_);
   }
 
  private:
@@ -280,9 +293,9 @@ class TraceSorter {
     global_max_ts_ = std::max(global_max_ts_, queue->max_ts_);
     global_min_ts_ = std::min(global_min_ts_, queue->min_ts_);
 
+    // Fast path: if, globally, we are within the window size, then just exit.
     if (global_max_ts_ - global_min_ts_ < window_size_ns_)
       return;
-
     SortAndExtractEventsBeyondWindow(window_size_ns_);
   }
 
diff --git a/src/trace_processor/trace_sorter_unittest.cc b/src/trace_processor/trace_sorter_unittest.cc
index 8183db934..c1588cd7f 100644
--- a/src/trace_processor/trace_sorter_unittest.cc
+++ b/src/trace_processor/trace_sorter_unittest.cc
@@ -33,6 +33,7 @@ namespace {
 using ::testing::_;
 using ::testing::InSequence;
 using ::testing::Invoke;
+using ::testing::MockFunction;
 using ::testing::NiceMock;
 
 class MockTraceParser : public ProtoTraceParser {
@@ -117,7 +118,7 @@ TEST_F(TraceSorterTest, Ordering) {
   EXPECT_CALL(*parser_, MOCK_ParseTracePacket(1100, view_3.data(), 3));
   EXPECT_CALL(*parser_, MOCK_ParseFtracePacket(2, 1200, view_4.data(), 4));
 
-  context_.sorter->set_window_ns_for_testing(200);
+  context_.sorter->SetWindowSizeNs(200);
   context_.sorter->PushFtraceEvent(2 /*cpu*/, 1200 /*timestamp*/,
                                    std::move(view_4));
   context_.sorter->FinalizeFtraceEventBatch(2);
@@ -130,6 +131,52 @@ TEST_F(TraceSorterTest, Ordering) {
   context_.sorter->ExtractEventsForced();
 }
 
+TEST_F(TraceSorterTest, SetWindowSize) {
+  TraceBlobView view_1 = test_buffer_.slice(0, 1);
+  TraceBlobView view_2 = test_buffer_.slice(0, 2);
+  TraceBlobView view_3 = test_buffer_.slice(0, 3);
+  TraceBlobView view_4 = test_buffer_.slice(0, 4);
+
+  MockFunction<void(std::string check_point_name)> check;
+
+  {
+    InSequence s;
+
+    EXPECT_CALL(*parser_, MOCK_ParseFtracePacket(0, 1000, view_1.data(), 1));
+    EXPECT_CALL(*parser_, MOCK_ParseTracePacket(1001, view_2.data(), 2));
+    EXPECT_CALL(check, Call("1"));
+    EXPECT_CALL(*parser_, MOCK_ParseTracePacket(1100, view_3.data(), 3));
+    EXPECT_CALL(check, Call("2"));
+    EXPECT_CALL(*parser_, MOCK_ParseFtracePacket(2, 1200, view_4.data(), 4));
+  }
+
+  context_.sorter->SetWindowSizeNs(200);
+  context_.sorter->PushFtraceEvent(2 /*cpu*/, 1200 /*timestamp*/,
+                                   std::move(view_4));
+  context_.sorter->FinalizeFtraceEventBatch(2);
+  context_.sorter->PushTracePacket(1001, std::move(view_2));
+  context_.sorter->PushTracePacket(1100, std::move(view_3));
+
+  context_.sorter->PushFtraceEvent(0 /*cpu*/, 1000 /*timestamp*/,
+                                   std::move(view_1));
+  context_.sorter->FinalizeFtraceEventBatch(0);
+
+  // At this point, we should just flush the 1000 and 1001 packets.
+  context_.sorter->SetWindowSizeNs(101);
+
+  // Inform the mock about where we are.
+  check.Call("1");
+
+  // Now we should flush the 1100 packet.
+  context_.sorter->SetWindowSizeNs(99);
+
+  // Inform the mock about where we are.
+  check.Call("2");
+
+  // Now we should flush the 1200 packet.
+  context_.sorter->ExtractEventsForced();
+}
+
 // Simulates a random stream of ftrace events happening on random CPUs.
 // Tests that the output of the TraceSorter matches the timestamp order
 // (% events happening at the same time on different CPUs).