tracing v2: Introduce TraceBufferWriter

components/tracing/core/trace_buffer_writer.cc

Index: components/tracing/core/trace_buffer_writer.cc
diff --git a/components/tracing/core/trace_buffer_writer.cc b/components/tracing/core/trace_buffer_writer.cc
new file mode 100644
index 0000000000000000000000000000000000000000..6f712f6e59206b0bde481d68c2a7ab6bbd4149af
--- /dev/null
+++ b/components/tracing/core/trace_buffer_writer.cc
@@ -0,0 +1,211 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "components/tracing/core/trace_buffer_writer.h"
+
+#include "base/compiler_specific.h"
+#include "base/logging.h"
+#include "components/tracing/core/proto_utils.h"
+
+namespace tracing {
+namespace v2 {
+
+namespace {
+
+// TODO(primiano) remove this in next CLs. This should just be taken from the
+// C++ class autogenerated from events_chunk.proto (crbug.com/608721).
+struct ChunkProto {
+  enum : uint32_t {
+    kWriterIdFieldNumber = 1,
+    kSeqIdInStreamFieldNumber = 2,
+    kEventsFieldNumber = 3,
+    kFirstEventContinuesFromPrevChunkFieldNumber = 4,
+    kLastEventContinuesOnNextChunkFieldNumber = 5
+  };
+};
+
+const size_t kEventPreambleSize = 1 + proto::kMessageLengthFieldSize;
+
+}  // namespace
+
+TraceBufferWriter::TraceBufferWriter(TraceRingBuffer* trace_ring_buffer,
+                                     uint32_t writer_id)
+    : trace_ring_buffer_(trace_ring_buffer),
+      writer_id_(writer_id),
+      chunk_seq_id_(0),
+      chunk_(nullptr),
+      continue_on_next_chunk_ptr_(nullptr),
+      event_data_start_in_current_chunk_(nullptr),
+      stream_writer_(this) {
+  event_.Reset(&stream_writer_);
+}
+
+TraceBufferWriter::~TraceBufferWriter() {}
+
+void TraceBufferWriter::FinalizeCurrentEvent() {
+  if (UNLIKELY(!chunk_))
+    return;
+
+  // Finalize the last event added. This ensures that it and all its nested
+  // fields are committed to the ring buffer and sealed. No further changes to
+  // the chunks's memory can be made from the |event_| after this point.
+  event_.Finalize();
+
+  // In the unlikely event that the last event did wrap over one or more chunks,
+  // is is now time to return those chunks (all but the active one) back.
+  TraceRingBuffer::Chunk* retained_chunk = chunk_->next_in_owner_list();
+  if (UNLIKELY(retained_chunk)) {
+    while (retained_chunk) {
+      TraceRingBuffer::Chunk* next = retained_chunk->next_in_owner_list();
+      retained_chunk->set_next_in_owner_list(nullptr);
+      trace_ring_buffer_->ReturnChunk(retained_chunk);
+      retained_chunk = next;
+    }
+    chunk_->set_next_in_owner_list(nullptr);
+  }
+}
+
+TraceEventHandle TraceBufferWriter::AddEvent() {
+  FinalizeCurrentEvent();
+
+  // In order to start a new event at least kMessageLengthFieldSize + 1 bytes
+  // are required in the chunk to write the preamble and size of the event
+  // itself. We take a bit more room here, it doesn't make a lot of sense
+  // starting a partial event that will fragment immediately after.
+  // TODO(primiano): replace 16 with a more reasonable size, that is, the size
+  // of a simple trace event with no args.
+  if (stream_writer_.bytes_available() < 16)
+    stream_writer_.Reset(AcquireNewChunk(false /* is_fragmenting_event */));
+
+  event_.Reset(&stream_writer_);
+  WriteEventPrambleForNewChunk(

[alph, 2016/08/04 19:06:52]

Preamble

[Primiano Tucci (use gerrit), 2016/08/05 11:23:38]

Oops.

+      stream_writer_.ReserveBytesUnsafe<kEventPreambleSize>());
+  DCHECK_EQ(stream_writer_.write_ptr(), event_data_start_in_current_chunk_);
+  return TraceEventHandle(static_cast<::tracing::proto::Event*>(&event_));
+}
+
+// This is invoked by the ProtoZeroMessage write methods when reaching the
+// end of the current chunk during a write.
+ContiguousMemoryRange TraceBufferWriter::GetNewBuffer() {
+  return AcquireNewChunk(true /* is_fragmenting_event */);
+}
+
+void TraceBufferWriter::FinalizeCurrentChunk() {
+  if (!chunk_)
+    return;
+  DCHECK_GE(stream_writer_.write_ptr(), chunk_->payload());
+  DCHECK_LE(stream_writer_.write_ptr(), chunk_->end());
+  const size_t used_size = stream_writer_.write_ptr() - chunk_->payload();
+  chunk_->set_used_size(static_cast<uint32_t>(used_size));
+}
+
+// There are paths that lead to AcquireNewChunk():
+// When |is_fragmenting_event| = false:
+//   AddEvent() is called and there isn't enough room in the current chunk to
+//   start a new event (or we don't have a chunk yet).
+// When |is_fragmenting_event| = true:
+//   The client is writing an event, a ProtoZeroMessage::Append* method hits
+//   the boundary of the chunk and requests a new one via GetNewBuffer().
+ContiguousMemoryRange TraceBufferWriter::AcquireNewChunk(
+    bool is_fragmenting_event) {
+  FinalizeCurrentChunk();
+  TraceRingBuffer::Chunk* new_chunk = trace_ring_buffer_->TakeChunk(writer_id_);
+  if (is_fragmenting_event) {
+    // continue_on_next_chunk_ptr_ points to the proto field of the previous
+    // chunk and indicates that its last event continues in |new_chunk|.
+    *continue_on_next_chunk_ptr_ = 1;
+
+    // Backfill the size field of the event with the partial size acccumulated

[alph, 2016/08/04 19:06:52]

s/ccc/cc/

[Primiano Tucci (use gerrit), 2016/08/05 11:23:38]

But it can acccccccumulate a lot :P

+    // so far in the old chunk. WriteEventPrambleForNewChunk() will take care

[alph, 2016/08/04 19:06:52]

Preamble

[Primiano Tucci (use gerrit), 2016/08/05 11:23:38]

yeah by virtue of autocompletion, I've been very consistent with this Pramble
apparently. Fixed all instances (I hope) :)

+    // of resetting the |size_field| of the event to the new chunk.
+    DCHECK_GE(event_data_start_in_current_chunk_, chunk_->payload());
+    DCHECK_LE(event_data_start_in_current_chunk_,
+              chunk_->end() - proto::kMessageLengthFieldSize);
+    const uint32_t event_partial_size = static_cast<uint32_t>(
+        stream_writer_.write_ptr() - event_data_start_in_current_chunk_);
+    proto::WriteRedundantVarIntU32<proto::kMessageLengthFieldSize>(
+        event_partial_size, event_.size_field().begin);
+    event_.inc_size_already_written(event_partial_size);
+
+    // If this is a continuation of an event, this writer needs to retain the
+    // old chunk. The client might still be able to write to it. This is to deal
+    // with the case of a nested message which is started in one chunk and
+    // ends in another one. The finalization needs to write-back the size field
+    // in the old chunk.
+    new_chunk->set_next_in_owner_list(chunk_);
+  } else if (chunk_) {
+    // Otherwise, if this is a new event, the previous chunk can be returned.
+    trace_ring_buffer_->ReturnChunk(chunk_);
+  }
+  chunk_ = new_chunk;
+
+  // Write the protobuf for the chunk header. The generated C++ stub for
+  // events_chunk.proto cannot be used here because that would re-enter this

[alph, 2016/08/04 19:06:52]

Why it would reenter. We're at the very beginning of the chunk. It should have
enough space to fit the header. right?

[Primiano Tucci (use gerrit), 2016/08/05 11:23:38]

SO the situation is the following:
 - somebody is appending an event to the chunk via the TBWriter.
 - we reach the end of chunk, it means that the current SSWriter's cur_range is
exhausted.
 - The exhaustion causes the SSWriter to call back GetNewBuffer() to get a new
range (which TBWriter will map to a new chunk).
 - GetNewBuffer() in turn calls this function.
 - But at the moment, until this function returns, we still have the exahusted
range under our feet. The range will be changed only once this function returns.

So if I try to use the writer itself here, will re-enter because the logic loop
is "oh we ran out of space. Let me request a new chunk. Let me write the header
of the new chunk. Of we ran out of space".
I tried reshuffling this a bit around (e.g. resetting the SSWriter cur_range
here as soon as we get a new chunk) but the code looked still quite awkward,
even if it did technically work. To me the awkward bit is the the logic cycle
where: proto messages use the TBWriter to serialize. But in turn TBWriter uses a
proto message to serialize the header of the chunk itself. 

+  // class and make this code extremely hard to reason about.
+  uint8_t* chunk_proto = new_chunk->payload();
+
+  proto::StaticAssertSingleBytePreamble<ChunkProto::kWriterIdFieldNumber>();
+  *chunk_proto++ = static_cast<uint8_t>(
+      proto::MakeTagVarInt(ChunkProto::kWriterIdFieldNumber));
+  chunk_proto = proto::WriteVarIntU32(writer_id_, chunk_proto);
+
+  proto::StaticAssertSingleBytePreamble<
+      ChunkProto::kSeqIdInStreamFieldNumber>();
+  *chunk_proto++ = static_cast<uint8_t>(
+      proto::MakeTagVarInt(ChunkProto::kSeqIdInStreamFieldNumber));
+  chunk_proto = proto::WriteVarIntU32(chunk_seq_id_, chunk_proto);
+
+  proto::StaticAssertSingleBytePreamble<
+      ChunkProto::kFirstEventContinuesFromPrevChunkFieldNumber>();
+  *chunk_proto++ = static_cast<uint8_t>(proto::MakeTagVarInt(
+      ChunkProto::kFirstEventContinuesFromPrevChunkFieldNumber));

[alph, 2016/08/04 19:06:52]

This is an optional field. We could just skip it if the event is not fragmented.
Right?

[Primiano Tucci (use gerrit), 2016/08/05 11:23:38]

ah good point. yes. done.

+  *chunk_proto++ = is_fragmenting_event ? 1u : 0u;
+
+  // At this point we don't know yet whether the last event in the chunk will
+  // fragment and continue in the next chunk. For the moment we put a zero as a

[alph, 2016/08/04 19:06:52]

Could we optionally write this field when we're reaching the end of the chunk?

[Primiano Tucci (use gerrit), 2016/08/05 11:23:38]

Ah I think I see the suggestion here.
Instead of reserving and then backfilling, which causes to have the bit awkward
pointer to bool (continue_on_next_chunk_ptr_ ), which I never liked, I can do
the following:
When I acquire a chunk I tell the SSW that it's end is chunk_end - 2, so I
reserve the right of writing the final two bytes for the boolean.
I write them in FinalizeCurrentChunk.
Yes, just tried it out and it removes quite some complexity. Thanks!

+  // placeholder and remember its position in the chunk. The actual value will
+  // be written the next time we will take a new chunk (above in this function).
+  proto::StaticAssertSingleBytePreamble<
+      ChunkProto::kLastEventContinuesOnNextChunkFieldNumber>();
+  *chunk_proto++ = static_cast<uint8_t>(proto::MakeTagVarInt(
+      ChunkProto::kLastEventContinuesOnNextChunkFieldNumber));
+  continue_on_next_chunk_ptr_ = chunk_proto;
+  *chunk_proto++ = 0;
+
+  ++chunk_seq_id_;
+
+  // If the new chunk was requested while writing an event (the event spans
+  // across chunks) write a new preamble for the partial event in the new chunk.
+  if (is_fragmenting_event)
+    chunk_proto = WriteEventPrambleForNewChunk(chunk_proto);
+
+  return {chunk_proto, new_chunk->end()};
+}
+
+// Writes the one-byte preamble for the start of either a new or a partial
+// event and reserves kMessageLengthFieldSize bytes for its length. Also
+// keeps size-field the bookeeping up to date. Returns the pointer in the chunk
+// past the event preamble, where the event proto should be written.
+uint8_t* TraceBufferWriter::WriteEventPrambleForNewChunk(uint8_t* begin) {
+  // The caller must have ensured to have enough room in the chunk. The event
+  // preamble itself cannot be fragmented.
+  uint8_t* const end = begin + kEventPreambleSize;
+  proto::StaticAssertSingleBytePreamble<ChunkProto::kEventsFieldNumber>();
+  *begin++ = static_cast<uint8_t>(
+      proto::MakeTagLengthDelimited(ChunkProto::kEventsFieldNumber));
+  ContiguousMemoryRange range = {begin, end};
+  event_.set_size_field(range);
+  event_data_start_in_current_chunk_ = end;
+  return end;
+}
+
+void TraceBufferWriter::Flush() {
+  FinalizeCurrentEvent();
+  FinalizeCurrentChunk();
+  trace_ring_buffer_->ReturnChunk(chunk_);
+  chunk_ = nullptr;
+}
+
+}  // namespace v2
+}  // namespace tracing