src/trace_processor/proto_trace_tokenizer.cc - third_party/perfetto - Git at Google

 /*
  * Copyright (C) 2018 The Android Open Source Project
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 #include "src/trace_processor/proto_trace_tokenizer.h"

 #include <string>

 #include "perfetto/base/logging.h"
 #include "perfetto/ext/base/utils.h"
 #include "perfetto/protozero/proto_decoder.h"
 #include "perfetto/protozero/proto_utils.h"
 #include "src/trace_processor/event_tracker.h"
 #include "src/trace_processor/process_tracker.h"
 #include "src/trace_processor/stats.h"
 #include "src/trace_processor/trace_blob_view.h"
 #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"
 #include "perfetto/trace/trace.pbzero.h"
 #include "perfetto/trace/track_event/task_execution.pbzero.h"
 #include "perfetto/trace/track_event/thread_descriptor.pbzero.h"
 #include "perfetto/trace/track_event/track_event.pbzero.h"

 namespace perfetto {
 namespace trace_processor {

 using protozero::ProtoDecoder;
 using protozero::proto_utils::MakeTagLengthDelimited;
 using protozero::proto_utils::MakeTagVarInt;
 using protozero::proto_utils::ParseVarInt;

 namespace {

 template <typename MessageType>
 void InternMessage(TraceProcessorContext* context,
                    ProtoIncrementalState::PacketSequenceState* state,
                    TraceBlobView message) {
   constexpr auto kIidFieldNumber = MessageType::kIidFieldNumber;

   uint32_t iid = 0;
   auto message_start = message.data();
   auto message_size = message.length();
   protozero::ProtoDecoder decoder(message_start, message_size);

   auto field = decoder.FindField(kIidFieldNumber);
   if (PERFETTO_UNLIKELY(!field)) {
     PERFETTO_ELOG("Interned message without interning_id");
     context->storage->IncrementStats(stats::interned_data_tokenizer_errors);
     return;
   }
   iid = field.as_uint32();

   auto res = state->GetInternedDataMap<MessageType>()->emplace(
       iid,
       ProtoIncrementalState::InternedDataView<MessageType>(std::move(message)));
   // If a message with this ID is already interned, its data should not have
   // changed (this is forbidden by the InternedData proto).
   // TODO(eseckler): This DCHECK assumes that the message is encoded the
   // same way whenever it is re-emitted.
   PERFETTO_DCHECK(res.second ||
                   (res.first->second.message.length() == message_size &&
                    memcmp(res.first->second.message.data(), message_start,
                           message_size) == 0));
 }

 }  // namespace

 ProtoTraceTokenizer::ProtoTraceTokenizer(TraceProcessorContext* ctx)
     : context_(ctx) {}
 ProtoTraceTokenizer::~ProtoTraceTokenizer() = default;

 util::Status ProtoTraceTokenizer::Parse(std::unique_ptr<uint8_t[]> owned_buf,
                                         size_t size) {
   uint8_t* data = &owned_buf[0];
   if (!partial_buf_.empty()) {
     // It takes ~5 bytes for a proto preamble + the varint size.
     const size_t kHeaderBytes = 5;
     if (PERFETTO_UNLIKELY(partial_buf_.size() < kHeaderBytes)) {
       size_t missing_len = std::min(kHeaderBytes - partial_buf_.size(), size);
       partial_buf_.insert(partial_buf_.end(), &data[0], &data[missing_len]);
       if (partial_buf_.size() < kHeaderBytes)
         return util::OkStatus();
       data += missing_len;
       size -= missing_len;
     }

     // At this point we have enough data in |partial_buf_| to read at least the
     // field header and know the size of the next TracePacket.
     constexpr uint8_t kTracePacketTag =
         MakeTagLengthDelimited(protos::pbzero::Trace::kPacketFieldNumber);
     const uint8_t* pos = &partial_buf_[0];
     uint8_t proto_field_tag = *pos;
     uint64_t field_size = 0;
     const uint8_t* next = ParseVarInt(++pos, &*partial_buf_.end(), &field_size);
     bool parse_failed = next == pos;
     pos = next;
     if (proto_field_tag != kTracePacketTag || field_size == 0 || parse_failed) {
       return util::ErrStatus(
           "Failed parsing a TracePacket from the partial buffer");
     }

     // At this point we know how big the TracePacket is.
     size_t hdr_size = static_cast<size_t>(pos - &partial_buf_[0]);
     size_t size_incl_header = static_cast<size_t>(field_size + hdr_size);
     PERFETTO_DCHECK(size_incl_header > partial_buf_.size());

     // There is a good chance that between the |partial_buf_| and the new |data|
     // of the current call we have enough bytes to parse a TracePacket.
     if (partial_buf_.size() + size >= size_incl_header) {
       // Create a new buffer for the whole TracePacket and copy into that:
       // 1) The beginning of the TracePacket (including the proto header) from
       //    the partial buffer.
       // 2) The rest of the TracePacket from the current |data| buffer (note
       //    that we might have consumed already a few bytes form |data| earlier
       //    in this function, hence we need to keep |off| into account).
       std::unique_ptr<uint8_t[]> buf(new uint8_t[size_incl_header]);
       memcpy(&buf[0], partial_buf_.data(), partial_buf_.size());
       // |size_missing| is the number of bytes for the rest of the TracePacket
       // in |data|.
       size_t size_missing = size_incl_header - partial_buf_.size();
       memcpy(&buf[partial_buf_.size()], &data[0], size_missing);
       data += size_missing;
       size -= size_missing;
       partial_buf_.clear();
       uint8_t* buf_start = &buf[0];  // Note that buf is std::moved below.
       util::Status status =
           ParseInternal(std::move(buf), buf_start, size_incl_header);
       if (PERFETTO_UNLIKELY(!status.ok()))
         return status;
     } else {
       partial_buf_.insert(partial_buf_.end(), data, &data[size]);
       return util::OkStatus();
     }
   }
   return ParseInternal(std::move(owned_buf), data, size);
 }

 util::Status ProtoTraceTokenizer::ParseInternal(
     std::unique_ptr<uint8_t[]> owned_buf,
     uint8_t* data,
     size_t size) {
   PERFETTO_DCHECK(data >= &owned_buf[0]);
   const uint8_t* start = &owned_buf[0];
   const size_t data_off = static_cast<size_t>(data - start);
   TraceBlobView whole_buf(std::move(owned_buf), data_off, size);

   protos::pbzero::Trace::Decoder decoder(data, size);
   for (auto it = decoder.packet(); it; ++it) {
     size_t field_offset = whole_buf.offset_of(it->data());
     util::Status status =
         ParsePacket(whole_buf.slice(field_offset, it->size()));
     if (PERFETTO_UNLIKELY(!status.ok()))
       return status;
   }

   const size_t bytes_left = decoder.bytes_left();
   if (bytes_left > 0) {
     PERFETTO_DCHECK(partial_buf_.empty());
     partial_buf_.insert(partial_buf_.end(), &data[decoder.read_offset()],
                         &data[decoder.read_offset() + bytes_left]);
   }
   return util::OkStatus();
 }

 util::Status ProtoTraceTokenizer::ParsePacket(TraceBlobView packet) {
   protos::pbzero::TracePacket::Decoder decoder(packet.data(), packet.length());
   if (PERFETTO_UNLIKELY(decoder.bytes_left()))
     return util::ErrStatus(
         "Failed to parse proto packet fully; the trace is probably corrupt.");

   auto timestamp = decoder.has_timestamp()
                        ? static_cast<int64_t>(decoder.timestamp())
                        : latest_timestamp_;
   latest_timestamp_ = std::max(timestamp, latest_timestamp_);

   if (decoder.incremental_state_cleared()) {
     HandleIncrementalStateCleared(decoder);
   } else if (decoder.previous_packet_dropped()) {
     HandlePreviousPacketDropped(decoder);
   }

   if (decoder.has_interned_data()) {
     auto field = decoder.interned_data();
     const size_t offset = packet.offset_of(field.data);
     ParseInternedData(decoder, packet.slice(offset, field.size));
   }

   if (decoder.has_ftrace_events()) {
     auto ftrace_field = decoder.ftrace_events();
     const size_t fld_off = packet.offset_of(ftrace_field.data);
     ParseFtraceBundle(packet.slice(fld_off, ftrace_field.size));
     return util::OkStatus();
   }

   if (decoder.has_track_event()) {
     ParseTrackEventPacket(decoder, std::move(packet));
     return util::OkStatus();
   }

   if (decoder.has_thread_descriptor()) {
     ParseThreadDescriptorPacket(decoder);
     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 = static_cast<int64_t>(trace_config.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));

   return util::OkStatus();
 }

 void ProtoTraceTokenizer::HandleIncrementalStateCleared(
     const protos::pbzero::TracePacket::Decoder& packet_decoder) {
   if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
     PERFETTO_ELOG(
         "incremental_state_cleared without trusted_packet_sequence_id");
     context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
     return;
   }
   GetIncrementalStateForPacketSequence(
       packet_decoder.trusted_packet_sequence_id())
       ->OnIncrementalStateCleared();
 }

 void ProtoTraceTokenizer::HandlePreviousPacketDropped(
     const protos::pbzero::TracePacket::Decoder& packet_decoder) {
   if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
     PERFETTO_ELOG("previous_packet_dropped without trusted_packet_sequence_id");
     context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
     return;
   }
   GetIncrementalStateForPacketSequence(
       packet_decoder.trusted_packet_sequence_id())
       ->OnPacketLoss();
 }

 void ProtoTraceTokenizer::ParseInternedData(
     const protos::pbzero::TracePacket::Decoder& packet_decoder,
     TraceBlobView interned_data) {
   if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
     PERFETTO_ELOG("InternedData packet without trusted_packet_sequence_id");
     context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
     return;
   }

   auto* state = GetIncrementalStateForPacketSequence(
       packet_decoder.trusted_packet_sequence_id());

   protos::pbzero::InternedData::Decoder interned_data_decoder(
       interned_data.data(), interned_data.length());

   // Store references to interned data submessages into the sequence's state.
   for (auto it = interned_data_decoder.event_categories(); it; ++it) {
     size_t offset = interned_data.offset_of(it->data());
     InternMessage<protos::pbzero::EventCategory>(
         context_, state, interned_data.slice(offset, it->size()));
   }

   for (auto it = interned_data_decoder.legacy_event_names(); it; ++it) {
     size_t offset = interned_data.offset_of(it->data());
     InternMessage<protos::pbzero::LegacyEventName>(
         context_, state, interned_data.slice(offset, it->size()));
   }

   for (auto it = interned_data_decoder.debug_annotation_names(); it; ++it) {
     size_t offset = interned_data.offset_of(it->data());
     InternMessage<protos::pbzero::DebugAnnotationName>(
         context_, state, interned_data.slice(offset, it->size()));
   }

   for (auto it = interned_data_decoder.source_locations(); it; ++it) {
     size_t offset = interned_data.offset_of(it->data());
     InternMessage<protos::pbzero::SourceLocation>(
         context_, state, interned_data.slice(offset, it->size()));
   }
 }

 void ProtoTraceTokenizer::ParseThreadDescriptorPacket(
     const protos::pbzero::TracePacket::Decoder& packet_decoder) {
   if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
     PERFETTO_ELOG("ThreadDescriptor packet without trusted_packet_sequence_id");
     context_->storage->IncrementStats(stats::track_event_tokenizer_errors);
     return;
   }

   auto* state = GetIncrementalStateForPacketSequence(
       packet_decoder.trusted_packet_sequence_id());

   // TrackEvents will be ignored while incremental state is invalid. As a
   // consequence, we should also ignore any ThreadDescriptors received in this
   // state. Otherwise, any delta-encoded timestamps would be calculated
   // incorrectly once we move out of the packet loss state. Instead, wait until
   // the first subsequent descriptor after incremental state is cleared.
   if (!state->IsIncrementalStateValid()) {
     context_->storage->IncrementStats(
         stats::track_event_tokenizer_skipped_packets);
     return;
   }

   auto thread_descriptor_field = packet_decoder.thread_descriptor();
   protos::pbzero::ThreadDescriptor::Decoder thread_descriptor_decoder(
       thread_descriptor_field.data, thread_descriptor_field.size);

   state->SetThreadDescriptor(
       thread_descriptor_decoder.pid(), thread_descriptor_decoder.tid(),
       thread_descriptor_decoder.reference_timestamp_us() * 1000,
       thread_descriptor_decoder.reference_thread_time_us() * 1000);

   base::StringView name;
   if (thread_descriptor_decoder.has_thread_name()) {
     name = thread_descriptor_decoder.thread_name();
   } else if (thread_descriptor_decoder.has_chrome_thread_type()) {
     using protos::pbzero::ThreadDescriptor;
     switch (thread_descriptor_decoder.chrome_thread_type()) {
       case ThreadDescriptor::CHROME_THREAD_MAIN:
         name = "CrProcessMain";
         break;
       case ThreadDescriptor::CHROME_THREAD_IO:
         name = "ChromeIOThread";
         break;
       case ThreadDescriptor::CHROME_THREAD_POOL_FG_WORKER:
         name = "ThreadPoolForegroundWorker&";
         break;
       case ThreadDescriptor::CHROME_THREAD_POOL_BG_WORKER:
         name = "ThreadPoolBackgroundWorker&";
         break;
       case ThreadDescriptor::CHROME_THREAD_POOL_FB_BLOCKING:
         name = "ThreadPoolSingleThreadForegroundBlocking&";
         break;
       case ThreadDescriptor::CHROME_THREAD_POOL_BG_BLOCKING:
         name = "ThreadPoolSingleThreadBackgroundBlocking&";
         break;
       case ThreadDescriptor::CHROME_THREAD_POOL_SERVICE:
         name = "ThreadPoolService";
         break;
       case ThreadDescriptor::CHROME_THREAD_COMPOSITOR_WORKER:
         name = "CompositorTileWorker&";
         break;
       case ThreadDescriptor::CHROME_THREAD_COMPOSITOR:
         name = "Compositor";
         break;
       case ThreadDescriptor::CHROME_THREAD_VIZ_COMPOSITOR:
         name = "VizCompositorThread";
         break;
       case ThreadDescriptor::CHROME_THREAD_SERVICE_WORKER:
         name = "ServiceWorkerThread&";
         break;
       case ThreadDescriptor::CHROME_THREAD_MEMORY_INFRA:
         name = "MemoryInfra";
         break;
       case ThreadDescriptor::CHROME_THREAD_SAMPLING_PROFILER:
         name = "StackSamplingProfiler";
         break;
       case ThreadDescriptor::CHROME_THREAD_UNSPECIFIED:
         name = "ChromeUnspecified";
         break;
     }
   }

   if (!name.empty()) {
     auto thread_name_id = context_->storage->InternString(name);
     ProcessTracker* procs = context_->process_tracker.get();
     procs->UpdateThreadName(
         static_cast<uint32_t>(thread_descriptor_decoder.tid()), thread_name_id);
   }
 }

 void ProtoTraceTokenizer::ParseTrackEventPacket(
     const protos::pbzero::TracePacket::Decoder& packet_decoder,
     TraceBlobView packet) {
   constexpr auto kTimestampDeltaUsFieldNumber =
       protos::pbzero::TrackEvent::kTimestampDeltaUsFieldNumber;
   constexpr auto kTimestampAbsoluteUsFieldNumber =
       protos::pbzero::TrackEvent::kTimestampAbsoluteUsFieldNumber;
   constexpr auto kThreadTimeDeltaUsFieldNumber =
       protos::pbzero::TrackEvent::kThreadTimeDeltaUsFieldNumber;
   constexpr auto kThreadTimeAbsoluteUsFieldNumber =
       protos::pbzero::TrackEvent::kThreadTimeAbsoluteUsFieldNumber;

   if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
     PERFETTO_ELOG("TrackEvent packet without trusted_packet_sequence_id");
     context_->storage->IncrementStats(stats::track_event_tokenizer_errors);
     return;
   }

   auto* state = GetIncrementalStateForPacketSequence(
       packet_decoder.trusted_packet_sequence_id());

   // TrackEvents can only be parsed correctly while incremental state for their
   // sequence is valid and after a ThreadDescriptor has been parsed.
   if (!state->IsTrackEventStateValid()) {
     context_->storage->IncrementStats(
         stats::track_event_tokenizer_skipped_packets);
     return;
   }

   auto field = packet_decoder.track_event();
   ProtoDecoder event_decoder(field.data, field.size);

   int64_t timestamp;
   int64_t thread_timestamp = 0;

   if (auto ts_delta_field =
           event_decoder.FindField(kTimestampDeltaUsFieldNumber)) {
     timestamp = state->IncrementAndGetTrackEventTimeNs(
         ts_delta_field.as_int64() * 1000);
   } else if (auto ts_absolute_field =
                  event_decoder.FindField(kTimestampAbsoluteUsFieldNumber)) {
     // One-off absolute timestamps don't affect delta computation.
     timestamp = ts_absolute_field.as_int64() * 1000;
   } else {
     PERFETTO_ELOG("TrackEvent without timestamp");
     context_->storage->IncrementStats(stats::track_event_tokenizer_errors);
     return;
   }

   if (auto tt_delta_field =
           event_decoder.FindField(kThreadTimeDeltaUsFieldNumber)) {
     thread_timestamp = state->IncrementAndGetTrackEventThreadTimeNs(
         tt_delta_field.as_int64() * 1000);
   } else if (auto tt_absolute_field =
                  event_decoder.FindField(kThreadTimeAbsoluteUsFieldNumber)) {
     // One-off absolute timestamps don't affect delta computation.
     thread_timestamp = tt_absolute_field.as_int64() * 1000;
   }

   context_->sorter->PushTrackEventPacket(timestamp, thread_timestamp, state,
                                          std::move(packet));
 }

 PERFETTO_ALWAYS_INLINE
 void ProtoTraceTokenizer::ParseFtraceBundle(TraceBlobView bundle) {
   protos::pbzero::FtraceEventBundle::Decoder decoder(bundle.data(),
                                                      bundle.length());

   if (PERFETTO_UNLIKELY(!decoder.has_cpu())) {
     PERFETTO_ELOG("CPU field not found in FtraceEventBundle");
     context_->storage->IncrementStats(stats::ftrace_bundle_tokenizer_errors);
     return;
   }

   uint32_t cpu = decoder.cpu();
   if (PERFETTO_UNLIKELY(cpu > base::kMaxCpus)) {
     PERFETTO_ELOG("CPU larger than kMaxCpus (%u > %zu)", cpu, base::kMaxCpus);
     return;
   }

   for (auto it = decoder.event(); it; ++it) {
     size_t off = bundle.offset_of(it->data());
     ParseFtraceEvent(cpu, bundle.slice(off, it->size()));
   }
   context_->sorter->FinalizeFtraceEventBatch(cpu);
 }

 PERFETTO_ALWAYS_INLINE
 void ProtoTraceTokenizer::ParseFtraceEvent(uint32_t cpu, TraceBlobView event) {
   constexpr auto kTimestampFieldNumber =
       protos::pbzero::FtraceEvent::kTimestampFieldNumber;
   const uint8_t* data = event.data();
   const size_t length = event.length();
   ProtoDecoder decoder(data, length);
   uint64_t raw_timestamp = 0;
   bool timestamp_found = false;

   // Speculate on the fact that the timestamp is often the 1st field of the
   // event.
   constexpr auto timestampFieldTag = MakeTagVarInt(kTimestampFieldNumber);
   if (PERFETTO_LIKELY(length > 10 && data[0] == timestampFieldTag)) {
     // Fastpath.
     const uint8_t* next = ParseVarInt(data + 1, data + 11, &raw_timestamp);
     timestamp_found = next != data + 1;
     decoder.Reset(next);
   } else {
     // Slowpath.
     if (auto ts_field = decoder.FindField(kTimestampFieldNumber)) {
       timestamp_found = true;
       raw_timestamp = ts_field.as_uint64();
     }
   }

   if (PERFETTO_UNLIKELY(!timestamp_found)) {
     PERFETTO_ELOG("Timestamp field not found in FtraceEvent");
     context_->storage->IncrementStats(stats::ftrace_bundle_tokenizer_errors);
     return;
   }

   int64_t timestamp = static_cast<int64_t>(raw_timestamp);
   latest_timestamp_ = std::max(timestamp, latest_timestamp_);

   // We don't need to parse this packet, just push it to be sorted with
   // the timestamp.
   context_->sorter->PushFtraceEvent(cpu, timestamp, std::move(event));
 }

 }  // namespace trace_processor
 }  // namespace perfetto
	/*
	* Copyright (C) 2018 The Android Open Source Project
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	#include "src/trace_processor/proto_trace_tokenizer.h"

	#include <string>

	#include "perfetto/base/logging.h"
	#include "perfetto/ext/base/utils.h"
	#include "perfetto/protozero/proto_decoder.h"
	#include "perfetto/protozero/proto_utils.h"
	#include "src/trace_processor/event_tracker.h"
	#include "src/trace_processor/process_tracker.h"
	#include "src/trace_processor/stats.h"
	#include "src/trace_processor/trace_blob_view.h"
	#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"
	#include "perfetto/trace/trace.pbzero.h"
	#include "perfetto/trace/track_event/task_execution.pbzero.h"
	#include "perfetto/trace/track_event/thread_descriptor.pbzero.h"
	#include "perfetto/trace/track_event/track_event.pbzero.h"

	namespace perfetto {
	namespace trace_processor {

	using protozero::ProtoDecoder;
	using protozero::proto_utils::MakeTagLengthDelimited;
	using protozero::proto_utils::MakeTagVarInt;
	using protozero::proto_utils::ParseVarInt;

	namespace {

	template <typename MessageType>
	void InternMessage(TraceProcessorContext* context,
	ProtoIncrementalState::PacketSequenceState* state,
	TraceBlobView message) {
	constexpr auto kIidFieldNumber = MessageType::kIidFieldNumber;

	uint32_t iid = 0;
	auto message_start = message.data();
	auto message_size = message.length();
	protozero::ProtoDecoder decoder(message_start, message_size);

	auto field = decoder.FindField(kIidFieldNumber);
	if (PERFETTO_UNLIKELY(!field)) {
	PERFETTO_ELOG("Interned message without interning_id");
	context->storage->IncrementStats(stats::interned_data_tokenizer_errors);
	return;
	}
	iid = field.as_uint32();

	auto res = state->GetInternedDataMap<MessageType>()->emplace(
	iid,
	ProtoIncrementalState::InternedDataView<MessageType>(std::move(message)));
	// If a message with this ID is already interned, its data should not have
	// changed (this is forbidden by the InternedData proto).
	// TODO(eseckler): This DCHECK assumes that the message is encoded the
	// same way whenever it is re-emitted.
	PERFETTO_DCHECK(res.second \|\|
	(res.first->second.message.length() == message_size &&
	memcmp(res.first->second.message.data(), message_start,
	message_size) == 0));
	}

	} // namespace

	ProtoTraceTokenizer::ProtoTraceTokenizer(TraceProcessorContext* ctx)
	: context_(ctx) {}
	ProtoTraceTokenizer::~ProtoTraceTokenizer() = default;

	util::Status ProtoTraceTokenizer::Parse(std::unique_ptr<uint8_t[]> owned_buf,
	size_t size) {
	uint8_t* data = &owned_buf[0];
	if (!partial_buf_.empty()) {
	// It takes ~5 bytes for a proto preamble + the varint size.
	const size_t kHeaderBytes = 5;
	if (PERFETTO_UNLIKELY(partial_buf_.size() < kHeaderBytes)) {
	size_t missing_len = std::min(kHeaderBytes - partial_buf_.size(), size);
	partial_buf_.insert(partial_buf_.end(), &data[0], &data[missing_len]);
	if (partial_buf_.size() < kHeaderBytes)
	return util::OkStatus();
	data += missing_len;
	size -= missing_len;
	}

	// At this point we have enough data in \|partial_buf_\| to read at least the
	// field header and know the size of the next TracePacket.
	constexpr uint8_t kTracePacketTag =
	MakeTagLengthDelimited(protos::pbzero::Trace::kPacketFieldNumber);
	const uint8_t* pos = &partial_buf_[0];
	uint8_t proto_field_tag = *pos;
	uint64_t field_size = 0;
	const uint8_t* next = ParseVarInt(++pos, &*partial_buf_.end(), &field_size);
	bool parse_failed = next == pos;
	pos = next;
	if (proto_field_tag != kTracePacketTag \|\| field_size == 0 \|\| parse_failed) {
	return util::ErrStatus(
	"Failed parsing a TracePacket from the partial buffer");
	}

	// At this point we know how big the TracePacket is.
	size_t hdr_size = static_cast<size_t>(pos - &partial_buf_[0]);
	size_t size_incl_header = static_cast<size_t>(field_size + hdr_size);
	PERFETTO_DCHECK(size_incl_header > partial_buf_.size());

	// There is a good chance that between the \|partial_buf_\| and the new \|data\|
	// of the current call we have enough bytes to parse a TracePacket.
	if (partial_buf_.size() + size >= size_incl_header) {
	// Create a new buffer for the whole TracePacket and copy into that:
	// 1) The beginning of the TracePacket (including the proto header) from
	// the partial buffer.
	// 2) The rest of the TracePacket from the current \|data\| buffer (note
	// that we might have consumed already a few bytes form \|data\| earlier
	// in this function, hence we need to keep \|off\| into account).
	std::unique_ptr<uint8_t[]> buf(new uint8_t[size_incl_header]);
	memcpy(&buf[0], partial_buf_.data(), partial_buf_.size());
	// \|size_missing\| is the number of bytes for the rest of the TracePacket
	// in \|data\|.
	size_t size_missing = size_incl_header - partial_buf_.size();
	memcpy(&buf[partial_buf_.size()], &data[0], size_missing);
	data += size_missing;
	size -= size_missing;
	partial_buf_.clear();
	uint8_t* buf_start = &buf[0]; // Note that buf is std::moved below.
	util::Status status =
	ParseInternal(std::move(buf), buf_start, size_incl_header);
	if (PERFETTO_UNLIKELY(!status.ok()))
	return status;
	} else {
	partial_buf_.insert(partial_buf_.end(), data, &data[size]);
	return util::OkStatus();
	}
	}
	return ParseInternal(std::move(owned_buf), data, size);
	}

	util::Status ProtoTraceTokenizer::ParseInternal(
	std::unique_ptr<uint8_t[]> owned_buf,
	uint8_t* data,
	size_t size) {
	PERFETTO_DCHECK(data >= &owned_buf[0]);
	const uint8_t* start = &owned_buf[0];
	const size_t data_off = static_cast<size_t>(data - start);
	TraceBlobView whole_buf(std::move(owned_buf), data_off, size);

	protos::pbzero::Trace::Decoder decoder(data, size);
	for (auto it = decoder.packet(); it; ++it) {
	size_t field_offset = whole_buf.offset_of(it->data());
	util::Status status =
	ParsePacket(whole_buf.slice(field_offset, it->size()));
	if (PERFETTO_UNLIKELY(!status.ok()))
	return status;
	}

	const size_t bytes_left = decoder.bytes_left();
	if (bytes_left > 0) {
	PERFETTO_DCHECK(partial_buf_.empty());
	partial_buf_.insert(partial_buf_.end(), &data[decoder.read_offset()],
	&data[decoder.read_offset() + bytes_left]);
	}
	return util::OkStatus();
	}

	util::Status ProtoTraceTokenizer::ParsePacket(TraceBlobView packet) {
	protos::pbzero::TracePacket::Decoder decoder(packet.data(), packet.length());
	if (PERFETTO_UNLIKELY(decoder.bytes_left()))
	return util::ErrStatus(
	"Failed to parse proto packet fully; the trace is probably corrupt.");

	auto timestamp = decoder.has_timestamp()
	? static_cast<int64_t>(decoder.timestamp())
	: latest_timestamp_;
	latest_timestamp_ = std::max(timestamp, latest_timestamp_);

	if (decoder.incremental_state_cleared()) {
	HandleIncrementalStateCleared(decoder);
	} else if (decoder.previous_packet_dropped()) {
	HandlePreviousPacketDropped(decoder);
	}

	if (decoder.has_interned_data()) {
	auto field = decoder.interned_data();
	const size_t offset = packet.offset_of(field.data);
	ParseInternedData(decoder, packet.slice(offset, field.size));
	}

	if (decoder.has_ftrace_events()) {
	auto ftrace_field = decoder.ftrace_events();
	const size_t fld_off = packet.offset_of(ftrace_field.data);
	ParseFtraceBundle(packet.slice(fld_off, ftrace_field.size));
	return util::OkStatus();
	}

	if (decoder.has_track_event()) {
	ParseTrackEventPacket(decoder, std::move(packet));
	return util::OkStatus();
	}

	if (decoder.has_thread_descriptor()) {
	ParseThreadDescriptorPacket(decoder);
	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 = static_cast<int64_t>(trace_config.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));

	return util::OkStatus();
	}

	void ProtoTraceTokenizer::HandleIncrementalStateCleared(
	const protos::pbzero::TracePacket::Decoder& packet_decoder) {
	if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
	PERFETTO_ELOG(
	"incremental_state_cleared without trusted_packet_sequence_id");
	context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
	return;
	}
	GetIncrementalStateForPacketSequence(
	packet_decoder.trusted_packet_sequence_id())
	->OnIncrementalStateCleared();
	}

	void ProtoTraceTokenizer::HandlePreviousPacketDropped(
	const protos::pbzero::TracePacket::Decoder& packet_decoder) {
	if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
	PERFETTO_ELOG("previous_packet_dropped without trusted_packet_sequence_id");
	context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
	return;
	}
	GetIncrementalStateForPacketSequence(
	packet_decoder.trusted_packet_sequence_id())
	->OnPacketLoss();
	}

	void ProtoTraceTokenizer::ParseInternedData(
	const protos::pbzero::TracePacket::Decoder& packet_decoder,
	TraceBlobView interned_data) {
	if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
	PERFETTO_ELOG("InternedData packet without trusted_packet_sequence_id");
	context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
	return;
	}

	auto* state = GetIncrementalStateForPacketSequence(
	packet_decoder.trusted_packet_sequence_id());

	protos::pbzero::InternedData::Decoder interned_data_decoder(
	interned_data.data(), interned_data.length());

	// Store references to interned data submessages into the sequence's state.
	for (auto it = interned_data_decoder.event_categories(); it; ++it) {
	size_t offset = interned_data.offset_of(it->data());
	InternMessage<protos::pbzero::EventCategory>(
	context_, state, interned_data.slice(offset, it->size()));
	}

	for (auto it = interned_data_decoder.legacy_event_names(); it; ++it) {
	size_t offset = interned_data.offset_of(it->data());
	InternMessage<protos::pbzero::LegacyEventName>(
	context_, state, interned_data.slice(offset, it->size()));
	}

	for (auto it = interned_data_decoder.debug_annotation_names(); it; ++it) {
	size_t offset = interned_data.offset_of(it->data());
	InternMessage<protos::pbzero::DebugAnnotationName>(
	context_, state, interned_data.slice(offset, it->size()));
	}

	for (auto it = interned_data_decoder.source_locations(); it; ++it) {
	size_t offset = interned_data.offset_of(it->data());
	InternMessage<protos::pbzero::SourceLocation>(
	context_, state, interned_data.slice(offset, it->size()));
	}
	}

	void ProtoTraceTokenizer::ParseThreadDescriptorPacket(
	const protos::pbzero::TracePacket::Decoder& packet_decoder) {
	if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
	PERFETTO_ELOG("ThreadDescriptor packet without trusted_packet_sequence_id");
	context_->storage->IncrementStats(stats::track_event_tokenizer_errors);
	return;
	}

	auto* state = GetIncrementalStateForPacketSequence(
	packet_decoder.trusted_packet_sequence_id());

	// TrackEvents will be ignored while incremental state is invalid. As a
	// consequence, we should also ignore any ThreadDescriptors received in this
	// state. Otherwise, any delta-encoded timestamps would be calculated
	// incorrectly once we move out of the packet loss state. Instead, wait until
	// the first subsequent descriptor after incremental state is cleared.
	if (!state->IsIncrementalStateValid()) {
	context_->storage->IncrementStats(
	stats::track_event_tokenizer_skipped_packets);
	return;
	}

	auto thread_descriptor_field = packet_decoder.thread_descriptor();
	protos::pbzero::ThreadDescriptor::Decoder thread_descriptor_decoder(
	thread_descriptor_field.data, thread_descriptor_field.size);

	state->SetThreadDescriptor(
	thread_descriptor_decoder.pid(), thread_descriptor_decoder.tid(),
	thread_descriptor_decoder.reference_timestamp_us() * 1000,
	thread_descriptor_decoder.reference_thread_time_us() * 1000);

	base::StringView name;
	if (thread_descriptor_decoder.has_thread_name()) {
	name = thread_descriptor_decoder.thread_name();
	} else if (thread_descriptor_decoder.has_chrome_thread_type()) {
	using protos::pbzero::ThreadDescriptor;
	switch (thread_descriptor_decoder.chrome_thread_type()) {
	case ThreadDescriptor::CHROME_THREAD_MAIN:
	name = "CrProcessMain";
	break;
	case ThreadDescriptor::CHROME_THREAD_IO:
	name = "ChromeIOThread";
	break;
	case ThreadDescriptor::CHROME_THREAD_POOL_FG_WORKER:
	name = "ThreadPoolForegroundWorker&";
	break;
	case ThreadDescriptor::CHROME_THREAD_POOL_BG_WORKER:
	name = "ThreadPoolBackgroundWorker&";
	break;
	case ThreadDescriptor::CHROME_THREAD_POOL_FB_BLOCKING:
	name = "ThreadPoolSingleThreadForegroundBlocking&";
	break;
	case ThreadDescriptor::CHROME_THREAD_POOL_BG_BLOCKING:
	name = "ThreadPoolSingleThreadBackgroundBlocking&";
	break;
	case ThreadDescriptor::CHROME_THREAD_POOL_SERVICE:
	name = "ThreadPoolService";
	break;
	case ThreadDescriptor::CHROME_THREAD_COMPOSITOR_WORKER:
	name = "CompositorTileWorker&";
	break;
	case ThreadDescriptor::CHROME_THREAD_COMPOSITOR:
	name = "Compositor";
	break;
	case ThreadDescriptor::CHROME_THREAD_VIZ_COMPOSITOR:
	name = "VizCompositorThread";
	break;
	case ThreadDescriptor::CHROME_THREAD_SERVICE_WORKER:
	name = "ServiceWorkerThread&";
	break;
	case ThreadDescriptor::CHROME_THREAD_MEMORY_INFRA:
	name = "MemoryInfra";
	break;
	case ThreadDescriptor::CHROME_THREAD_SAMPLING_PROFILER:
	name = "StackSamplingProfiler";
	break;
	case ThreadDescriptor::CHROME_THREAD_UNSPECIFIED:
	name = "ChromeUnspecified";
	break;
	}
	}

	if (!name.empty()) {
	auto thread_name_id = context_->storage->InternString(name);
	ProcessTracker* procs = context_->process_tracker.get();
	procs->UpdateThreadName(
	static_cast<uint32_t>(thread_descriptor_decoder.tid()), thread_name_id);
	}
	}

	void ProtoTraceTokenizer::ParseTrackEventPacket(
	const protos::pbzero::TracePacket::Decoder& packet_decoder,
	TraceBlobView packet) {
	constexpr auto kTimestampDeltaUsFieldNumber =
	protos::pbzero::TrackEvent::kTimestampDeltaUsFieldNumber;
	constexpr auto kTimestampAbsoluteUsFieldNumber =
	protos::pbzero::TrackEvent::kTimestampAbsoluteUsFieldNumber;
	constexpr auto kThreadTimeDeltaUsFieldNumber =
	protos::pbzero::TrackEvent::kThreadTimeDeltaUsFieldNumber;
	constexpr auto kThreadTimeAbsoluteUsFieldNumber =
	protos::pbzero::TrackEvent::kThreadTimeAbsoluteUsFieldNumber;

	if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
	PERFETTO_ELOG("TrackEvent packet without trusted_packet_sequence_id");
	context_->storage->IncrementStats(stats::track_event_tokenizer_errors);
	return;
	}

	auto* state = GetIncrementalStateForPacketSequence(
	packet_decoder.trusted_packet_sequence_id());

	// TrackEvents can only be parsed correctly while incremental state for their
	// sequence is valid and after a ThreadDescriptor has been parsed.
	if (!state->IsTrackEventStateValid()) {
	context_->storage->IncrementStats(
	stats::track_event_tokenizer_skipped_packets);
	return;
	}

	auto field = packet_decoder.track_event();
	ProtoDecoder event_decoder(field.data, field.size);

	int64_t timestamp;
	int64_t thread_timestamp = 0;

	if (auto ts_delta_field =
	event_decoder.FindField(kTimestampDeltaUsFieldNumber)) {
	timestamp = state->IncrementAndGetTrackEventTimeNs(
	ts_delta_field.as_int64() * 1000);
	} else if (auto ts_absolute_field =
	event_decoder.FindField(kTimestampAbsoluteUsFieldNumber)) {
	// One-off absolute timestamps don't affect delta computation.
	timestamp = ts_absolute_field.as_int64() * 1000;
	} else {
	PERFETTO_ELOG("TrackEvent without timestamp");
	context_->storage->IncrementStats(stats::track_event_tokenizer_errors);
	return;
	}

	if (auto tt_delta_field =
	event_decoder.FindField(kThreadTimeDeltaUsFieldNumber)) {
	thread_timestamp = state->IncrementAndGetTrackEventThreadTimeNs(
	tt_delta_field.as_int64() * 1000);
	} else if (auto tt_absolute_field =
	event_decoder.FindField(kThreadTimeAbsoluteUsFieldNumber)) {
	// One-off absolute timestamps don't affect delta computation.
	thread_timestamp = tt_absolute_field.as_int64() * 1000;
	}

	context_->sorter->PushTrackEventPacket(timestamp, thread_timestamp, state,
	std::move(packet));
	}

	PERFETTO_ALWAYS_INLINE
	void ProtoTraceTokenizer::ParseFtraceBundle(TraceBlobView bundle) {
	protos::pbzero::FtraceEventBundle::Decoder decoder(bundle.data(),
	bundle.length());

	if (PERFETTO_UNLIKELY(!decoder.has_cpu())) {
	PERFETTO_ELOG("CPU field not found in FtraceEventBundle");
	context_->storage->IncrementStats(stats::ftrace_bundle_tokenizer_errors);
	return;
	}

	uint32_t cpu = decoder.cpu();
	if (PERFETTO_UNLIKELY(cpu > base::kMaxCpus)) {
	PERFETTO_ELOG("CPU larger than kMaxCpus (%u > %zu)", cpu, base::kMaxCpus);
	return;
	}

	for (auto it = decoder.event(); it; ++it) {
	size_t off = bundle.offset_of(it->data());
	ParseFtraceEvent(cpu, bundle.slice(off, it->size()));
	}
	context_->sorter->FinalizeFtraceEventBatch(cpu);
	}

	PERFETTO_ALWAYS_INLINE
	void ProtoTraceTokenizer::ParseFtraceEvent(uint32_t cpu, TraceBlobView event) {
	constexpr auto kTimestampFieldNumber =
	protos::pbzero::FtraceEvent::kTimestampFieldNumber;
	const uint8_t* data = event.data();
	const size_t length = event.length();
	ProtoDecoder decoder(data, length);
	uint64_t raw_timestamp = 0;
	bool timestamp_found = false;

	// Speculate on the fact that the timestamp is often the 1st field of the
	// event.
	constexpr auto timestampFieldTag = MakeTagVarInt(kTimestampFieldNumber);
	if (PERFETTO_LIKELY(length > 10 && data[0] == timestampFieldTag)) {
	// Fastpath.
	const uint8_t* next = ParseVarInt(data + 1, data + 11, &raw_timestamp);
	timestamp_found = next != data + 1;
	decoder.Reset(next);
	} else {
	// Slowpath.
	if (auto ts_field = decoder.FindField(kTimestampFieldNumber)) {
	timestamp_found = true;
	raw_timestamp = ts_field.as_uint64();
	}
	}

	if (PERFETTO_UNLIKELY(!timestamp_found)) {
	PERFETTO_ELOG("Timestamp field not found in FtraceEvent");
	context_->storage->IncrementStats(stats::ftrace_bundle_tokenizer_errors);
	return;
	}

	int64_t timestamp = static_cast<int64_t>(raw_timestamp);
	latest_timestamp_ = std::max(timestamp, latest_timestamp_);

	// We don't need to parse this packet, just push it to be sorted with
	// the timestamp.
	context_->sorter->PushFtraceEvent(cpu, timestamp, std::move(event));
	}

	} // namespace trace_processor
	} // namespace perfetto