common/protobuf/kudu/util/debug/trace_event_impl.cc

// Copyright (c) 2012 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 "kudu/util/debug/trace_event_impl.h"

#include <sched.h>
#include <unistd.h>

#include <algorithm>
#include <cinttypes>
#include <cstdlib>
#include <cstring>
#include <functional>
#include <list>
#include <memory>
#include <sstream>
#include <utility>
#include <vector>

#include <gflags/gflags.h>

#include "kudu/gutil/dynamic_annotations.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/mathlimits.h"
#include "kudu/gutil/port.h"
#include "kudu/gutil/ref_counted_memory.h"
#include "kudu/gutil/singleton.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/split.h"
#include "kudu/gutil/strings/stringpiece.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/strings/util.h"
#include "kudu/gutil/sysinfo.h"
#include "kudu/gutil/walltime.h"
#include "kudu/util/atomic.h"
#include "kudu/util/debug/trace_event.h"
#include "kudu/util/debug/trace_event_synthetic_delay.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/monotime.h"
#include "kudu/util/status.h"
#include "kudu/util/thread.h"
#include "kudu/util/threadlocal.h"

DEFINE_string(trace_to_console, "",
              "Trace pattern specifying which trace events should be dumped "
              "directly to the console");
TAG_FLAG(trace_to_console, experimental);

// The thread buckets for the sampling profiler.
TRACE_EVENT_API_ATOMIC_WORD g_trace_state[3];

using base::SpinLockHolder;
using strings::SubstituteAndAppend;
using std::string;
using std::unique_ptr;
using std::vector;

namespace kudu {
namespace debug {

__thread TraceLog::PerThreadInfo* TraceLog::thread_local_info_ = nullptr;

namespace {

// Controls the number of trace events we will buffer in-memory
// before throwing them away.
const size_t kTraceBufferChunkSize = TraceBufferChunk::kTraceBufferChunkSize;
const size_t kTraceEventVectorBufferChunks = 256000 / kTraceBufferChunkSize;
const size_t kTraceEventRingBufferChunks = kTraceEventVectorBufferChunks / 4;
const size_t kTraceEventBatchChunks = 1000 / kTraceBufferChunkSize;
// Can store results for 30 seconds with 1 ms sampling interval.
const size_t kMonitorTraceEventBufferChunks = 30000 / kTraceBufferChunkSize;
// ECHO_TO_CONSOLE needs a small buffer to hold the unfinished COMPLETE events.
const size_t kEchoToConsoleTraceEventBufferChunks = 256;

const char kSyntheticDelayCategoryFilterPrefix[] = "DELAY(";

#define MAX_CATEGORY_GROUPS 100

// Parallel arrays g_category_groups and g_category_group_enabled are separate
// so that a pointer to a member of g_category_group_enabled can be easily
// converted to an index into g_category_groups. This allows macros to deal
// only with char enabled pointers from g_category_group_enabled, and we can
// convert internally to determine the category name from the char enabled
// pointer.
const char* g_category_groups[MAX_CATEGORY_GROUPS] = {
  "toplevel",
  "tracing already shutdown",
  "tracing categories exhausted; must increase MAX_CATEGORY_GROUPS",
  "__metadata"};

// The enabled flag is char instead of bool so that the API can be used from C.
unsigned char g_category_group_enabled[MAX_CATEGORY_GROUPS] = { 0 };
// Indexes here have to match the g_category_groups array indexes above.
const int kCategoryAlreadyShutdown = 1;
const int kCategoryCategoriesExhausted = 2;
const int kCategoryMetadata = 3;
const int kNumBuiltinCategories = 4;
// Skip default categories.
AtomicWord g_category_index = kNumBuiltinCategories;

// The name of the current thread. This is used to decide if the current
// thread name has changed. We combine all the seen thread names into the
// output name for the thread.
__thread const char* g_current_thread_name = "";

static void NOTIMPLEMENTED() {
  LOG(FATAL);
}

class TraceBufferRingBuffer : public TraceBuffer {
 public:
  explicit TraceBufferRingBuffer(size_t max_chunks)
      : max_chunks_(max_chunks),
        recyclable_chunks_queue_(new size_t[queue_capacity()]),
        queue_head_(0),
        queue_tail_(max_chunks),
        current_iteration_index_(0),
        current_chunk_seq_(1) {
    chunks_.reserve(max_chunks);
    for (size_t i = 0; i < max_chunks; ++i)
      recyclable_chunks_queue_[i] = i;
  }

  ~TraceBufferRingBuffer() {
    STLDeleteElements(&chunks_);
  }

  virtual unique_ptr<TraceBufferChunk> GetChunk(size_t* index) OVERRIDE {
    // Because the number of threads is much less than the number of chunks,
    // the queue should never be empty.
    DCHECK(!QueueIsEmpty());

    *index = recyclable_chunks_queue_[queue_head_];
    queue_head_ = NextQueueIndex(queue_head_);
    current_iteration_index_ = queue_head_;

    if (*index >= chunks_.size())
      chunks_.resize(*index + 1);

    TraceBufferChunk* chunk = chunks_[*index];
    chunks_[*index] = nullptr;  // Put NULL in the slot of a in-flight chunk.
    if (chunk)
      chunk->Reset(current_chunk_seq_++);
    else
      chunk = new TraceBufferChunk(current_chunk_seq_++);

    return unique_ptr<TraceBufferChunk>(chunk);
  }

  virtual void ReturnChunk(size_t index,
                           unique_ptr<TraceBufferChunk> chunk) OVERRIDE {
    // When this method is called, the queue should not be full because it
    // can contain all chunks including the one to be returned.
    DCHECK(!QueueIsFull());
    DCHECK(chunk);
    DCHECK_LT(index, chunks_.size());
    DCHECK(!chunks_[index]);
    chunks_[index] = chunk.release();
    recyclable_chunks_queue_[queue_tail_] = index;
    queue_tail_ = NextQueueIndex(queue_tail_);
  }

  virtual bool IsFull() const OVERRIDE {
    return false;
  }

  virtual size_t Size() const OVERRIDE {
    // This is approximate because not all of the chunks are full.
    return chunks_.size() * kTraceBufferChunkSize;
  }

  virtual size_t Capacity() const OVERRIDE {
    return max_chunks_ * kTraceBufferChunkSize;
  }

  virtual TraceEvent* GetEventByHandle(TraceEventHandle handle) OVERRIDE {
    if (handle.chunk_index >= chunks_.size())
      return nullptr;
    TraceBufferChunk* chunk = chunks_[handle.chunk_index];
    if (!chunk || chunk->seq() != handle.chunk_seq)
      return nullptr;
    return chunk->GetEventAt(handle.event_index);
  }

  virtual const TraceBufferChunk* NextChunk() OVERRIDE {
    if (chunks_.empty())
      return nullptr;

    while (current_iteration_index_ != queue_tail_) {
      size_t chunk_index = recyclable_chunks_queue_[current_iteration_index_];
      current_iteration_index_ = NextQueueIndex(current_iteration_index_);
      if (chunk_index >= chunks_.size()) // Skip uninitialized chunks.
        continue;
      DCHECK(chunks_[chunk_index]);
      return chunks_[chunk_index];
    }
    return nullptr;
  }

  virtual unique_ptr<TraceBuffer> CloneForIteration() const OVERRIDE {
    unique_ptr<ClonedTraceBuffer> cloned_buffer(new ClonedTraceBuffer());
    for (size_t queue_index = queue_head_; queue_index != queue_tail_;
        queue_index = NextQueueIndex(queue_index)) {
      size_t chunk_index = recyclable_chunks_queue_[queue_index];
      if (chunk_index >= chunks_.size()) // Skip uninitialized chunks.
        continue;
      TraceBufferChunk* chunk = chunks_[chunk_index];
      cloned_buffer->chunks_.push_back(chunk ? chunk->Clone().release() : nullptr);
    }
    return std::move(cloned_buffer);
  }

 private:
  class ClonedTraceBuffer : public TraceBuffer {
   public:
    ClonedTraceBuffer() : current_iteration_index_(0) {}
    ~ClonedTraceBuffer() {
      STLDeleteElements(&chunks_);
    }

    // The only implemented method.
    virtual const TraceBufferChunk* NextChunk() OVERRIDE {
      return current_iteration_index_ < chunks_.size() ?
          chunks_[current_iteration_index_++] : nullptr;
    }

    virtual unique_ptr<TraceBufferChunk> GetChunk(size_t* /*index*/) OVERRIDE {
      NOTIMPLEMENTED();
      return unique_ptr<TraceBufferChunk>();
    }
    virtual void ReturnChunk(size_t /*index*/,
                             unique_ptr<TraceBufferChunk> /*chunk*/) OVERRIDE {
      NOTIMPLEMENTED();
    }
    virtual bool IsFull() const OVERRIDE { return false; }
    virtual size_t Size() const OVERRIDE { return 0; }
    virtual size_t Capacity() const OVERRIDE { return 0; }
    virtual TraceEvent* GetEventByHandle(TraceEventHandle /*handle*/) OVERRIDE {
      return nullptr;
    }
    virtual unique_ptr<TraceBuffer> CloneForIteration() const OVERRIDE {
      NOTIMPLEMENTED();
      return unique_ptr<TraceBuffer>();
    }

    size_t current_iteration_index_;
    vector<TraceBufferChunk*> chunks_;
  };

  bool QueueIsEmpty() const {
    return queue_head_ == queue_tail_;
  }

  size_t QueueSize() const {
    return queue_tail_ > queue_head_ ? queue_tail_ - queue_head_ :
        queue_tail_ + queue_capacity() - queue_head_;
  }

  bool QueueIsFull() const {
    return QueueSize() == queue_capacity() - 1;
  }

  size_t queue_capacity() const {
    // One extra space to help distinguish full state and empty state.
    return max_chunks_ + 1;
  }

  size_t NextQueueIndex(size_t index) const {
    index++;
    if (index >= queue_capacity())
      index = 0;
    return index;
  }

  size_t max_chunks_;
  vector<TraceBufferChunk*> chunks_;

  unique_ptr<size_t[]> recyclable_chunks_queue_;
  size_t queue_head_;
  size_t queue_tail_;

  size_t current_iteration_index_;
  uint32_t current_chunk_seq_;

  DISALLOW_COPY_AND_ASSIGN(TraceBufferRingBuffer);
};

class TraceBufferVector : public TraceBuffer {
 public:
  TraceBufferVector()
      : in_flight_chunk_count_(0),
        current_iteration_index_(0) {
    chunks_.reserve(kTraceEventVectorBufferChunks);
  }
  ~TraceBufferVector() {
    STLDeleteElements(&chunks_);
  }

  virtual unique_ptr<TraceBufferChunk> GetChunk(size_t* index) OVERRIDE {
    // This function may be called when adding normal events or indirectly from
    // AddMetadataEventsWhileLocked(). We can not DECHECK(!IsFull()) because we
    // have to add the metadata events and flush thread-local buffers even if
    // the buffer is full.
    *index = chunks_.size();
    chunks_.push_back(nullptr);  // Put NULL in the slot of a in-flight chunk.
    ++in_flight_chunk_count_;
    // + 1 because zero chunk_seq is not allowed.
    return unique_ptr<TraceBufferChunk>(
        new TraceBufferChunk(static_cast<uint32_t>(*index) + 1));
  }

  virtual void ReturnChunk(size_t index,
                           unique_ptr<TraceBufferChunk> chunk) OVERRIDE {
    DCHECK_GT(in_flight_chunk_count_, 0U);
    DCHECK_LT(index, chunks_.size());
    DCHECK(!chunks_[index]);
    --in_flight_chunk_count_;
    chunks_[index] = chunk.release();
  }

  virtual bool IsFull() const OVERRIDE {
    return chunks_.size() >= kTraceEventVectorBufferChunks;
  }

  virtual size_t Size() const OVERRIDE {
    // This is approximate because not all of the chunks are full.
    return chunks_.size() * kTraceBufferChunkSize;
  }

  virtual size_t Capacity() const OVERRIDE {
    return kTraceEventVectorBufferChunks * kTraceBufferChunkSize;
  }

  virtual TraceEvent* GetEventByHandle(TraceEventHandle handle) OVERRIDE {
    if (handle.chunk_index >= chunks_.size())
      return nullptr;
    TraceBufferChunk* chunk = chunks_[handle.chunk_index];
    if (!chunk || chunk->seq() != handle.chunk_seq)
      return nullptr;
    return chunk->GetEventAt(handle.event_index);
  }

  virtual const TraceBufferChunk* NextChunk() OVERRIDE {
    while (current_iteration_index_ < chunks_.size()) {
      // Skip in-flight chunks.
      const TraceBufferChunk* chunk = chunks_[current_iteration_index_++];
      if (chunk)
        return chunk;
    }
    return nullptr;
  }

  virtual unique_ptr<TraceBuffer> CloneForIteration() const OVERRIDE {
    NOTIMPLEMENTED();
    return unique_ptr<TraceBuffer>();
  }

 private:
  size_t in_flight_chunk_count_;
  size_t current_iteration_index_;
  vector<TraceBufferChunk*> chunks_;

  DISALLOW_COPY_AND_ASSIGN(TraceBufferVector);
};

template <typename T>
void InitializeMetadataEvent(TraceEvent* trace_event,
                             int thread_id,
                             const char* metadata_name, const char* arg_name,
                             const T& value) {
  if (!trace_event)
    return;

  int num_args = 1;
  unsigned char arg_type;
  uint64_t arg_value;
  ::trace_event_internal::SetTraceValue(value, &arg_type, &arg_value);
  trace_event->Initialize(thread_id,
                          MicrosecondsInt64(0), MicrosecondsInt64(0), TRACE_EVENT_PHASE_METADATA,
                          &g_category_group_enabled[kCategoryMetadata],
                          metadata_name, ::trace_event_internal::kNoEventId,
                          num_args, &arg_name, &arg_type, &arg_value, nullptr,
                          TRACE_EVENT_FLAG_NONE);
}

// RAII object which marks '*dst' with a non-zero value while in scope.
// This assumes that no other threads write to '*dst'.
class MarkFlagInScope {
 public:
  explicit MarkFlagInScope(Atomic32* dst)
      : dst_(dst) {
    // We currently use Acquire_AtomicExchange here because it appears
    // to be the cheapest way of getting an "Acquire_Store" barrier. Actually
    // using Acquire_Store generates more assembly instructions and benchmarks
    // slightly slower.
    //
    // TODO: it would be even faster to avoid the memory barrier here entirely,
    // and do an asymmetric barrier, for example by having the flusher thread
    // send a signal to every registered thread, or wait until every other thread
    // has experienced at least one context switch. A number of options for this
    // are outlined in:
    // http://home.comcast.net/~pjbishop/Dave/Asymmetric-Dekker-Synchronization.txt
    Atomic32 old_val = base::subtle::Acquire_AtomicExchange(dst_, 1);
    DCHECK_EQ(old_val, 0);
  }
  ~MarkFlagInScope() {
    base::subtle::Release_Store(dst_, 0);
  }

 private:
  Atomic32* dst_;
  DISALLOW_COPY_AND_ASSIGN(MarkFlagInScope);
};
}  // anonymous namespace

TraceLog::ThreadLocalEventBuffer* TraceLog::PerThreadInfo::AtomicTakeBuffer() {
  return reinterpret_cast<TraceLog::ThreadLocalEventBuffer*>(
    base::subtle::Acquire_AtomicExchange(
      reinterpret_cast<AtomicWord*>(&event_buffer_),
      0));
}

void TraceBufferChunk::Reset(uint32_t new_seq) {
  for (size_t i = 0; i < next_free_; ++i)
    chunk_[i].Reset();
  next_free_ = 0;
  seq_ = new_seq;
}

TraceEvent* TraceBufferChunk::AddTraceEvent(size_t* event_index) {
  DCHECK(!IsFull());
  *event_index = next_free_++;
  return &chunk_[*event_index];
}

unique_ptr<TraceBufferChunk> TraceBufferChunk::Clone() const {
  unique_ptr<TraceBufferChunk> cloned_chunk(new TraceBufferChunk(seq_));
  cloned_chunk->next_free_ = next_free_;
  for (size_t i = 0; i < next_free_; ++i)
    cloned_chunk->chunk_[i].CopyFrom(chunk_[i]);
  return cloned_chunk;
}

// A helper class that allows the lock to be acquired in the middle of the scope
// and unlocks at the end of scope if locked.
class TraceLog::OptionalAutoLock {
 public:
  explicit OptionalAutoLock(base::SpinLock& lock) // NOLINT(google-runtime-references)
      : lock_(lock),
        locked_(false) {
  }

  ~OptionalAutoLock() {
    if (locked_)
      lock_.Unlock();
  }

  void EnsureAcquired() {
    if (!locked_) {
      lock_.Lock();
      locked_ = true;
    }
  }

 private:
  base::SpinLock& lock_;
  bool locked_;
  DISALLOW_COPY_AND_ASSIGN(OptionalAutoLock);
};

// Use this function instead of TraceEventHandle constructor to keep the
// overhead of ScopedTracer (trace_event.h) constructor minimum.
void MakeHandle(uint32_t chunk_seq, size_t chunk_index, size_t event_index,
                TraceEventHandle* handle) {
  DCHECK(chunk_seq);
  DCHECK(chunk_index < (1u << 16));
  DCHECK(event_index < (1u << 16));
  handle->chunk_seq = chunk_seq;
  handle->chunk_index = static_cast<uint16_t>(chunk_index);
  handle->event_index = static_cast<uint16_t>(event_index);
}

////////////////////////////////////////////////////////////////////////////////
//
// TraceEvent
//
////////////////////////////////////////////////////////////////////////////////

namespace {

size_t GetAllocLength(const char* str) { return str ? strlen(str) + 1 : 0; }

// Copies |*member| into |*buffer|, sets |*member| to point to this new
// location, and then advances |*buffer| by the amount written.
void CopyTraceEventParameter(char** buffer,
                             const char** member,
                             const char* end) {
  if (*member) {
    size_t written = strings::strlcpy(*buffer, *member, end - *buffer) + 1;
    DCHECK_LE(static_cast<int>(written), end - *buffer);
    *member = *buffer;
    *buffer += written;
  }
}

}  // namespace

TraceEvent::TraceEvent()
    : duration_(-1),
      thread_duration_(-1),
      id_(0u),
      category_group_enabled_(nullptr),
      name_(nullptr),
      thread_id_(0),
      phase_(TRACE_EVENT_PHASE_BEGIN),
      flags_(0) {
  for (auto& arg_name : arg_names_) {
    arg_name = nullptr;
  }
  memset(arg_values_, 0, sizeof(arg_values_));
}

TraceEvent::~TraceEvent() {
}

void TraceEvent::CopyFrom(const TraceEvent& other) {
  timestamp_ = other.timestamp_;
  thread_timestamp_ = other.thread_timestamp_;
  duration_ = other.duration_;
  id_ = other.id_;
  category_group_enabled_ = other.category_group_enabled_;
  name_ = other.name_;
  thread_id_ = other.thread_id_;
  phase_ = other.phase_;
  flags_ = other.flags_;
  parameter_copy_storage_ = other.parameter_copy_storage_;

  for (int i = 0; i < kTraceMaxNumArgs; ++i) {
    arg_names_[i] = other.arg_names_[i];
    arg_types_[i] = other.arg_types_[i];
    arg_values_[i] = other.arg_values_[i];
    convertable_values_[i] = other.convertable_values_[i];
  }
}

void TraceEvent::Initialize(
    int thread_id,
    MicrosecondsInt64 timestamp,
    MicrosecondsInt64 thread_timestamp,
    char phase,
    const unsigned char* category_group_enabled,
    const char* name,
    uint64_t id,
    int num_args,
    const char** arg_names,
    const unsigned char* arg_types,
    const uint64_t* arg_values,
    const scoped_refptr<ConvertableToTraceFormat>* convertable_values,
    unsigned char flags) {
  timestamp_ = timestamp;
  thread_timestamp_ = thread_timestamp;
  duration_ = -1;
  id_ = id;
  category_group_enabled_ = category_group_enabled;
  name_ = name;
  thread_id_ = thread_id;
  phase_ = phase;
  flags_ = flags;

  // Clamp num_args since it may have been set by a third_party library.
  num_args = (num_args > kTraceMaxNumArgs) ? kTraceMaxNumArgs : num_args;
  int i = 0;
  for (; i < num_args; ++i) {
    arg_names_[i] = arg_names[i];
    arg_types_[i] = arg_types[i];

    if (arg_types[i] == TRACE_VALUE_TYPE_CONVERTABLE)
      convertable_values_[i] = convertable_values[i];
    else
      arg_values_[i].as_uint = arg_values[i];
  }
  for (; i < kTraceMaxNumArgs; ++i) {
    arg_names_[i] = nullptr;
    arg_values_[i].as_uint = 0u;
    convertable_values_[i] = nullptr;
    arg_types_[i] = TRACE_VALUE_TYPE_UINT;
  }

  bool copy = !!(flags & TRACE_EVENT_FLAG_COPY);
  size_t alloc_size = 0;
  if (copy) {
    alloc_size += GetAllocLength(name);
    for (i = 0; i < num_args; ++i) {
      alloc_size += GetAllocLength(arg_names_[i]);
      if (arg_types_[i] == TRACE_VALUE_TYPE_STRING)
        arg_types_[i] = TRACE_VALUE_TYPE_COPY_STRING;
    }
  }

  bool arg_is_copy[kTraceMaxNumArgs];
  for (i = 0; i < num_args; ++i) {
    // No copying of convertable types, we retain ownership.
    if (arg_types_[i] == TRACE_VALUE_TYPE_CONVERTABLE)
      continue;

    // We only take a copy of arg_vals if they are of type COPY_STRING.
    arg_is_copy[i] = (arg_types_[i] == TRACE_VALUE_TYPE_COPY_STRING);
    if (arg_is_copy[i])
      alloc_size += GetAllocLength(arg_values_[i].as_string);
  }

  if (alloc_size) {
    parameter_copy_storage_ = new RefCountedString;
    parameter_copy_storage_->data().resize(alloc_size);
    char* ptr = string_as_array(&parameter_copy_storage_->data());
    const char* end = ptr + alloc_size;
    if (copy) {
      CopyTraceEventParameter(&ptr, &name_, end);
      for (i = 0; i < num_args; ++i) {
        CopyTraceEventParameter(&ptr, &arg_names_[i], end);
      }
    }
    for (i = 0; i < num_args; ++i) {
      if (arg_types_[i] == TRACE_VALUE_TYPE_CONVERTABLE)
        continue;
      if (arg_is_copy[i])
        CopyTraceEventParameter(&ptr, &arg_values_[i].as_string, end);
    }
    DCHECK_EQ(end, ptr) << "Overrun by " << ptr - end;
  }
}

void TraceEvent::Reset() {
  // Only reset fields that won't be initialized in Initialize(), or that may
  // hold references to other objects.
  duration_ = -1;
  parameter_copy_storage_ = nullptr;
  for (int i = 0; i < kTraceMaxNumArgs && arg_names_[i]; ++i)
    convertable_values_[i] = nullptr;
}

void TraceEvent::UpdateDuration(const MicrosecondsInt64& now,
                                const MicrosecondsInt64& thread_now) {
  DCHECK(duration_ == -1);
  duration_ = now - timestamp_;
  thread_duration_ = thread_now - thread_timestamp_;
}

namespace {
// Escape the given string using JSON rules.
void JsonEscape(StringPiece s, string* out) {
  out->reserve(out->size() + s.size() * 2);
  const char* p_end = s.data() + s.size();
  for (const char* p = s.data(); p != p_end; p++) {
    // Only the following characters need to be escaped, according to json.org.
    // In particular, it's illegal to escape the single-quote character, and
    // JSON does not support the "\x" escape sequence like C/Java.
    switch (*p) {
      case '"':
      case '\\':
        out->push_back('\\');
        out->push_back(*p);
        break;
      case '\b':
        out->append("\\b");
        break;
      case '\f':
        out->append("\\f");
        break;
      case '\n':
        out->append("\\n");
        break;
      case '\r':
        out->append("\\r");
        break;
      case '\t':
        out->append("\\t");
        break;
      default:
        out->push_back(*p);
    }
  }
}
} // anonymous namespace

// static
void TraceEvent::AppendValueAsJSON(unsigned char type,
                                   TraceEvent::TraceValue value,
                                   std::string* out) {
  switch (type) {
    case TRACE_VALUE_TYPE_BOOL:
      *out += value.as_bool ? "true" : "false";
      break;
    case TRACE_VALUE_TYPE_UINT:
      SubstituteAndAppend(out, "$0", static_cast<uint64_t>(value.as_uint));
      break;
    case TRACE_VALUE_TYPE_INT:
      SubstituteAndAppend(out, "$0", static_cast<int64_t>(value.as_int));
      break;
    case TRACE_VALUE_TYPE_DOUBLE: {
      // FIXME: base/json/json_writer.cc is using the same code,
      //        should be made into a common method.
      std::string real;
      double val = value.as_double;
      if (MathLimits<double>::IsFinite(val)) {
        real = strings::Substitute("$0", val);
        // Ensure that the number has a .0 if there's no decimal or 'e'.  This
        // makes sure that when we read the JSON back, it's interpreted as a
        // real rather than an int.
        if (real.find('.') == std::string::npos &&
            real.find('e') == std::string::npos &&
            real.find('E') == std::string::npos) {
          real.append(".0");
        }
        // The JSON spec requires that non-integer values in the range (-1,1)
        // have a zero before the decimal point - ".52" is not valid, "0.52" is.
        if (real[0] == '.') {
          real.insert(0, "0");
        } else if (real.length() > 1 && real[0] == '-' && real[1] == '.') {
          // "-.1" bad "-0.1" good
          real.insert(1, "0");
        }
      } else if (MathLimits<double>::IsNaN(val)){
        // The JSON spec doesn't allow NaN and Infinity (since these are
        // objects in EcmaScript).  Use strings instead.
        real = "\"NaN\"";
      } else if (val < 0) {
        real = "\"-Infinity\"";
      } else {
        real = "\"Infinity\"";
      }
      SubstituteAndAppend(out, "$0", real);
      break;
    }
    case TRACE_VALUE_TYPE_POINTER:
      // JSON only supports double and int numbers.
      // So as not to lose bits from a 64-bit pointer, output as a hex string.
      StringAppendF(out, "\"0x%" PRIx64 "\"", static_cast<uint64_t>(
                                     reinterpret_cast<intptr_t>(
                                     value.as_pointer)));
      break;
    case TRACE_VALUE_TYPE_STRING:
    case TRACE_VALUE_TYPE_COPY_STRING:
      *out += "\"";
      JsonEscape(value.as_string ? value.as_string : "NULL", out);
      *out += "\"";
      break;
    default:
      LOG(FATAL) << "Don't know how to print this value";
      break;
  }
}

void TraceEvent::AppendAsJSON(std::string* out) const {
  int64_t time_int64 = timestamp_;
  int process_id = TraceLog::GetInstance()->process_id();
  // Category group checked at category creation time.
  DCHECK(!strchr(name_, '"'));
  StringAppendF(out,
      "{\"cat\":\"%s\",\"pid\":%i,\"tid\":%i,\"ts\":%" PRId64 ","
      "\"ph\":\"%c\",\"name\":\"%s\",\"args\":{",
      TraceLog::GetCategoryGroupName(category_group_enabled_),
      process_id,
      thread_id_,
      time_int64,
      phase_,
      name_);

  // Output argument names and values, stop at first NULL argument name.
  for (int i = 0; i < kTraceMaxNumArgs && arg_names_[i]; ++i) {
    if (i > 0)
      *out += ",";
    *out += "\"";
    *out += arg_names_[i];
    *out += "\":";

    if (arg_types_[i] == TRACE_VALUE_TYPE_CONVERTABLE)
      convertable_values_[i]->AppendAsTraceFormat(out);
    else
      AppendValueAsJSON(arg_types_[i], arg_values_[i], out);
  }
  *out += "}";

  if (phase_ == TRACE_EVENT_PHASE_COMPLETE) {
    int64_t duration = duration_;
    if (duration != -1)
      StringAppendF(out, ",\"dur\":%" PRId64, duration);
    if (thread_timestamp_ >= 0) {
      int64_t thread_duration = thread_duration_;
      if (thread_duration != -1)
        StringAppendF(out, ",\"tdur\":%" PRId64, thread_duration);
    }
  }

  // Output tts if thread_timestamp is valid.
  if (thread_timestamp_ >= 0) {
    int64_t thread_time_int64 = thread_timestamp_;
    StringAppendF(out, ",\"tts\":%" PRId64, thread_time_int64);
  }

  // If id_ is set, print it out as a hex string so we don't loose any
  // bits (it might be a 64-bit pointer).
  if (flags_ & TRACE_EVENT_FLAG_HAS_ID)
    StringAppendF(out, ",\"id\":\"0x%" PRIx64 "\"", static_cast<uint64_t>(id_));

  // Instant events also output their scope.
  if (phase_ == TRACE_EVENT_PHASE_INSTANT) {
    char scope = '?';
    switch (flags_ & TRACE_EVENT_FLAG_SCOPE_MASK) {
      case TRACE_EVENT_SCOPE_GLOBAL:
        scope = TRACE_EVENT_SCOPE_NAME_GLOBAL;
        break;

      case TRACE_EVENT_SCOPE_PROCESS:
        scope = TRACE_EVENT_SCOPE_NAME_PROCESS;
        break;

      case TRACE_EVENT_SCOPE_THREAD:
        scope = TRACE_EVENT_SCOPE_NAME_THREAD;
        break;
    }
    StringAppendF(out, ",\"s\":\"%c\"", scope);
  }

  *out += "}";
}

void TraceEvent::AppendPrettyPrinted(std::ostringstream* out) const {
  *out << name_ << "[";
  *out << TraceLog::GetCategoryGroupName(category_group_enabled_);
  *out << "]";
  if (arg_names_[0]) {
    *out << ", {";
    for (int i = 0; i < kTraceMaxNumArgs && arg_names_[i]; ++i) {
      if (i > 0)
        *out << ", ";
      *out << arg_names_[i] << ":";
      std::string value_as_text;

      if (arg_types_[i] == TRACE_VALUE_TYPE_CONVERTABLE)
        convertable_values_[i]->AppendAsTraceFormat(&value_as_text);
      else
        AppendValueAsJSON(arg_types_[i], arg_values_[i], &value_as_text);

      *out << value_as_text;
    }
    *out << "}";
  }
}

////////////////////////////////////////////////////////////////////////////////
//
// TraceResultBuffer
//
////////////////////////////////////////////////////////////////////////////////

string TraceResultBuffer::FlushTraceLogToString() {
  return DoFlush(false);
}

string TraceResultBuffer::FlushTraceLogToStringButLeaveBufferIntact() {
  return DoFlush(true);
}

string TraceResultBuffer::DoFlush(bool leave_intact) {
  TraceResultBuffer buf;
  auto* buf_ptr = &buf;
  TraceLog* tl = TraceLog::GetInstance();
  auto cb = [buf_ptr](const scoped_refptr<RefCountedString>& s, bool has_more_events) {
    buf_ptr->Collect(s, has_more_events);
  };
  if (leave_intact) {
    tl->FlushButLeaveBufferIntact(cb);
  } else {
    tl->Flush(cb);
  }
  buf.json_.append("]}\n");
  return buf.json_;
}

TraceResultBuffer::TraceResultBuffer()
  : first_(true) {
}
TraceResultBuffer::~TraceResultBuffer() {
}

void TraceResultBuffer::Collect(
  const scoped_refptr<RefCountedString>& s,
  bool has_more_events) {
  if (first_) {
    json_.append("{\"traceEvents\": [\n");
    first_ = false;
  } else if (!s->data().empty()) {
    // Sometimes we get sent an empty chunk at the end,
    // and we don't want to end up with an extra trailing ','
    json_.append(",\n");
  }
  json_.append(s->data());
}

////////////////////////////////////////////////////////////////////////////////
//
// TraceSamplingThread
//
////////////////////////////////////////////////////////////////////////////////
class TraceBucketData;

typedef std::function<void(TraceBucketData*)> TraceSampleCallback;

class TraceBucketData {
 public:
  TraceBucketData(AtomicWord* bucket,
                  const char* name,
                  TraceSampleCallback callback);
  ~TraceBucketData();

  TRACE_EVENT_API_ATOMIC_WORD* bucket;
  const char* bucket_name;
  TraceSampleCallback callback;
};

// This object must be created on the IO thread.
class TraceSamplingThread {
 public:
  TraceSamplingThread();
  virtual ~TraceSamplingThread();

  void ThreadMain();

  static void DefaultSamplingCallback(TraceBucketData* bucket_data);

  void Stop();

 private:
  friend class TraceLog;

  void GetSamples();
  // Not thread-safe. Once the ThreadMain has been called, this can no longer
  // be called.
  void RegisterSampleBucket(TRACE_EVENT_API_ATOMIC_WORD* bucket,
                            const char* const name,
                            TraceSampleCallback callback);
  // Splits a combined "category\0name" into the two component parts.
  static void ExtractCategoryAndName(const char* combined,
                                     const char** category,
                                     const char** name);
  std::vector<TraceBucketData> sample_buckets_;
  bool thread_running_;
  AtomicBool cancellation_flag_;
};


TraceSamplingThread::TraceSamplingThread()
  : thread_running_(false),
    cancellation_flag_(false) {
}

TraceSamplingThread::~TraceSamplingThread() {
}

void TraceSamplingThread::ThreadMain() {
  thread_running_ = true;
  const MonoDelta sleepDelta = MonoDelta::FromMicroseconds(1000);
  while (!cancellation_flag_.Load()) {
    SleepFor(sleepDelta);
    GetSamples();
  }
}

// static
void TraceSamplingThread::DefaultSamplingCallback(
    TraceBucketData* bucket_data) {
  TRACE_EVENT_API_ATOMIC_WORD category_and_name =
      TRACE_EVENT_API_ATOMIC_LOAD(*bucket_data->bucket);
  if (!category_and_name)
    return;
  const char* const combined =
      reinterpret_cast<const char* const>(category_and_name);
  const char* category_group;
  const char* name;
  ExtractCategoryAndName(combined, &category_group, &name);
  TRACE_EVENT_API_ADD_TRACE_EVENT(TRACE_EVENT_PHASE_SAMPLE,
      TraceLog::GetCategoryGroupEnabled(category_group),
      name, 0, 0, nullptr, nullptr, nullptr, nullptr, 0);
}

void TraceSamplingThread::GetSamples() {
  for (auto& sample_bucket : sample_buckets_) {
    TraceBucketData* bucket_data = &sample_bucket;
    bucket_data->callback(bucket_data);
  }
}

void TraceSamplingThread::RegisterSampleBucket(
    TRACE_EVENT_API_ATOMIC_WORD* bucket,
    const char* const name,
    TraceSampleCallback callback) {
  // Access to sample_buckets_ doesn't cause races with the sampling thread
  // that uses the sample_buckets_, because it is guaranteed that
  // RegisterSampleBucket is called before the sampling thread is created.
  DCHECK(!thread_running_);
  sample_buckets_.emplace_back(bucket, name, callback);
}

// static
void TraceSamplingThread::ExtractCategoryAndName(const char* combined,
                                                 const char** category,
                                                 const char** name) {
  *category = combined;
  *name = &combined[strlen(combined) + 1];
}

void TraceSamplingThread::Stop() {
  cancellation_flag_.Store(true);
}

TraceBucketData::TraceBucketData(AtomicWord* bucket, const char* name,
                                 TraceSampleCallback callback)
    : bucket(bucket), bucket_name(name), callback(std::move(callback)) {}

TraceBucketData::~TraceBucketData() {
}

////////////////////////////////////////////////////////////////////////////////
//
// TraceLog
//
////////////////////////////////////////////////////////////////////////////////

class TraceLog::ThreadLocalEventBuffer {
 public:
  explicit ThreadLocalEventBuffer(TraceLog* trace_log);
  virtual ~ThreadLocalEventBuffer();

  TraceEvent* AddTraceEvent(TraceEventHandle* handle);

  TraceEvent* GetEventByHandle(TraceEventHandle handle) {
    if (!chunk_ || handle.chunk_seq != chunk_->seq() ||
        handle.chunk_index != chunk_index_)
      return nullptr;

    return chunk_->GetEventAt(handle.event_index);
  }

  int generation() const { return generation_; }

  void Flush(int64_t tid);

 private:
  // Check that the current thread is the one that constructed this trace buffer.
  void CheckIsOwnerThread() const {
    DCHECK_EQ(kudu::Thread::UniqueThreadId(), owner_tid_);
  }

  // Since TraceLog is a leaky singleton, trace_log_ will always be valid
  // as long as the thread exists.
  TraceLog* trace_log_;
  unique_ptr<TraceBufferChunk> chunk_;
  size_t chunk_index_;
  int generation_;

  // The TID of the thread that constructed this event buffer. Only this thread
  // may add trace events.
  int64_t owner_tid_;

  DISALLOW_COPY_AND_ASSIGN(ThreadLocalEventBuffer);
};

TraceLog::ThreadLocalEventBuffer::ThreadLocalEventBuffer(TraceLog* trace_log)
    : trace_log_(trace_log),
      chunk_index_(0),
      generation_(trace_log->generation()),
      owner_tid_(kudu::Thread::UniqueThreadId()) {
}

TraceLog::ThreadLocalEventBuffer::~ThreadLocalEventBuffer() {
}

TraceEvent* TraceLog::ThreadLocalEventBuffer::AddTraceEvent(
    TraceEventHandle* handle) {
  CheckIsOwnerThread();

  if (chunk_ && chunk_->IsFull()) {
    SpinLockHolder lock(&trace_log_->lock_);
    Flush(Thread::UniqueThreadId());
    chunk_.reset();
  }
  if (!chunk_) {
    SpinLockHolder lock(&trace_log_->lock_);
    chunk_ = trace_log_->logged_events_->GetChunk(&chunk_index_);
    trace_log_->CheckIfBufferIsFullWhileLocked();
  }
  if (!chunk_)
    return nullptr;

  size_t event_index;
  TraceEvent* trace_event = chunk_->AddTraceEvent(&event_index);
  if (trace_event && handle)
    MakeHandle(chunk_->seq(), chunk_index_, event_index, handle);

  return trace_event;
}

void TraceLog::ThreadLocalEventBuffer::Flush(int64_t tid) {
  DCHECK(trace_log_->lock_.IsHeld());

  if (!chunk_)
    return;

  if (trace_log_->CheckGeneration(generation_)) {
    // Return the chunk to the buffer only if the generation matches.
    trace_log_->logged_events_->ReturnChunk(chunk_index_, std::move(chunk_));
  }
}

// static
TraceLog* TraceLog::GetInstance() {
  return Singleton<TraceLog>::get();
}

TraceLog::TraceLog()
    : mode_(DISABLED),
      num_traces_recorded_(0),
      event_callback_(0),
      dispatching_to_observer_list_(false),
      process_sort_index_(0),
      process_id_hash_(0),
      process_id_(0),
      time_offset_(0),
      watch_category_(0),
      trace_options_(RECORD_UNTIL_FULL),
      sampling_thread_handle_(nullptr),
      category_filter_(CategoryFilter::kDefaultCategoryFilterString),
      event_callback_category_filter_(
          CategoryFilter::kDefaultCategoryFilterString),
      thread_shared_chunk_index_(0),
      generation_(0) {
  // Trace is enabled or disabled on one thread while other threads are
  // accessing the enabled flag. We don't care whether edge-case events are
  // traced or not, so we allow races on the enabled flag to keep the trace
  // macros fast.
  ANNOTATE_BENIGN_RACE_SIZED(g_category_group_enabled,
                             sizeof(g_category_group_enabled),
                             "trace_event category enabled");
  for (int i = 0; i < MAX_CATEGORY_GROUPS; ++i) {
    ANNOTATE_BENIGN_RACE(&g_category_group_enabled[i],
                         "trace_event category enabled");
  }
  SetProcessID(static_cast<int>(getpid()));

  string filter = FLAGS_trace_to_console;
  if (!filter.empty()) {
    SetEnabled(CategoryFilter(filter), RECORDING_MODE, ECHO_TO_CONSOLE);
    LOG(ERROR) << "Tracing to console with CategoryFilter '" << filter << "'.";
  }

  logged_events_.reset(CreateTraceBuffer());
}

TraceLog::~TraceLog() {
}

const unsigned char* TraceLog::GetCategoryGroupEnabled(
    const char* category_group) {
  TraceLog* tracelog = GetInstance();
  if (!tracelog) {
    DCHECK(!g_category_group_enabled[kCategoryAlreadyShutdown]);
    return &g_category_group_enabled[kCategoryAlreadyShutdown];
  }
  return tracelog->GetCategoryGroupEnabledInternal(category_group);
}

const char* TraceLog::GetCategoryGroupName(
    const unsigned char* category_group_enabled) {
  // Calculate the index of the category group by finding
  // category_group_enabled in g_category_group_enabled array.
  uintptr_t category_begin =
      reinterpret_cast<uintptr_t>(g_category_group_enabled);
  uintptr_t category_ptr = reinterpret_cast<uintptr_t>(category_group_enabled);
  DCHECK(category_ptr >= category_begin &&
         category_ptr < reinterpret_cast<uintptr_t>(
             g_category_group_enabled + MAX_CATEGORY_GROUPS)) <<
      "out of bounds category pointer";
  uintptr_t category_index =
      (category_ptr - category_begin) / sizeof(g_category_group_enabled[0]);
  return g_category_groups[category_index];
}

void TraceLog::UpdateCategoryGroupEnabledFlag(int category_index) {
  unsigned char enabled_flag = 0;
  const char* category_group = g_category_groups[category_index];
  if (mode_ == RECORDING_MODE &&
      category_filter_.IsCategoryGroupEnabled(category_group))
    enabled_flag |= ENABLED_FOR_RECORDING;
  else if (mode_ == MONITORING_MODE &&
      category_filter_.IsCategoryGroupEnabled(category_group))
    enabled_flag |= ENABLED_FOR_MONITORING;
  if (event_callback_ &&
      event_callback_category_filter_.IsCategoryGroupEnabled(category_group))
    enabled_flag |= ENABLED_FOR_EVENT_CALLBACK;
  g_category_group_enabled[category_index] = enabled_flag;
}

void TraceLog::UpdateCategoryGroupEnabledFlags() {
  int category_index = base::subtle::NoBarrier_Load(&g_category_index);
  for (int i = 0; i < category_index; i++)
    UpdateCategoryGroupEnabledFlag(i);
}

void TraceLog::UpdateSyntheticDelaysFromCategoryFilter() {
  ResetTraceEventSyntheticDelays();
  const CategoryFilter::StringList& delays =
      category_filter_.GetSyntheticDelayValues();
  CategoryFilter::StringList::const_iterator ci;
  for (ci = delays.begin(); ci != delays.end(); ++ci) {
    std::list<string> tokens = strings::Split(*ci, ";");
    if (tokens.empty()) continue;

    TraceEventSyntheticDelay* delay =
        TraceEventSyntheticDelay::Lookup(tokens.front());
    tokens.pop_front();
    while (!tokens.empty()) {
      std::string token = tokens.front();
      tokens.pop_front();
      char* duration_end;
      double target_duration = strtod(token.c_str(), &duration_end);
      if (duration_end != token.c_str()) {
        delay->SetTargetDuration(MonoDelta::FromSeconds(target_duration));
      } else if (token == "static") {
        delay->SetMode(TraceEventSyntheticDelay::STATIC);
      } else if (token == "oneshot") {
        delay->SetMode(TraceEventSyntheticDelay::ONE_SHOT);
      } else if (token == "alternating") {
        delay->SetMode(TraceEventSyntheticDelay::ALTERNATING);
      }
    }
  }
}

const unsigned char* TraceLog::GetCategoryGroupEnabledInternal(
    const char* category_group) {
  DCHECK(!strchr(category_group, '"')) <<
      "Category groups may not contain double quote";
  // The g_category_groups is append only, avoid using a lock for the fast path.
  int current_category_index = base::subtle::Acquire_Load(&g_category_index);

  // Search for pre-existing category group.
  for (int i = 0; i < current_category_index; ++i) {
    if (strcmp(g_category_groups[i], category_group) == 0) {
      return &g_category_group_enabled[i];
    }
  }

  unsigned char* category_group_enabled = nullptr;
  // This is the slow path: the lock is not held in the case above, so more
  // than one thread could have reached here trying to add the same category.
  // Only hold to lock when actually appending a new category, and
  // check the categories groups again.
  SpinLockHolder lock(&lock_);
  int category_index = base::subtle::Acquire_Load(&g_category_index);
  for (int i = 0; i < category_index; ++i) {
    if (strcmp(g_category_groups[i], category_group) == 0) {
      return &g_category_group_enabled[i];
    }
  }

  // Create a new category group.
  DCHECK(category_index < MAX_CATEGORY_GROUPS) <<
      "must increase MAX_CATEGORY_GROUPS";
  if (category_index < MAX_CATEGORY_GROUPS) {
    // Don't hold on to the category_group pointer, so that we can create
    // category groups with strings not known at compile time (this is
    // required by SetWatchEvent).
    const char* new_group = strdup(category_group);
    // NOTE: new_group is leaked, but this is a small finite amount of data
    g_category_groups[category_index] = new_group;
    DCHECK(!g_category_group_enabled[category_index]);
    // Note that if both included and excluded patterns in the
    // CategoryFilter are empty, we exclude nothing,
    // thereby enabling this category group.
    UpdateCategoryGroupEnabledFlag(category_index);
    category_group_enabled = &g_category_group_enabled[category_index];
    // Update the max index now.
    base::subtle::Release_Store(&g_category_index, category_index + 1);
  } else {
    category_group_enabled =
        &g_category_group_enabled[kCategoryCategoriesExhausted];
  }
  return category_group_enabled;
}

void TraceLog::GetKnownCategoryGroups(
    std::vector<std::string>* category_groups) {
  SpinLockHolder lock(&lock_);
  int category_index = base::subtle::NoBarrier_Load(&g_category_index);
  for (int i = kNumBuiltinCategories; i < category_index; i++)
    category_groups->push_back(g_category_groups[i]);
}

void TraceLog::SetEnabled(const CategoryFilter& category_filter,
                          Mode mode,
                          Options options) {
  std::vector<EnabledStateObserver*> observer_list;
  {
    SpinLockHolder lock(&lock_);

    // Can't enable tracing when Flush() is in progress.
    Options old_options = trace_options();

    if (IsEnabled()) {
      if (options != old_options) {
        DLOG(ERROR) << "Attempting to re-enable tracing with a different "
                    << "set of options.";
      }

      if (mode != mode_) {
        DLOG(ERROR) << "Attempting to re-enable tracing with a different mode.";
      }

      category_filter_.Merge(category_filter);
      UpdateCategoryGroupEnabledFlags();
      return;
    }

    if (dispatching_to_observer_list_) {
      DLOG(ERROR) <<
          "Cannot manipulate TraceLog::Enabled state from an observer.";
      return;
    }

    mode_ = mode;

    if (options != old_options) {
      base::subtle::NoBarrier_Store(&trace_options_, options);
      UseNextTraceBuffer();
    }

    num_traces_recorded_++;

    category_filter_ = CategoryFilter(category_filter);
    UpdateCategoryGroupEnabledFlags();
    UpdateSyntheticDelaysFromCategoryFilter();

    if (options & ENABLE_SAMPLING) {
      sampling_thread_.reset(new TraceSamplingThread);
      sampling_thread_->RegisterSampleBucket(
          &g_trace_state[0],
          "bucket0",
          &TraceSamplingThread::DefaultSamplingCallback);
      sampling_thread_->RegisterSampleBucket(
          &g_trace_state[1],
          "bucket1",
          &TraceSamplingThread::DefaultSamplingCallback);
      sampling_thread_->RegisterSampleBucket(
          &g_trace_state[2],
          "bucket2",
          &TraceSamplingThread::DefaultSamplingCallback);

      Status s = Thread::CreateWithFlags(
          "tracing", "sampler",
          [this]() { this->sampling_thread_->ThreadMain(); },
          Thread::NO_STACK_WATCHDOG, &sampling_thread_handle_);
      if (!s.ok()) {
        LOG(DFATAL) << "failed to create trace sampling thread: " << s.ToString();
      }
    }

    dispatching_to_observer_list_ = true;
    observer_list = enabled_state_observer_list_;
  }
  // Notify observers outside the lock in case they trigger trace events.
  for (const auto& observer : observer_list)
    observer->OnTraceLogEnabled();

  {
    SpinLockHolder lock(&lock_);
    dispatching_to_observer_list_ = false;
  }
}

CategoryFilter TraceLog::GetCurrentCategoryFilter() {
  SpinLockHolder lock(&lock_);
  return category_filter_;
}

void TraceLog::SetDisabled() {
  SpinLockHolder lock(&lock_);
  SetDisabledWhileLocked();
}

void TraceLog::SetDisabledWhileLocked() {
  DCHECK(lock_.IsHeld());

  if (!IsEnabled())
    return;

  if (dispatching_to_observer_list_) {
    DLOG(ERROR)
        << "Cannot manipulate TraceLog::Enabled state from an observer.";
    return;
  }

  mode_ = DISABLED;

  if (sampling_thread_.get()) {
    // Stop the sampling thread.
    sampling_thread_->Stop();
    lock_.Unlock();
    sampling_thread_handle_->Join();
    lock_.Lock();
    sampling_thread_handle_.reset();
    sampling_thread_.reset();
  }

  category_filter_.Clear();
  base::subtle::NoBarrier_Store(&watch_category_, 0);
  watch_event_name_ = "";
  UpdateCategoryGroupEnabledFlags();
  AddMetadataEventsWhileLocked();

  dispatching_to_observer_list_ = true;
  std::vector<EnabledStateObserver*> observer_list =
      enabled_state_observer_list_;

  {
    // Dispatch to observers outside the lock in case the observer triggers a
    // trace event.
    lock_.Unlock();
    for (const auto& observer : observer_list)
      observer->OnTraceLogDisabled();
    lock_.Lock();
  }
  dispatching_to_observer_list_ = false;
}

int TraceLog::GetNumTracesRecorded() {
  SpinLockHolder lock(&lock_);
  if (!IsEnabled())
    return -1;
  return num_traces_recorded_;
}

void TraceLog::AddEnabledStateObserver(EnabledStateObserver* listener) {
  enabled_state_observer_list_.push_back(listener);
}

void TraceLog::RemoveEnabledStateObserver(EnabledStateObserver* listener) {
  auto it = std::find(enabled_state_observer_list_.begin(),
                      enabled_state_observer_list_.end(), listener);
  if (it != enabled_state_observer_list_.end())
    enabled_state_observer_list_.erase(it);
}

bool TraceLog::HasEnabledStateObserver(EnabledStateObserver* listener) const {
  auto it = std::find(enabled_state_observer_list_.begin(),
                      enabled_state_observer_list_.end(), listener);
  return it != enabled_state_observer_list_.end();
}

float TraceLog::GetBufferPercentFull() const {
  SpinLockHolder lock(&lock_);
  return static_cast<float>(static_cast<double>(logged_events_->Size()) /
                            logged_events_->Capacity());
}

bool TraceLog::BufferIsFull() const {
  SpinLockHolder lock(&lock_);
  return logged_events_->IsFull();
}

TraceBuffer* TraceLog::CreateTraceBuffer() {
  Options options = trace_options();
  if (options & RECORD_CONTINUOUSLY)
    return new TraceBufferRingBuffer(kTraceEventRingBufferChunks);
  else if ((options & ENABLE_SAMPLING) && mode_ == MONITORING_MODE)
    return new TraceBufferRingBuffer(kMonitorTraceEventBufferChunks);
  else if (options & ECHO_TO_CONSOLE)
    return new TraceBufferRingBuffer(kEchoToConsoleTraceEventBufferChunks);
  return new TraceBufferVector();
}

TraceEvent* TraceLog::AddEventToThreadSharedChunkWhileLocked(
    TraceEventHandle* handle, bool check_buffer_is_full) {
  DCHECK(lock_.IsHeld());

  if (thread_shared_chunk_ && thread_shared_chunk_->IsFull()) {
    logged_events_->ReturnChunk(thread_shared_chunk_index_,
                                std::move(thread_shared_chunk_));
  }

  if (!thread_shared_chunk_) {
    thread_shared_chunk_ = logged_events_->GetChunk(
        &thread_shared_chunk_index_);
    if (check_buffer_is_full)
      CheckIfBufferIsFullWhileLocked();
  }
  if (!thread_shared_chunk_)
    return nullptr;

  size_t event_index;
  TraceEvent* trace_event = thread_shared_chunk_->AddTraceEvent(&event_index);
  if (trace_event && handle) {
    MakeHandle(thread_shared_chunk_->seq(), thread_shared_chunk_index_,
               event_index, handle);
  }
  return trace_event;
}

void TraceLog::CheckIfBufferIsFullWhileLocked() {
  DCHECK(lock_.IsHeld());
  if (logged_events_->IsFull())
    SetDisabledWhileLocked();
}

void TraceLog::SetEventCallbackEnabled(const CategoryFilter& category_filter,
                                       EventCallback cb) {
  SpinLockHolder lock(&lock_);
  base::subtle::NoBarrier_Store(&event_callback_,
                          reinterpret_cast<AtomicWord>(cb));
  event_callback_category_filter_ = category_filter;
  UpdateCategoryGroupEnabledFlags();
};

void TraceLog::SetEventCallbackDisabled() {
  SpinLockHolder lock(&lock_);
  base::subtle::NoBarrier_Store(&event_callback_, 0);
  UpdateCategoryGroupEnabledFlags();
}

// Flush() works as the following:
//
// We ensure by taking the global lock that we have exactly one Flusher thread
// (the caller of this function) and some number of "target" threads. We do
// not want to block the target threads, since they are running application code,
// so this implementation takes an approach based on asymmetric synchronization.
//
// For each active thread, we grab its PerThreadInfo object, which may contain
// a pointer to its active trace chunk. We use an AtomicExchange to swap this
// out for a null pointer. This ensures that, on the *next* TRACE call made by
// that thread, it will see a NULL buffer and create a _new_ trace buffer. That
// new buffer would be assigned the generation of the next collection and we don't
// have to worry about it in the current Flush().
//
// However, the swap doesn't ensure that the thread doesn't already have a local copy of
// the 'event_buffer_' that we are trying to flush. So, if the thread is in the
// middle of a Trace call, we have to wait until it exits. We do that by spinning
// on the 'is_in_trace_event_' member of that thread's thread-local structure.
//
// After we've swapped the buffer pointer and waited on the thread to exit any
// concurrent Trace() call, we know that no other thread can hold a pointer to
// the trace buffer, and we can safely flush it and delete it.
void TraceLog::Flush(const TraceLog::OutputCallback& cb) {
  if (IsEnabled()) {
    // Can't flush when tracing is enabled because otherwise PostTask would
    // - generate more trace events;
    // - deschedule the calling thread on some platforms causing inaccurate
    //   timing of the trace events.
    scoped_refptr<RefCountedString> empty_result = new RefCountedString;
    if (cb)
      cb(empty_result, false);
    LOG(WARNING) << "Ignored TraceLog::Flush called when tracing is enabled";
    return;
  }

  int generation = this->generation();
  {
    // Holding the active threads lock ensures that no thread will exit and
    // delete its own PerThreadInfo object.
    MutexLock l(active_threads_lock_);
    for (const ActiveThreadMap::value_type& entry : active_threads_) {
      int64_t tid = entry.first;
      PerThreadInfo* thr_info = entry.second;

      // Swap out their buffer from their thread-local data.
      // After this, any _future_ trace calls on that thread will create a new buffer
      // and not use the one we obtain here.
      ThreadLocalEventBuffer* buf = thr_info->AtomicTakeBuffer();

      // If this thread hasn't traced anything since our last
      // flush, we can skip it.
      if (!buf) {
        continue;
      }

      // The buffer may still be in use by that thread if they're in a call. Sleep until
      // they aren't, so we can flush/delete their old buffer.
      //
      // It's important that we do not hold 'lock_' here, because otherwise we can get a
      // deadlock: a thread may be in the middle of a trace event (is_in_trace_event_ ==
      // true) and waiting to take lock_, while we are holding the lock and waiting for it
      // to not be in the trace event.
      while (base::subtle::Acquire_Load(&thr_info->is_in_trace_event_)) {
        sched_yield();
      }

      {
        SpinLockHolder lock(&lock_);
        buf->Flush(tid);
      }
      delete buf;
    }
  }

  {
    SpinLockHolder lock(&lock_);

    if (thread_shared_chunk_) {
      logged_events_->ReturnChunk(thread_shared_chunk_index_,
                                  std::move(thread_shared_chunk_));
    }
  }

  FinishFlush(generation, cb);
}

void TraceLog::ConvertTraceEventsToTraceFormat(
    unique_ptr<TraceBuffer> logged_events,
    const TraceLog::OutputCallback& flush_output_callback) {

  if (!flush_output_callback)
    return;

  // The callback need to be called at least once even if there is no events
  // to let the caller know the completion of flush.
  bool has_more_events = true;
  do {
    scoped_refptr<RefCountedString> json_events_str_ptr =
        new RefCountedString();

    for (size_t i = 0; i < kTraceEventBatchChunks; ++i) {
      const TraceBufferChunk* chunk = logged_events->NextChunk();
      if (!chunk) {
        has_more_events = false;
        break;
      }
      for (size_t j = 0; j < chunk->size(); ++j) {
        if (i > 0 || j > 0)
          json_events_str_ptr->data().append(",");
        chunk->GetEventAt(j)->AppendAsJSON(&(json_events_str_ptr->data()));
      }
    }

    flush_output_callback(json_events_str_ptr, has_more_events);
  } while (has_more_events);
  logged_events.reset();
}

void TraceLog::FinishFlush(int generation,
                           const TraceLog::OutputCallback& flush_output_callback) {
  unique_ptr<TraceBuffer> previous_logged_events;

  if (!CheckGeneration(generation))
    return;

  {
    SpinLockHolder lock(&lock_);

    previous_logged_events.swap(logged_events_);
    UseNextTraceBuffer();
  }

  ConvertTraceEventsToTraceFormat(std::move(previous_logged_events),
                                  flush_output_callback);
}

void TraceLog::FlushButLeaveBufferIntact(
    const TraceLog::OutputCallback& flush_output_callback) {
  unique_ptr<TraceBuffer> previous_logged_events;
  {
    SpinLockHolder lock(&lock_);
    if (mode_ == DISABLED || (trace_options_ & RECORD_CONTINUOUSLY) == 0) {
      scoped_refptr<RefCountedString> empty_result = new RefCountedString;
      flush_output_callback(empty_result, false);
      LOG(WARNING) << "Ignored TraceLog::FlushButLeaveBufferIntact when monitoring is not enabled";
      return;
    }

    AddMetadataEventsWhileLocked();
    if (thread_shared_chunk_) {
      // Return the chunk to the main buffer to flush the sampling data.
      logged_events_->ReturnChunk(thread_shared_chunk_index_,
                                  std::move(thread_shared_chunk_));
    }
    previous_logged_events = logged_events_->CloneForIteration();
  }

  ConvertTraceEventsToTraceFormat(std::move(previous_logged_events),
                                  flush_output_callback);
}

void TraceLog::UseNextTraceBuffer() {
  logged_events_.reset(CreateTraceBuffer());
  base::subtle::NoBarrier_AtomicIncrement(&generation_, 1);
  thread_shared_chunk_.reset();
  thread_shared_chunk_index_ = 0;
}

TraceEventHandle TraceLog::AddTraceEvent(
    char phase,
    const unsigned char* category_group_enabled,
    const char* name,
    uint64_t id,
    int num_args,
    const char** arg_names,
    const unsigned char* arg_types,
    const uint64_t* arg_values,
    const scoped_refptr<ConvertableToTraceFormat>* convertable_values,
    unsigned char flags) {
  int thread_id = static_cast<int>(kudu::Thread::UniqueThreadId());
  MicrosecondsInt64 now = GetMonoTimeMicros();
  return AddTraceEventWithThreadIdAndTimestamp(phase, category_group_enabled,
                                               name, id, thread_id, now,
                                               num_args, arg_names,
                                               arg_types, arg_values,
                                               convertable_values, flags);
}

TraceLog::PerThreadInfo* TraceLog::SetupThreadLocalBuffer() {
  int64_t cur_tid = Thread::UniqueThreadId();

  auto thr_info = new PerThreadInfo();
  thr_info->event_buffer_ = nullptr;
  thr_info->is_in_trace_event_ = 0;
  thread_local_info_ = thr_info;

  threadlocal::internal::AddDestructor(&TraceLog::ThreadExitingCB, this);

  {
    MutexLock lock(active_threads_lock_);
    InsertOrDie(&active_threads_, cur_tid, thr_info);
  }
  return thr_info;
}

void TraceLog::ThreadExitingCB(void* arg) {
  static_cast<TraceLog*>(arg)->ThreadExiting();
}

void TraceLog::ThreadExiting() {
  PerThreadInfo* thr_info = thread_local_info_;
  if (!thr_info) {
    return;
  }

  int64_t cur_tid = Thread::UniqueThreadId();

  // Flush our own buffer back to the central event buffer.
  // We do the atomic exchange because a flusher thread may
  // also be trying to flush us at the same time, and we need to avoid
  // conflict.
  ThreadLocalEventBuffer* buf = thr_info->AtomicTakeBuffer();
  if (buf) {
    SpinLockHolder lock(&lock_);
    buf->Flush(Thread::UniqueThreadId());
  }
  delete buf;

  {
    MutexLock lock(active_threads_lock_);
    active_threads_.erase(cur_tid);
  }
  delete thr_info;
}

TraceEventHandle TraceLog::AddTraceEventWithThreadIdAndTimestamp(
    char phase,
    const unsigned char* category_group_enabled,
    const char* name,
    uint64_t id,
    int thread_id,
    const MicrosecondsInt64& timestamp,
    int num_args,
    const char** arg_names,
    const unsigned char* arg_types,
    const uint64_t* arg_values,
    const scoped_refptr<ConvertableToTraceFormat>* convertable_values,
    unsigned char flags) {
  TraceEventHandle handle = { 0, 0, 0 };
  if (!*category_group_enabled)
    return handle;

  DCHECK(name);

  if (flags & TRACE_EVENT_FLAG_MANGLE_ID)
    id ^= process_id_hash_;

  MicrosecondsInt64 now = OffsetTimestamp(timestamp);
  MicrosecondsInt64 thread_now = GetThreadCpuTimeMicros();

  PerThreadInfo* thr_info = thread_local_info_;
  if (PREDICT_FALSE(!thr_info)) {
    thr_info = SetupThreadLocalBuffer();
  }

  // Avoid re-entrance of AddTraceEvent. This may happen in GPU process when
  // ECHO_TO_CONSOLE is enabled: AddTraceEvent -> LOG(ERROR) ->
  // GpuProcessLogMessageHandler -> PostPendingTask -> TRACE_EVENT ...
  if (base::subtle::NoBarrier_Load(&thr_info->is_in_trace_event_))
    return handle;

  MarkFlagInScope thread_is_in_trace_event(&thr_info->is_in_trace_event_);

  ThreadLocalEventBuffer* thread_local_event_buffer =
    reinterpret_cast<ThreadLocalEventBuffer*>(
      base::subtle::NoBarrier_Load(
        reinterpret_cast<AtomicWord*>(&thr_info->event_buffer_)));

  // If we have an event buffer, but it's a left-over from a previous trace,
  // delete it.
  if (PREDICT_FALSE(thread_local_event_buffer &&
                    !CheckGeneration(thread_local_event_buffer->generation()))) {
    // We might also race against a flusher thread, so we have to atomically
    // take the buffer.
    thread_local_event_buffer = thr_info->AtomicTakeBuffer();
    delete thread_local_event_buffer;
    thread_local_event_buffer = nullptr;
  }

  // If there is no current buffer, create one for this event.
  if (PREDICT_FALSE(!thread_local_event_buffer)) {
    thread_local_event_buffer = new ThreadLocalEventBuffer(this);

    base::subtle::NoBarrier_Store(
      reinterpret_cast<AtomicWord*>(&thr_info->event_buffer_),
      reinterpret_cast<AtomicWord>(thread_local_event_buffer));
  }

  // Check and update the current thread name only if the event is for the
  // current thread to avoid locks in most cases.
  if (thread_id == static_cast<int>(Thread::UniqueThreadId())) {
    Thread* kudu_thr = Thread::current_thread();
    if (kudu_thr) {
      const char* new_name = kudu_thr->name().c_str();
      // Check if the thread name has been set or changed since the previous
      // call (if any), but don't bother if the new name is empty. Note this will
      // not detect a thread name change within the same char* buffer address: we
      // favor common case performance over corner case correctness.
      if (PREDICT_FALSE(new_name != g_current_thread_name &&
                        new_name && *new_name)) {
        g_current_thread_name = new_name;

        SpinLockHolder thread_info_lock(&thread_info_lock_);

        auto existing_name = thread_names_.find(thread_id);
        if (existing_name == thread_names_.end()) {
          // This is a new thread id, and a new name.
          thread_names_[thread_id] = new_name;
        } else {
          // This is a thread id that we've seen before, but potentially with a
          // new name.
          std::vector<StringPiece> existing_names = strings::Split(existing_name->second, ",");
          bool found = std::find(existing_names.begin(),
                                 existing_names.end(),
                                 new_name) != existing_names.end();
          if (!found) {
            if (existing_names.size())
              existing_name->second.push_back(',');
            existing_name->second.append(new_name);
          }
        }
      }
    }
  }

  std::string console_message;
  if (*category_group_enabled &
      (ENABLED_FOR_RECORDING | ENABLED_FOR_MONITORING)) {
    TraceEvent* trace_event = thread_local_event_buffer->AddTraceEvent(&handle);

    if (trace_event) {
      trace_event->Initialize(thread_id, now, thread_now, phase,
                              category_group_enabled, name, id,
                              num_args, arg_names, arg_types, arg_values,
                              convertable_values, flags);

#if defined(OS_ANDROID)
      trace_event->SendToATrace();
#endif
    }

    if (trace_options() & ECHO_TO_CONSOLE) {
      console_message = EventToConsoleMessage(
          phase == TRACE_EVENT_PHASE_COMPLETE ? TRACE_EVENT_PHASE_BEGIN : phase,
          timestamp, trace_event);
    }
  }

  if (PREDICT_FALSE(console_message.size()))
    LOG(ERROR) << console_message;

  if (PREDICT_FALSE(reinterpret_cast<const unsigned char*>(
                      base::subtle::NoBarrier_Load(&watch_category_)) == category_group_enabled)) {
    bool event_name_matches;
    WatchEventCallback watch_event_callback_copy;
    {
      SpinLockHolder lock(&lock_);
      event_name_matches = watch_event_name_ == name;
      watch_event_callback_copy = watch_event_callback_;
    }
    if (event_name_matches) {
      if (watch_event_callback_copy)
        watch_event_callback_copy();
    }
  }

  if (PREDICT_FALSE(*category_group_enabled & ENABLED_FOR_EVENT_CALLBACK)) {
    EventCallback event_callback = reinterpret_cast<EventCallback>(
      base::subtle::NoBarrier_Load(&event_callback_));
    if (event_callback) {
      event_callback(now,
                     phase == TRACE_EVENT_PHASE_COMPLETE ?
                         TRACE_EVENT_PHASE_BEGIN : phase,
                     category_group_enabled, name, id,
                     num_args, arg_names, arg_types, arg_values,
                     flags);
    }
  }

  return handle;
}

// May be called when a COMPELETE event ends and the unfinished event has been
// recycled (phase == TRACE_EVENT_PHASE_END and trace_event == NULL).
std::string TraceLog::EventToConsoleMessage(unsigned char phase,
                                            const MicrosecondsInt64& timestamp,
                                            TraceEvent* trace_event) {
  SpinLockHolder thread_info_lock(&thread_info_lock_);

  // The caller should translate TRACE_EVENT_PHASE_COMPLETE to
  // TRACE_EVENT_PHASE_BEGIN or TRACE_EVENT_END.
  DCHECK(phase != TRACE_EVENT_PHASE_COMPLETE);

  MicrosecondsInt64 duration = 0;
  int thread_id = trace_event ?
      trace_event->thread_id() : Thread::UniqueThreadId();
  if (phase == TRACE_EVENT_PHASE_END) {
    duration = timestamp - thread_event_start_times_[thread_id].top();
    thread_event_start_times_[thread_id].pop();
  }

  std::string thread_name = thread_names_[thread_id];
  if (thread_colors_.find(thread_name) == thread_colors_.end())
    thread_colors_[thread_name] = (thread_colors_.size() % 6) + 1;

  std::ostringstream log;
  log << StringPrintf("%s: \x1b[0;3%dm",
                            thread_name.c_str(),
                            thread_colors_[thread_name]);

  size_t depth = 0;
  if (thread_event_start_times_.find(thread_id) !=
      thread_event_start_times_.end())
    depth = thread_event_start_times_[thread_id].size();

  for (size_t i = 0; i < depth; ++i)
    log << "| ";

  if (trace_event)
    trace_event->AppendPrettyPrinted(&log);
  if (phase == TRACE_EVENT_PHASE_END)
    log << StringPrintf(" (%.3f ms)", duration / 1000.0f);

  log << "\x1b[0;m";

  if (phase == TRACE_EVENT_PHASE_BEGIN)
    thread_event_start_times_[thread_id].push(timestamp);

  return log.str();
}

void TraceLog::AddTraceEventEtw(char phase,
                                const char* name,
                                const void* id,
                                const char* extra) {
#if defined(OS_WIN)
  TraceEventETWProvider::Trace(name, phase, id, extra);
#endif
  INTERNAL_TRACE_EVENT_ADD(phase, "ETW Trace Event", name,
                           TRACE_EVENT_FLAG_COPY, "id", id, "extra", extra);
}

void TraceLog::AddTraceEventEtw(char phase,
                                const char* name,
                                const void* id,
                                const std::string& extra) {
#if defined(OS_WIN)
  TraceEventETWProvider::Trace(name, phase, id, extra);
#endif
  INTERNAL_TRACE_EVENT_ADD(phase, "ETW Trace Event", name,
                           TRACE_EVENT_FLAG_COPY, "id", id, "extra", extra);
}

void TraceLog::UpdateTraceEventDuration(
    const unsigned char* category_group_enabled,
    const char* name,
    TraceEventHandle handle) {

  PerThreadInfo* thr_info = thread_local_info_;
  if (!thr_info) {
    thr_info = SetupThreadLocalBuffer();
  }

  // Avoid re-entrance of AddTraceEvent. This may happen in GPU process when
  // ECHO_TO_CONSOLE is enabled: AddTraceEvent -> LOG(ERROR) ->
  // GpuProcessLogMessageHandler -> PostPendingTask -> TRACE_EVENT ...
  if (base::subtle::NoBarrier_Load(&thr_info->is_in_trace_event_))
    return;
  MarkFlagInScope thread_is_in_trace_event(&thr_info->is_in_trace_event_);

  MicrosecondsInt64 thread_now = GetThreadCpuTimeMicros();
  MicrosecondsInt64 now = OffsetNow();

  std::string console_message;
  if (*category_group_enabled & ENABLED_FOR_RECORDING) {
    OptionalAutoLock lock(lock_);

    TraceEvent* trace_event = GetEventByHandleInternal(handle, &lock);
    if (trace_event) {
      DCHECK(trace_event->phase() == TRACE_EVENT_PHASE_COMPLETE);
      trace_event->UpdateDuration(now, thread_now);
#if defined(OS_ANDROID)
      trace_event->SendToATrace();
#endif
    }

    if (trace_options() & ECHO_TO_CONSOLE) {
      console_message = EventToConsoleMessage(TRACE_EVENT_PHASE_END,
                                              now, trace_event);
    }
  }

  if (console_message.size())
    LOG(ERROR) << console_message;

  if (*category_group_enabled & ENABLED_FOR_EVENT_CALLBACK) {
    EventCallback event_callback = reinterpret_cast<EventCallback>(
      base::subtle::NoBarrier_Load(&event_callback_));
    if (event_callback) {
      event_callback(now, TRACE_EVENT_PHASE_END, category_group_enabled, name,
                     trace_event_internal::kNoEventId, 0, nullptr, nullptr, nullptr,
                     TRACE_EVENT_FLAG_NONE);
    }
  }
}

void TraceLog::SetWatchEvent(const std::string& category_name,
                             const std::string& event_name,
                             const WatchEventCallback& callback) {
  const unsigned char* category = GetCategoryGroupEnabled(
      category_name.c_str());
  SpinLockHolder lock(&lock_);
  base::subtle::NoBarrier_Store(&watch_category_,
                          reinterpret_cast<AtomicWord>(category));
  watch_event_name_ = event_name;
  watch_event_callback_ = callback;
}

void TraceLog::CancelWatchEvent() {
  SpinLockHolder lock(&lock_);
  base::subtle::NoBarrier_Store(&watch_category_, 0);
  watch_event_name_ = "";
  watch_event_callback_ = nullptr;
}

void TraceLog::AddMetadataEventsWhileLocked() {
  DCHECK(lock_.IsHeld());

#if !defined(OS_NACL)  // NaCl shouldn't expose the process id.
  InitializeMetadataEvent(AddEventToThreadSharedChunkWhileLocked(nullptr, false),
                          0,
                          "num_cpus", "number",
                          base::NumCPUs());
#endif


  int current_thread_id = static_cast<int>(kudu::Thread::UniqueThreadId());
  if (process_sort_index_ != 0) {
    InitializeMetadataEvent(AddEventToThreadSharedChunkWhileLocked(nullptr, false),
                            current_thread_id,
                            "process_sort_index", "sort_index",
                            process_sort_index_);
  }

  if (process_name_.size()) {
    InitializeMetadataEvent(AddEventToThreadSharedChunkWhileLocked(nullptr, false),
                            current_thread_id,
                            "process_name", "name",
                            process_name_);
  }

  if (process_labels_.size() > 0) {
    std::vector<std::string> labels;
    for(auto& label : process_labels_) {
      labels.push_back(label.second);
    }
    InitializeMetadataEvent(AddEventToThreadSharedChunkWhileLocked(nullptr, false),
                            current_thread_id,
                            "process_labels", "labels",
                            JoinStrings(labels, ","));
  }

  // Thread sort indices.
  for(auto& sort_index : thread_sort_indices_) {
    if (sort_index.second == 0)
      continue;
    InitializeMetadataEvent(AddEventToThreadSharedChunkWhileLocked(nullptr, false),
                            sort_index.first,
                            "thread_sort_index", "sort_index",
                            sort_index.second);
  }

  // Thread names.
  SpinLockHolder thread_info_lock(&thread_info_lock_);
  for(auto& name : thread_names_) {
    if (name.second.empty())
      continue;
    InitializeMetadataEvent(AddEventToThreadSharedChunkWhileLocked(nullptr, false),
                            name.first,
                            "thread_name", "name",
                            name.second);
  }
}


TraceEvent* TraceLog::GetEventByHandle(TraceEventHandle handle) {
  return GetEventByHandleInternal(handle, nullptr);
}

TraceEvent* TraceLog::GetEventByHandleInternal(TraceEventHandle handle,
                                               OptionalAutoLock* lock) {
  TraceLog::PerThreadInfo* thr_info = TraceLog::thread_local_info_;

  if (!handle.chunk_seq)
    return nullptr;

  if (thr_info) {
    ThreadLocalEventBuffer* buf =
      reinterpret_cast<ThreadLocalEventBuffer*>(
        base::subtle::NoBarrier_Load(
          reinterpret_cast<AtomicWord*>(&thr_info->event_buffer_)));

    if (buf) {
      DCHECK_EQ(1, ANNOTATE_UNPROTECTED_READ(thr_info->is_in_trace_event_));

      TraceEvent* trace_event = buf->GetEventByHandle(handle);
      if (trace_event)
        return trace_event;
    }
  }

  // The event has been out-of-control of the thread local buffer.
  // Try to get the event from the main buffer with a lock.
  if (lock)
    lock->EnsureAcquired();

  if (thread_shared_chunk_ &&
      handle.chunk_index == thread_shared_chunk_index_) {
    return handle.chunk_seq == thread_shared_chunk_->seq() ?
        thread_shared_chunk_->GetEventAt(handle.event_index) : nullptr;
  }

  return logged_events_->GetEventByHandle(handle);
}

ATTRIBUTE_NO_SANITIZE_INTEGER
void TraceLog::SetProcessID(int process_id) {
  process_id_ = process_id;
  // Create a FNV hash from the process ID for XORing.
  // See http://isthe.com/chongo/tech/comp/fnv/ for algorithm details.
  uint64_t offset_basis = 14695981039346656037ull;
  uint64_t fnv_prime = 1099511628211ull;
  uint64_t pid = static_cast<uint64_t>(process_id_);
  process_id_hash_ = (offset_basis ^ pid) * fnv_prime;
}

void TraceLog::SetProcessSortIndex(int sort_index) {
  SpinLockHolder lock(&lock_);
  process_sort_index_ = sort_index;
}

void TraceLog::SetProcessName(const std::string& process_name) {
  SpinLockHolder lock(&lock_);
  process_name_ = process_name;
}

void TraceLog::UpdateProcessLabel(
    int label_id, const std::string& current_label) {
  if(!current_label.length())
    return RemoveProcessLabel(label_id);

  SpinLockHolder lock(&lock_);
  process_labels_[label_id] = current_label;
}

void TraceLog::RemoveProcessLabel(int label_id) {
  SpinLockHolder lock(&lock_);
  auto it = process_labels_.find(label_id);
  if (it == process_labels_.end())
    return;

  process_labels_.erase(it);
}

void TraceLog::SetThreadSortIndex(int64_t thread_id, int sort_index) {
  SpinLockHolder lock(&lock_);
  thread_sort_indices_[static_cast<int>(thread_id)] = sort_index;
}

void TraceLog::SetTimeOffset(MicrosecondsInt64 offset) {
  time_offset_ = offset;
}

size_t TraceLog::GetObserverCountForTest() const {
  return enabled_state_observer_list_.size();
}

bool CategoryFilter::IsEmptyOrContainsLeadingOrTrailingWhitespace(
    const std::string& str) {
  return  str.empty() ||
          str.at(0) == ' ' ||
          str.at(str.length() - 1) == ' ';
}

bool CategoryFilter::DoesCategoryGroupContainCategory(
    const char* category_group,
    const char* category) const {
  DCHECK(category);
  vector<string> pieces = strings::Split(category_group, ",");
  for (const string& category_group_token : pieces) {
    // Don't allow empty tokens, nor tokens with leading or trailing space.
    DCHECK(!CategoryFilter::IsEmptyOrContainsLeadingOrTrailingWhitespace(
        category_group_token))
        << "Disallowed category string";

    if (MatchPattern(category_group_token.c_str(), category))
      return true;
  }
  return false;
}

CategoryFilter::CategoryFilter(const std::string& filter_string) {
  if (!filter_string.empty())
    Initialize(filter_string);
  else
    Initialize(CategoryFilter::kDefaultCategoryFilterString);
}

CategoryFilter::CategoryFilter(const CategoryFilter& cf)
    : included_(cf.included_),
      disabled_(cf.disabled_),
      excluded_(cf.excluded_),
      delays_(cf.delays_) {
}

CategoryFilter::~CategoryFilter() {
}

CategoryFilter& CategoryFilter::operator=(const CategoryFilter& rhs) {
  if (this == &rhs)
    return *this;

  included_ = rhs.included_;
  disabled_ = rhs.disabled_;
  excluded_ = rhs.excluded_;
  delays_ = rhs.delays_;
  return *this;
}

void CategoryFilter::Initialize(const std::string& filter_string) {
  // Tokenize list of categories, delimited by ','.
  vector<string> tokens = strings::Split(filter_string, ",");
  // Add each token to the appropriate list (included_,excluded_).
  for (string category : tokens) {
    // Ignore empty categories.
    if (category.empty())
      continue;
    // Synthetic delays are of the form 'DELAY(delay;option;option;...)'.
    if (category.find(kSyntheticDelayCategoryFilterPrefix) == 0 &&
        category.at(category.size() - 1) == ')') {
      category = category.substr(
          strlen(kSyntheticDelayCategoryFilterPrefix),
          category.size() - strlen(kSyntheticDelayCategoryFilterPrefix) - 1);
      size_t name_length = category.find(';');
      if (name_length != std::string::npos && name_length > 0 &&
          name_length != category.size() - 1) {
        delays_.push_back(category);
      }
    } else if (category.at(0) == '-') {
      // Excluded categories start with '-'.
      // Remove '-' from category string.
      category = category.substr(1);
      excluded_.push_back(category);
    } else if (category.compare(0, strlen(TRACE_DISABLED_BY_DEFAULT("")),
                                TRACE_DISABLED_BY_DEFAULT("")) == 0) {
      disabled_.push_back(category);
    } else {
      included_.push_back(category);
    }
  }
}

void CategoryFilter::WriteString(const StringList& values,
                                 std::string* out,
                                 bool included) const {
  bool prepend_comma = !out->empty();
  int token_cnt = 0;
  for (const auto& value : values) {
    if (token_cnt > 0 || prepend_comma)
      StringAppendF(out, ",");
    StringAppendF(out, "%s%s", (included ? "" : "-"), value.c_str());
    ++token_cnt;
  }
}

void CategoryFilter::WriteString(const StringList& delays,
                                 std::string* out) const {
  bool prepend_comma = !out->empty();
  int token_cnt = 0;
  for (const auto& delay : delays) {
    if (token_cnt > 0 || prepend_comma)
      StringAppendF(out, ",");
    StringAppendF(out, "%s%s)", kSyntheticDelayCategoryFilterPrefix,
                  delay.c_str());
    ++token_cnt;
  }
}

std::string CategoryFilter::ToString() const {
  std::string filter_string;
  WriteString(included_, &filter_string, true);
  WriteString(disabled_, &filter_string, true);
  WriteString(excluded_, &filter_string, false);
  WriteString(delays_, &filter_string);
  return filter_string;
}

bool CategoryFilter::IsCategoryGroupEnabled(
    const char* category_group_name) const {
  // TraceLog should call this method only as  part of enabling/disabling
  // categories.
  StringList::const_iterator ci;

  // Check the disabled- filters and the disabled-* wildcard first so that a
  // "*" filter does not include the disabled.
  for (ci = disabled_.begin(); ci != disabled_.end(); ++ci) {
    if (DoesCategoryGroupContainCategory(category_group_name, ci->c_str()))
      return true;
  }
  if (DoesCategoryGroupContainCategory(category_group_name,
                                       TRACE_DISABLED_BY_DEFAULT("*")))
    return false;

  for (ci = included_.begin(); ci != included_.end(); ++ci) {
    if (DoesCategoryGroupContainCategory(category_group_name, ci->c_str()))
      return true;
  }

  for (ci = excluded_.begin(); ci != excluded_.end(); ++ci) {
    if (DoesCategoryGroupContainCategory(category_group_name, ci->c_str()))
      return false;
  }
  // If the category group is not excluded, and there are no included patterns
  // we consider this pattern enabled.
  return included_.empty();
}

bool CategoryFilter::HasIncludedPatterns() const {
  return !included_.empty();
}

void CategoryFilter::Merge(const CategoryFilter& nested_filter) {
  // Keep included patterns only if both filters have an included entry.
  // Otherwise, one of the filter was specifying "*" and we want to honour the
  // broadest filter.
  if (HasIncludedPatterns() && nested_filter.HasIncludedPatterns()) {
    included_.insert(included_.end(),
                     nested_filter.included_.begin(),
                     nested_filter.included_.end());
  } else {
    included_.clear();
  }

  disabled_.insert(disabled_.end(),
                   nested_filter.disabled_.begin(),
                   nested_filter.disabled_.end());
  excluded_.insert(excluded_.end(),
                   nested_filter.excluded_.begin(),
                   nested_filter.excluded_.end());
  delays_.insert(delays_.end(),
                 nested_filter.delays_.begin(),
                 nested_filter.delays_.end());
}

void CategoryFilter::Clear() {
  included_.clear();
  disabled_.clear();
  excluded_.clear();
}

const CategoryFilter::StringList&
    CategoryFilter::GetSyntheticDelayValues() const {
  return delays_;
}

}  // namespace debug
}  // namespace kudu

namespace trace_event_internal {

ScopedTraceBinaryEfficient::ScopedTraceBinaryEfficient(
    const char* category_group, const char* name) {
  // The single atom works because for now the category_group can only be "gpu".
  DCHECK(strcmp(category_group, "gpu") == 0);
  static TRACE_EVENT_API_ATOMIC_WORD atomic = 0;
  INTERNAL_TRACE_EVENT_GET_CATEGORY_INFO_CUSTOM_VARIABLES(
      category_group, atomic, category_group_enabled_);
  name_ = name;
  if (*category_group_enabled_) {
    event_handle_ =
        TRACE_EVENT_API_ADD_TRACE_EVENT_WITH_THREAD_ID_AND_TIMESTAMP(
            TRACE_EVENT_PHASE_COMPLETE, category_group_enabled_, name,
            trace_event_internal::kNoEventId,
            static_cast<int>(kudu::Thread::UniqueThreadId()),
            GetMonoTimeMicros(),
            0, nullptr, nullptr, nullptr, nullptr, TRACE_EVENT_FLAG_NONE);
  }
}

ScopedTraceBinaryEfficient::~ScopedTraceBinaryEfficient() {
  if (*category_group_enabled_) {
    TRACE_EVENT_API_UPDATE_TRACE_EVENT_DURATION(category_group_enabled_,
                                                name_, event_handle_);
  }
}

}  // namespace trace_event_internal