src/trace_processor/importers/proto/proto_trace_tokenizer.cc - platform/external/perfetto - Gitiles

 /*
  * 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/importers/proto/proto_trace_tokenizer.h"

 #include <string>

 #include <zlib.h>

 #include "perfetto/base/logging.h"
 #include "perfetto/ext/base/optional.h"
 #include "perfetto/ext/base/string_view.h"
 #include "perfetto/ext/base/utils.h"
 #include "perfetto/protozero/proto_decoder.h"
 #include "perfetto/protozero/proto_utils.h"
 #include "perfetto/trace_processor/status.h"
 #include "src/trace_processor/clock_tracker.h"
 #include "src/trace_processor/event_tracker.h"
 #include "src/trace_processor/importers/ftrace/ftrace_module.h"
 #include "src/trace_processor/importers/proto/packet_sequence_state.h"
 #include "src/trace_processor/importers/proto/proto_incremental_state.h"
 #include "src/trace_processor/importers/proto/track_event_module.h"
 #include "src/trace_processor/stats.h"
 #include "src/trace_processor/trace_sorter.h"
 #include "src/trace_processor/trace_storage.h"

 #include "protos/perfetto/config/trace_config.pbzero.h"
 #include "protos/perfetto/trace/clock_snapshot.pbzero.h"
 #include "protos/perfetto/trace/profiling/profile_common.pbzero.h"
 #include "protos/perfetto/trace/trace.pbzero.h"
 #include "protos/perfetto/trace/trace_packet.pbzero.h"

 namespace perfetto {
 namespace trace_processor {

 using protozero::proto_utils::MakeTagLengthDelimited;
 using protozero::proto_utils::ParseVarInt;

 namespace {

 constexpr uint8_t kTracePacketTag =
     MakeTagLengthDelimited(protos::pbzero::Trace::kPacketFieldNumber);

 TraceBlobView Decompress(TraceBlobView input) {
   uint8_t out[4096];
   std::string s;

   z_stream stream{};
   stream.next_in = const_cast<uint8_t*>(input.data());
   stream.avail_in = static_cast<unsigned int>(input.length());

   if (inflateInit(&stream) != Z_OK)
     return TraceBlobView(nullptr, 0, 0);

   int ret;
   do {
     stream.next_out = out;
     stream.avail_out = sizeof(out);
     ret = inflate(&stream, Z_NO_FLUSH);
     if (ret != Z_STREAM_END && ret != Z_OK) {
       inflateEnd(&stream);
       return TraceBlobView(nullptr, 0, 0);
     }
     s.append(reinterpret_cast<char*>(out), sizeof(out) - stream.avail_out);
   } while (ret != Z_STREAM_END);
   inflateEnd(&stream);

   std::unique_ptr<uint8_t[]> output(new uint8_t[s.size()]);
   memcpy(output.get(), s.data(), s.size());
   return TraceBlobView(std::move(output), 0, s.size());
 }

 }  // 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.
     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) {
     protozero::ConstBytes packet = *it;
     size_t field_offset = whole_buf.offset_of(packet.data);
     util::Status status =
         ParsePacket(whole_buf.slice(field_offset, packet.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())
           : std::max(latest_timestamp_, context_->sorter->max_timestamp());

   const uint32_t seq_id = decoder.trusted_packet_sequence_id();

   if ((decoder.has_chrome_events() || decoder.has_chrome_metadata()) &&
       (!decoder.timestamp_clock_id() ||
        decoder.timestamp_clock_id() ==
            protos::pbzero::ClockSnapshot::Clock::MONOTONIC)) {
     // Chrome event timestamps are in MONOTONIC domain, but may occur in traces
     // where (a) no clock snapshots exist or (b) no clock_id is specified for
     // their timestamps. Adjust to trace time if we have a clock snapshot.
     // TODO(eseckler): Set timestamp_clock_id and emit ClockSnapshots in chrome
     // and then remove this.
     auto trace_ts = context_->clock_tracker->ToTraceTime(
         protos::pbzero::ClockSnapshot::Clock::MONOTONIC, timestamp);
     if (trace_ts.has_value())
       timestamp = trace_ts.value();
   } else if (decoder.timestamp_clock_id()) {
     // If the TracePacket specifies a non-zero clock-id, translate the timestamp
     // into the trace-time clock domain.
     PERFETTO_DCHECK(decoder.has_timestamp());
     ClockTracker::ClockId clock_id = decoder.timestamp_clock_id();
     bool is_seq_scoped = ClockTracker::IsReservedSeqScopedClockId(clock_id);
     if (is_seq_scoped) {
       if (!seq_id) {
         return util::ErrStatus(
             "TracePacket specified a sequence-local clock id (%" PRIu32
             ") but the TraceWriter's sequence_id is zero (the service is "
             "probably too old)",
             decoder.timestamp_clock_id());
       }
       clock_id = ClockTracker::SeqScopedClockIdToGlobal(
           seq_id, decoder.timestamp_clock_id());
     }
     auto trace_ts = context_->clock_tracker->ToTraceTime(clock_id, timestamp);
     if (!trace_ts.has_value()) {
       // ToTraceTime() will increase the |clock_sync_failure| stat on failure.
       static const char seq_extra_err[] =
           " Because the clock id is sequence-scoped, the ClockSnapshot must be "
           "emitted on the same TraceWriter sequence of the packet that refers "
           "to that clock id.";
       return util::ErrStatus(
           "Failed to convert TracePacket's timestamp from clock_id=%" PRIu32
           " seq_id=%" PRIu32
           ". This is usually due to the lack of a prior ClockSnapshot proto.%s",
           decoder.timestamp_clock_id(), seq_id,
           is_seq_scoped ? seq_extra_err : "");
     }
     timestamp = trace_ts.value();
   }
   latest_timestamp_ = std::max(timestamp, latest_timestamp_);

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

   uint32_t sequence_flags = decoder.sequence_flags();

   if (decoder.incremental_state_cleared() ||
       sequence_flags &
           protos::pbzero::TracePacket::SEQ_INCREMENTAL_STATE_CLEARED) {
     HandleIncrementalStateCleared(decoder);
   } else if (decoder.previous_packet_dropped()) {
     HandlePreviousPacketDropped(decoder);
   }

   if (decoder.sequence_flags() &
       protos::pbzero::TracePacket::SEQ_NEEDS_INCREMENTAL_STATE) {
     if (!seq_id) {
       return util::ErrStatus(
           "TracePacket specified SEQ_NEEDS_INCREMENTAL_STATE but the "
           "TraceWriter's sequence_id is zero (the service is "
           "probably too old)");
     }

     if (!state->IsIncrementalStateValid()) {
       context_->storage->IncrementStats(stats::tokenizer_skipped_packets);
       return util::OkStatus();
     }
   }

   if (decoder.has_clock_snapshot()) {
     return ParseClockSnapshot(decoder.clock_snapshot(),
                               decoder.trusted_packet_sequence_id());
   }

   // TODO(eseckler): Parse TracePacketDefaults.

   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));
   }

   ModuleResult res = ModuleResult::Ignored();
   res = context_->ftrace_module->TokenizePacket(decoder, &packet, timestamp,
                                                 state);
   if (!res.ignored())
     return res.ToStatus();

   res = context_->track_event_module->TokenizePacket(decoder, &packet,
                                                      timestamp, state);
   if (!res.ignored())
     return res.ToStatus();

   auto& modules = context_->modules_by_field;
   for (uint32_t field_id = 1; field_id < modules.size(); ++field_id) {
     if (modules[field_id] && decoder.Get(field_id).valid()) {
       modules[field_id]->TokenizePacket(decoder, &packet, timestamp, state,
                                         field_id);
       if (!res.ignored())
         return res.ToStatus();
     }
   }

   if (decoder.has_compressed_packets()) {
     protozero::ConstBytes field = decoder.compressed_packets();
     const size_t field_off = packet.offset_of(field.data);
     TraceBlobView compressed_packets = packet.slice(field_off, field.size);
     TraceBlobView packets = Decompress(std::move(compressed_packets));

     const uint8_t* start = packets.data();
     const uint8_t* end = packets.data() + packets.length();
     const uint8_t* ptr = start;
     while ((end - ptr) > 2) {
       const uint8_t* packet_start = ptr;
       if (PERFETTO_UNLIKELY(*ptr != kTracePacketTag))
         return util::ErrStatus("Expected TracePacket tag");
       uint64_t packet_size = 0;
       ptr = ParseVarInt(++ptr, end, &packet_size);
       size_t packet_offset = static_cast<size_t>(ptr - start);
       ptr += packet_size;
       if (PERFETTO_UNLIKELY((ptr - packet_start) < 2 || ptr > end))
         return util::ErrStatus("Invalid packet size");
       util::Status status = ParsePacket(
           packets.slice(packet_offset, static_cast<size_t>(packet_size)));
       if (PERFETTO_UNLIKELY(!status.ok()))
         return status;
     }

     return util::OkStatus();
   }

   // If we're not forcing a full sort and this is a write_into_file trace, then
   // use flush_period_ms as an indiciator for how big the sliding window for the
   // sorter should be.
   if (!context_->config.force_full_sort && 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, state, 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());

   // Don't parse interned data entries until incremental state is valid, because
   // they could otherwise be associated with the wrong generation in the state.
   if (!state->IsIncrementalStateValid()) {
     context_->storage->IncrementStats(stats::tokenizer_skipped_packets);
     return;
   }

   // Store references to interned data submessages into the sequence's state.
   protozero::ProtoDecoder decoder(interned_data.data(), interned_data.length());
   for (protozero::Field f = decoder.ReadField(); f.valid();
        f = decoder.ReadField()) {
     auto bytes = f.as_bytes();
     auto offset = interned_data.offset_of(bytes.data);
     state->InternMessage(f.id(), interned_data.slice(offset, bytes.size));
   }
 }

 util::Status ProtoTraceTokenizer::ParseClockSnapshot(ConstBytes blob,
                                                      uint32_t seq_id) {
   std::vector<ClockTracker::ClockValue> clocks;
   protos::pbzero::ClockSnapshot::Decoder evt(blob.data, blob.size);
   for (auto it = evt.clocks(); it; ++it) {
     protos::pbzero::ClockSnapshot::Clock::Decoder clk(*it);
     ClockTracker::ClockId clock_id = clk.clock_id();
     if (ClockTracker::IsReservedSeqScopedClockId(clk.clock_id())) {
       if (!seq_id) {
         return util::ErrStatus(
             "ClockSnapshot packet is specifying a sequence-scoped clock id "
             "(%" PRIu64 ") but the TracePacket sequence_id is zero",
             clock_id);
       }
       clock_id = ClockTracker::SeqScopedClockIdToGlobal(seq_id, clk.clock_id());
     }
     int64_t unit_multiplier_ns =
         clk.unit_multiplier_ns()
             ? static_cast<int64_t>(clk.unit_multiplier_ns())
             : 1;
     clocks.emplace_back(clock_id, clk.timestamp(), unit_multiplier_ns,
                         clk.is_incremental());
   }
   context_->clock_tracker->AddSnapshot(clocks);
   return util::OkStatus();
 }

 }  // 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/importers/proto/proto_trace_tokenizer.h"

	#include <string>

	#include <zlib.h>

	#include "perfetto/base/logging.h"
	#include "perfetto/ext/base/optional.h"
	#include "perfetto/ext/base/string_view.h"
	#include "perfetto/ext/base/utils.h"
	#include "perfetto/protozero/proto_decoder.h"
	#include "perfetto/protozero/proto_utils.h"
	#include "perfetto/trace_processor/status.h"
	#include "src/trace_processor/clock_tracker.h"
	#include "src/trace_processor/event_tracker.h"
	#include "src/trace_processor/importers/ftrace/ftrace_module.h"
	#include "src/trace_processor/importers/proto/packet_sequence_state.h"
	#include "src/trace_processor/importers/proto/proto_incremental_state.h"
	#include "src/trace_processor/importers/proto/track_event_module.h"
	#include "src/trace_processor/stats.h"
	#include "src/trace_processor/trace_sorter.h"
	#include "src/trace_processor/trace_storage.h"

	#include "protos/perfetto/config/trace_config.pbzero.h"
	#include "protos/perfetto/trace/clock_snapshot.pbzero.h"
	#include "protos/perfetto/trace/profiling/profile_common.pbzero.h"
	#include "protos/perfetto/trace/trace.pbzero.h"
	#include "protos/perfetto/trace/trace_packet.pbzero.h"

	namespace perfetto {
	namespace trace_processor {

	using protozero::proto_utils::MakeTagLengthDelimited;
	using protozero::proto_utils::ParseVarInt;

	namespace {

	constexpr uint8_t kTracePacketTag =
	MakeTagLengthDelimited(protos::pbzero::Trace::kPacketFieldNumber);

	TraceBlobView Decompress(TraceBlobView input) {
	uint8_t out[4096];
	std::string s;

	z_stream stream{};
	stream.next_in = const_cast<uint8_t*>(input.data());
	stream.avail_in = static_cast<unsigned int>(input.length());

	if (inflateInit(&stream) != Z_OK)
	return TraceBlobView(nullptr, 0, 0);

	int ret;
	do {
	stream.next_out = out;
	stream.avail_out = sizeof(out);
	ret = inflate(&stream, Z_NO_FLUSH);
	if (ret != Z_STREAM_END && ret != Z_OK) {
	inflateEnd(&stream);
	return TraceBlobView(nullptr, 0, 0);
	}
	s.append(reinterpret_cast<char*>(out), sizeof(out) - stream.avail_out);
	} while (ret != Z_STREAM_END);
	inflateEnd(&stream);

	std::unique_ptr<uint8_t[]> output(new uint8_t[s.size()]);
	memcpy(output.get(), s.data(), s.size());
	return TraceBlobView(std::move(output), 0, s.size());
	}

	} // 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.
	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) {
	protozero::ConstBytes packet = *it;
	size_t field_offset = whole_buf.offset_of(packet.data);
	util::Status status =
	ParsePacket(whole_buf.slice(field_offset, packet.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())
	: std::max(latest_timestamp_, context_->sorter->max_timestamp());

	const uint32_t seq_id = decoder.trusted_packet_sequence_id();

	if ((decoder.has_chrome_events() \|\| decoder.has_chrome_metadata()) &&
	(!decoder.timestamp_clock_id() \|\|
	decoder.timestamp_clock_id() ==
	protos::pbzero::ClockSnapshot::Clock::MONOTONIC)) {
	// Chrome event timestamps are in MONOTONIC domain, but may occur in traces
	// where (a) no clock snapshots exist or (b) no clock_id is specified for
	// their timestamps. Adjust to trace time if we have a clock snapshot.
	// TODO(eseckler): Set timestamp_clock_id and emit ClockSnapshots in chrome
	// and then remove this.
	auto trace_ts = context_->clock_tracker->ToTraceTime(
	protos::pbzero::ClockSnapshot::Clock::MONOTONIC, timestamp);
	if (trace_ts.has_value())
	timestamp = trace_ts.value();
	} else if (decoder.timestamp_clock_id()) {
	// If the TracePacket specifies a non-zero clock-id, translate the timestamp
	// into the trace-time clock domain.
	PERFETTO_DCHECK(decoder.has_timestamp());
	ClockTracker::ClockId clock_id = decoder.timestamp_clock_id();
	bool is_seq_scoped = ClockTracker::IsReservedSeqScopedClockId(clock_id);
	if (is_seq_scoped) {
	if (!seq_id) {
	return util::ErrStatus(
	"TracePacket specified a sequence-local clock id (%" PRIu32
	") but the TraceWriter's sequence_id is zero (the service is "
	"probably too old)",
	decoder.timestamp_clock_id());
	}
	clock_id = ClockTracker::SeqScopedClockIdToGlobal(
	seq_id, decoder.timestamp_clock_id());
	}
	auto trace_ts = context_->clock_tracker->ToTraceTime(clock_id, timestamp);
	if (!trace_ts.has_value()) {
	// ToTraceTime() will increase the \|clock_sync_failure\| stat on failure.
	static const char seq_extra_err[] =
	" Because the clock id is sequence-scoped, the ClockSnapshot must be "
	"emitted on the same TraceWriter sequence of the packet that refers "
	"to that clock id.";
	return util::ErrStatus(
	"Failed to convert TracePacket's timestamp from clock_id=%" PRIu32
	" seq_id=%" PRIu32
	". This is usually due to the lack of a prior ClockSnapshot proto.%s",
	decoder.timestamp_clock_id(), seq_id,
	is_seq_scoped ? seq_extra_err : "");
	}
	timestamp = trace_ts.value();
	}
	latest_timestamp_ = std::max(timestamp, latest_timestamp_);

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

	uint32_t sequence_flags = decoder.sequence_flags();

	if (decoder.incremental_state_cleared() \|\|
	sequence_flags &
	protos::pbzero::TracePacket::SEQ_INCREMENTAL_STATE_CLEARED) {
	HandleIncrementalStateCleared(decoder);
	} else if (decoder.previous_packet_dropped()) {
	HandlePreviousPacketDropped(decoder);
	}

	if (decoder.sequence_flags() &
	protos::pbzero::TracePacket::SEQ_NEEDS_INCREMENTAL_STATE) {
	if (!seq_id) {
	return util::ErrStatus(
	"TracePacket specified SEQ_NEEDS_INCREMENTAL_STATE but the "
	"TraceWriter's sequence_id is zero (the service is "
	"probably too old)");
	}

	if (!state->IsIncrementalStateValid()) {
	context_->storage->IncrementStats(stats::tokenizer_skipped_packets);
	return util::OkStatus();
	}
	}

	if (decoder.has_clock_snapshot()) {
	return ParseClockSnapshot(decoder.clock_snapshot(),
	decoder.trusted_packet_sequence_id());
	}

	// TODO(eseckler): Parse TracePacketDefaults.

	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));
	}

	ModuleResult res = ModuleResult::Ignored();
	res = context_->ftrace_module->TokenizePacket(decoder, &packet, timestamp,
	state);
	if (!res.ignored())
	return res.ToStatus();

	res = context_->track_event_module->TokenizePacket(decoder, &packet,
	timestamp, state);
	if (!res.ignored())
	return res.ToStatus();

	auto& modules = context_->modules_by_field;
	for (uint32_t field_id = 1; field_id < modules.size(); ++field_id) {
	if (modules[field_id] && decoder.Get(field_id).valid()) {
	modules[field_id]->TokenizePacket(decoder, &packet, timestamp, state,
	field_id);
	if (!res.ignored())
	return res.ToStatus();
	}
	}

	if (decoder.has_compressed_packets()) {
	protozero::ConstBytes field = decoder.compressed_packets();
	const size_t field_off = packet.offset_of(field.data);
	TraceBlobView compressed_packets = packet.slice(field_off, field.size);
	TraceBlobView packets = Decompress(std::move(compressed_packets));

	const uint8_t* start = packets.data();
	const uint8_t* end = packets.data() + packets.length();
	const uint8_t* ptr = start;
	while ((end - ptr) > 2) {
	const uint8_t* packet_start = ptr;
	if (PERFETTO_UNLIKELY(*ptr != kTracePacketTag))
	return util::ErrStatus("Expected TracePacket tag");
	uint64_t packet_size = 0;
	ptr = ParseVarInt(++ptr, end, &packet_size);
	size_t packet_offset = static_cast<size_t>(ptr - start);
	ptr += packet_size;
	if (PERFETTO_UNLIKELY((ptr - packet_start) < 2 \|\| ptr > end))
	return util::ErrStatus("Invalid packet size");
	util::Status status = ParsePacket(
	packets.slice(packet_offset, static_cast<size_t>(packet_size)));
	if (PERFETTO_UNLIKELY(!status.ok()))
	return status;
	}

	return util::OkStatus();
	}

	// If we're not forcing a full sort and this is a write_into_file trace, then
	// use flush_period_ms as an indiciator for how big the sliding window for the
	// sorter should be.
	if (!context_->config.force_full_sort && 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, state, 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());

	// Don't parse interned data entries until incremental state is valid, because
	// they could otherwise be associated with the wrong generation in the state.
	if (!state->IsIncrementalStateValid()) {
	context_->storage->IncrementStats(stats::tokenizer_skipped_packets);
	return;
	}

	// Store references to interned data submessages into the sequence's state.
	protozero::ProtoDecoder decoder(interned_data.data(), interned_data.length());
	for (protozero::Field f = decoder.ReadField(); f.valid();
	f = decoder.ReadField()) {
	auto bytes = f.as_bytes();
	auto offset = interned_data.offset_of(bytes.data);
	state->InternMessage(f.id(), interned_data.slice(offset, bytes.size));
	}
	}

	util::Status ProtoTraceTokenizer::ParseClockSnapshot(ConstBytes blob,
	uint32_t seq_id) {
	std::vector<ClockTracker::ClockValue> clocks;
	protos::pbzero::ClockSnapshot::Decoder evt(blob.data, blob.size);
	for (auto it = evt.clocks(); it; ++it) {
	protos::pbzero::ClockSnapshot::Clock::Decoder clk(*it);
	ClockTracker::ClockId clock_id = clk.clock_id();
	if (ClockTracker::IsReservedSeqScopedClockId(clk.clock_id())) {
	if (!seq_id) {
	return util::ErrStatus(
	"ClockSnapshot packet is specifying a sequence-scoped clock id "
	"(%" PRIu64 ") but the TracePacket sequence_id is zero",
	clock_id);
	}
	clock_id = ClockTracker::SeqScopedClockIdToGlobal(seq_id, clk.clock_id());
	}
	int64_t unit_multiplier_ns =
	clk.unit_multiplier_ns()
	? static_cast<int64_t>(clk.unit_multiplier_ns())
	: 1;
	clocks.emplace_back(clock_id, clk.timestamp(), unit_multiplier_ns,
	clk.is_incremental());
	}
	context_->clock_tracker->AddSnapshot(clocks);
	return util::OkStatus();
	}

	} // namespace trace_processor
	} // namespace perfetto