Thread-local trace-event buffers

base/debug/trace_event_impl.cc

Index: base/debug/trace_event_impl.cc
diff --git a/base/debug/trace_event_impl.cc b/base/debug/trace_event_impl.cc
index cc22424f15e505c96dac6cdc356b404be6e87406..1b6bd38a5127659caecd433065c73fd60372689b 100644
--- a/base/debug/trace_event_impl.cc
+++ b/base/debug/trace_event_impl.cc
@@ -14,6 +14,7 @@
 #include "base/format_macros.h"
 #include "base/lazy_instance.h"
 #include "base/memory/singleton.h"
+#include "base/message_loop/message_loop.h"
 #include "base/process/process_metrics.h"
 #include "base/stl_util.h"
 #include "base/strings/string_split.h"
@@ -27,7 +28,6 @@
 #include "base/third_party/dynamic_annotations/dynamic_annotations.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/thread_id_name_manager.h"
-#include "base/threading/thread_local.h"
 #include "base/time/time.h"
 
 #if defined(OS_WIN)
@@ -48,16 +48,17 @@ BASE_EXPORT TRACE_EVENT_API_ATOMIC_WORD g_trace_state[3];
 namespace base {
 namespace debug {
 
+namespace {
+
 // Controls the number of trace events we will buffer in-memory
 // before throwing them away.
 const size_t kTraceEventBufferSize = 500000;
+const size_t kTraceEventThreadLocalBufferSize = 1024;
 const size_t kTraceEventBatchSize = 1000;
 const size_t kTraceEventInitialBufferSize = 1024;
 
 #define MAX_CATEGORY_GROUPS 100
 
-namespace {
-
 // 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
@@ -95,8 +96,6 @@ size_t NextIndex(size_t index) {
   return index;
 }
 
-}  // namespace
-
 class TraceBufferRingBuffer : public TraceBuffer {
  public:
   TraceBufferRingBuffer()
@@ -178,11 +177,14 @@ class TraceBufferVector : public TraceBuffer {
   }
 
   virtual void AddEvent(const TraceEvent& event) OVERRIDE {
-    // Note, we have two callers which need to be handled. The first is
-    // AddTraceEventWithThreadIdAndTimestamp() which checks Size() and does an
-    // early exit if full. The second is AddThreadNameMetadataEvents().
+    // Note, we have two callers which need to be handled:
+    // - AddEventToMainBufferWhileLocked() which has two cases:
+    //   - called directly from AddTraceEventWithThreadIdAndTimeStamp()
+    //     which checks if buffer is full and does an early exit if full;
+    //   - called from ThreadLocalEventBuffer::FlushWhileLocked();
+    // - AddThreadNameMetadataEvents().
     // We can not DECHECK(!IsFull()) because we have to add the metadata
-    // events even if the buffer is full.
+    // events and flush thread-local buffers even if the buffer is full.
     logged_events_.push_back(event);
   }
 
@@ -257,6 +259,8 @@ class TraceBufferDiscardsEvents : public TraceBuffer {
   }
 };
 
+}  // namespace
+
 ////////////////////////////////////////////////////////////////////////////////
 //
 // TraceEvent
@@ -741,6 +745,83 @@ TraceBucketData::~TraceBucketData() {
 //
 ////////////////////////////////////////////////////////////////////////////////
 
+class TraceLog::ThreadLocalEventBuffer
+    : public MessageLoop::DestructionObserver {
+ public:
+  ThreadLocalEventBuffer(TraceLog* trace_log);
+  virtual ~ThreadLocalEventBuffer();
+
+  void AddEvent(const TraceEvent& event, NotificationHelper* notifier);
+
+ private:
+
+  // MessageLoop::DestructionObserver
+  virtual void WillDestroyCurrentMessageLoop() OVERRIDE;
+
+  void FlushWhileLocked(NotificationHelper* notifier);
+
+  void CheckThread() {

[dsinclair, 2013/08/14 15:15:59]

Name is a bit generic, what about something like:

CheckThisIsCurrentBuffer() ?

[Xianzhu, 2013/08/14 21:28:34]

Done.

+    DCHECK(trace_log_->thread_local_event_buffer_.Get() == this);
+  }
+
+  TraceLog* trace_log_;

[dsinclair, 2013/08/14 15:15:59]

Are we making the assumption that, since TraceLog is a leaky singleton this will
always be valid as long as the thread exists?

If so, can we add a comment to that affect?

[Xianzhu, 2013/08/14 21:28:34]

Yes. Done.

+  std::vector<TraceEvent> logged_events_;
+
+  DISALLOW_COPY_AND_ASSIGN(ThreadLocalEventBuffer);
+};
+
+TraceLog::ThreadLocalEventBuffer::ThreadLocalEventBuffer(TraceLog* trace_log)
+    : trace_log_(trace_log) {
+  logged_events_.reserve(kTraceEventThreadLocalBufferSize);
+
+  MessageLoop* message_loop = MessageLoop::current();
+  DCHECK(message_loop);

[dsinclair, 2013/08/14 15:15:59]

What does it mean if this fails? Is it possible in Release mode that this will
happen?

[Xianzhu, 2013/08/14 21:28:34]

It never fails because ThreadLocalEventBuffer is created only for threads having
message loops. Removed the DCHECK and added comments.

+  message_loop->AddDestructionObserver(this);
+
+  AutoLock lock(trace_log->lock_);
+  trace_log->thread_message_loops_.insert(message_loop);
+}
+
+TraceLog::ThreadLocalEventBuffer::~ThreadLocalEventBuffer() {
+  CheckThread();
+  NotificationHelper notifier(trace_log_);
+  {
+    AutoLock lock(trace_log_->lock_);
+    FlushWhileLocked(&notifier);
+    trace_log_->thread_message_loops_.erase(MessageLoop::current());
+    if (trace_log_->flush_message_loop_) {
+      trace_log_->flush_message_loop_->PostTask(
+          FROM_HERE,
+          Bind(&TraceLog::FlushNextThreadOrFinish, Unretained(trace_log_)));
+    }
+  }
+  notifier.SendNotificationIfAny();
+  trace_log_->thread_local_event_buffer_.Set(NULL);
+}
+
+void TraceLog::ThreadLocalEventBuffer::AddEvent(const TraceEvent& event,
+                                                NotificationHelper* notifier) {
+  CheckThread();
+  logged_events_.push_back(event);
+  if (logged_events_.size() >= kTraceEventThreadLocalBufferSize) {
+    AutoLock lock(trace_log_->lock_);
+    FlushWhileLocked(notifier);
+  }
+}
+
+void TraceLog::ThreadLocalEventBuffer::WillDestroyCurrentMessageLoop() {
+  delete this;
+}
+
+void TraceLog::ThreadLocalEventBuffer::FlushWhileLocked(
+    NotificationHelper* notifier) {

[dsinclair, 2013/08/14 15:15:59]

lock_.AssertAcquired()?

[Xianzhu, 2013/08/14 21:28:34]

Done.

+  for (size_t i = 0; i < logged_events_.size(); ++i) {
+    trace_log_->AddEventToMainBufferWhileLocked(logged_events_[i],
+                                                notifier);
+  }
+  logged_events_.resize(0);
+}
+
 TraceLog::NotificationHelper::NotificationHelper(TraceLog* trace_log)
     : trace_log_(trace_log),
       notification_(0) {
@@ -751,6 +832,7 @@ TraceLog::NotificationHelper::~NotificationHelper() {
 
 void TraceLog::NotificationHelper::AddNotificationWhileLocked(
     int notification) {
+  trace_log_->lock_.AssertAcquired();
   if (trace_log_->notification_callback_.is_null())
     return;
   if (notification_ == 0)
@@ -794,16 +876,30 @@ TraceLog::Options TraceLog::TraceOptionsFromString(const std::string& options) {
   return static_cast<Options>(ret);
 }
 
+namespace {
+
+struct EchoToConsoleContext {

[dsinclair, 2013/08/14 15:15:59]

Can we move this up into the other anonymous namespace at the top of the file?

[Xianzhu, 2013/08/14 21:28:34]

Done.

+  hash_map<int, std::stack<TimeTicks> > thread_event_start_times;
+  hash_map<std::string, int> thread_colors;
+};
+
+}  // namespace
+
 TraceLog::TraceLog()
     : enable_count_(0),
       num_traces_recorded_(0),
-      event_callback_(NULL),
+      buffer_is_full_(0),
+      event_callback_(0),
       dispatching_to_observer_list_(false),
       process_sort_index_(0),
-      watch_category_(NULL),
+      echo_to_console_context_(0),
+      process_id_hash_(0),
+      process_id_(0),
+      watch_category_(0),
       trace_options_(RECORD_UNTIL_FULL),
       sampling_thread_handle_(0),
-      category_filter_(CategoryFilter::kDefaultCategoryFilterString) {
+      category_filter_(CategoryFilter::kDefaultCategoryFilterString),
+      flush_message_loop_(NULL) {
   // 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
@@ -839,6 +935,7 @@ TraceLog::TraceLog()
 }
 
 TraceLog::~TraceLog() {
+  delete reinterpret_cast<EchoToConsoleContext*>(echo_to_console_context_);
 }
 
 const unsigned char* TraceLog::GetCategoryGroupEnabled(
@@ -961,7 +1058,9 @@ void TraceLog::SetEnabled(const CategoryFilter& category_filter,
 
     if (options != trace_options_) {
       trace_options_ = options;
+      EnableEchoToConsole(options & ECHO_TO_CONSOLE);
       logged_events_.reset(GetTraceBuffer());
+      subtle::NoBarrier_Store(&buffer_is_full_, 0);
     }
 
     if (dispatching_to_observer_list_) {
@@ -1039,7 +1138,7 @@ void TraceLog::SetDisabled() {
     }
 
     category_filter_.Clear();
-    watch_category_ = NULL;
+    subtle::NoBarrier_Store(&watch_category_, 0);
     watch_event_name_ = "";
     UpdateCategoryGroupEnabledFlags();
     AddMetadataEvents();
@@ -1105,22 +1204,61 @@ TraceBuffer* TraceLog::GetTraceBuffer() {
   return new TraceBufferVector();
 }
 
+bool TraceLog::AddEventToMainBufferWhileLocked(const TraceEvent& trace_event,
+                                               NotificationHelper* notifier) {
+  // Don't check buffer_is_full_ because we want the remaining thread-local
+  // events to be flushed into the main buffer with this method, otherwise
+  // we may lose some early events of a thread that generates events sparsely.
+  lock_.AssertAcquired();
+  logged_events_->AddEvent(trace_event);
+  if (!subtle::NoBarrier_Load(&buffer_is_full_) && logged_events_->IsFull()) {
+    subtle::NoBarrier_Store(&buffer_is_full_,
+                            static_cast<subtle::AtomicWord>(1));
+    notifier->AddNotificationWhileLocked(TRACE_BUFFER_FULL);
+    return false;
+  }
+  return true;
+}
+
 void TraceLog::SetEventCallback(EventCallback cb) {
-  AutoLock lock(lock_);
-  event_callback_ = cb;
+  subtle::NoBarrier_Store(&event_callback_,
+                          reinterpret_cast<subtle::AtomicWord>(cb));
 };
 
 void TraceLog::Flush(const TraceLog::OutputCallback& cb) {
   // Ignore memory allocations from here down.
   INTERNAL_TRACE_MEMORY(TRACE_DISABLED_BY_DEFAULT("memory"),
                         TRACE_MEMORY_IGNORE);
+  {
+    AutoLock lock(lock_);
+    DCHECK(!flush_message_loop_);
+    flush_message_loop_ = MessageLoop::current();
+    DCHECK(flush_message_loop_);
+    flush_output_callback_ = cb;
+  }
+  FlushNextThreadOrFinish();
+}
+
+void TraceLog::FlushNextThreadOrFinish() {
   scoped_ptr<TraceBuffer> previous_logged_events;
   {
     AutoLock lock(lock_);
+    hash_set<MessageLoop*>::const_iterator next_message_loop =
+        thread_message_loops_.begin();
+    if (next_message_loop != thread_message_loops_.end()) {

[dsinclair, 2013/08/14 15:15:59]

The flush logic is, I think, the biggest confusion. Let me just check I've got
this right.

threadA kicks off the flush. It stores itself into flush_message_loop.

threadA then gets the first thread message loop and tells it to flush. That
thread, when finished will send a message to threadA to tell it to flush the
next one. This will work through all of the loops, including the message loop
for threadA and flush all the thread data.

threadA will then continue the rest of the flush procedure.

Correct? Would it be possible to get a comment here to explain the flushing?

[Xianzhu, 2013/08/14 21:28:34]

Exactly :)
Added comments.

+      (*next_message_loop)->PostTask(
+          FROM_HERE,
+          Bind(&TraceLog::DestroyThreadLocalEventBuffer, Unretained(this)));
+      return;
+    }
+
     previous_logged_events.swap(logged_events_);
     logged_events_.reset(GetTraceBuffer());
+    subtle::NoBarrier_Store(&buffer_is_full_, 0);
   }  // release lock
 
+  flush_message_loop_ = NULL;
+
   while (previous_logged_events->HasMoreEvents()) {
     scoped_refptr<RefCountedString> json_events_str_ptr =
         new RefCountedString();
@@ -1136,10 +1274,15 @@ void TraceLog::Flush(const TraceLog::OutputCallback& cb) {
         break;
     }
 
-    cb.Run(json_events_str_ptr);
+    flush_output_callback_.Run(json_events_str_ptr);
   }
 }
 
+void TraceLog::DestroyThreadLocalEventBuffer() {
+  delete thread_local_event_buffer_.Get();
+  thread_local_event_buffer_.Set(NULL);
+}
+
 void TraceLog::AddTraceEvent(
     char phase,
     const unsigned char* category_group_enabled,
@@ -1187,13 +1330,22 @@ void TraceLog::AddTraceEventWithThreadIdAndTimestamp(
     return;
 
   TimeTicks now = timestamp - time_offset_;
-  EventCallback event_callback_copy;
 
   NotificationHelper notifier(this);
 
-  // Check and update the current thread name only if the event is for the
-  // current thread to avoid locks in most cases.
+  ThreadLocalEventBuffer* thread_local_event_buffer = NULL;
+
   if (thread_id == static_cast<int>(PlatformThread::CurrentId())) {
+    if (MessageLoop::current()) {

[dsinclair, 2013/08/14 15:15:59]

Why is this inside the if (thread_id ...) check? In theory, threadA could
receive the thread_id of threadB and, as long as threadA has a message loop, we
should be able to add the event to the threadA local events.

[Xianzhu, 2013/08/14 21:28:34]

Done.

+      thread_local_event_buffer = thread_local_event_buffer_.Get();
+      if (!thread_local_event_buffer) {
+        thread_local_event_buffer = new ThreadLocalEventBuffer(this);
+        thread_local_event_buffer_.Set(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.
     const char* new_name = ThreadIdNameManager::GetInstance()->
         GetName(thread_id);
     // Check if the thread name has been set or changed since the previous
@@ -1203,8 +1355,6 @@ void TraceLog::AddTraceEventWithThreadIdAndTimestamp(
     if (new_name != g_current_thread_name.Get().Get() &&
         new_name && *new_name) {
       g_current_thread_name.Get().Set(new_name);
-
-      AutoLock lock(lock_);
       hash_map<int, std::string>::iterator existing_name =

[dsinclair, 2013/08/14 15:15:59]

Why is this safe to do without holding the lock? Is it because the only thing
accessed/edited is the current thread_id's data?

[Xianzhu, 2013/08/14 21:28:34]

This is a bug. Added lock.

           thread_names_.find(thread_id);
       if (existing_name == thread_names_.end()) {
@@ -1226,40 +1376,51 @@ void TraceLog::AddTraceEventWithThreadIdAndTimestamp(
     }
   }
 
-  TraceEvent trace_event(thread_id,
-      now, phase, category_group_enabled, name, id,
-      num_args, arg_names, arg_types, arg_values,
-      convertable_values, flags);
-
-  do {
-    AutoLock lock(lock_);
+  if (!subtle::NoBarrier_Load(&buffer_is_full_)) {
+    TraceEvent trace_event(thread_id,
+        now, phase, category_group_enabled, name, id,
+        num_args, arg_names, arg_types, arg_values,
+        convertable_values, flags);
 
-    event_callback_copy = event_callback_;
-    if (logged_events_->IsFull())
-      break;
+    if (thread_local_event_buffer) {
+      thread_local_event_buffer->AddEvent(trace_event, &notifier);
+    } else {
+      AutoLock lock(lock_);
+      AddEventToMainBufferWhileLocked(trace_event, &notifier);
+    }
 
-    logged_events_->AddEvent(trace_event);
+    EchoToConsoleContext* echo_to_console_context =
+        reinterpret_cast<EchoToConsoleContext*>(subtle::NoBarrier_Load(
+            &echo_to_console_context_));
+    if (echo_to_console_context) {
+      // ECHO_TO_CONSOLE is enabled.
+      AutoLock lock(lock_);
 
-    if (trace_options_ & ECHO_TO_CONSOLE) {
       TimeDelta duration;
       if (phase == TRACE_EVENT_PHASE_END) {
-        duration = timestamp - thread_event_start_times_[thread_id].top();
-        thread_event_start_times_[thread_id].pop();
+        duration = timestamp -
+            echo_to_console_context->thread_event_start_times[thread_id].top();
+        echo_to_console_context->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;
+      if (echo_to_console_context->thread_colors.find(thread_name) ==
+          echo_to_console_context->thread_colors.end())
+        echo_to_console_context->thread_colors[thread_name] =
+            (echo_to_console_context->thread_colors.size() % 6) + 1;
 
       std::ostringstream log;
-      log << base::StringPrintf("%s: \x1b[0;3%dm",
-                                thread_name.c_str(),
-                                thread_colors_[thread_name]);
+      log << base::StringPrintf(
+          "%s: \x1b[0;3%dm",
+          thread_name.c_str(),
+          echo_to_console_context->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();
+      if (echo_to_console_context->thread_event_start_times.find(thread_id) !=
+          echo_to_console_context->thread_event_start_times.end()) {
+        depth = echo_to_console_context->thread_event_start_times[thread_id]
+            .size();
+      }
 
       for (size_t i = 0; i < depth; ++i)
         log << "| ";
@@ -1270,22 +1431,42 @@ void TraceLog::AddTraceEventWithThreadIdAndTimestamp(
 
       LOG(ERROR) << log.str() << "\x1b[0;m";
 
-      if (phase == TRACE_EVENT_PHASE_BEGIN)
-        thread_event_start_times_[thread_id].push(timestamp);
+      if (phase == TRACE_EVENT_PHASE_BEGIN) {
+        echo_to_console_context->thread_event_start_times[thread_id].push(
+            timestamp);
+      }
     }
+  }
 
-    if (logged_events_->IsFull())
-      notifier.AddNotificationWhileLocked(TRACE_BUFFER_FULL);
-
-    if (watch_category_ == category_group_enabled && watch_event_name_ == name)
+  if (reinterpret_cast<const unsigned char*>(subtle::NoBarrier_Load(
+      &watch_category_)) == category_group_enabled) {
+    AutoLock lock(lock_);
+    if (watch_event_name_ == name)
       notifier.AddNotificationWhileLocked(EVENT_WATCH_NOTIFICATION);
-  } while (0); // release lock
+  }
 
   notifier.SendNotificationIfAny();
-  if (event_callback_copy != NULL) {
-    event_callback_copy(phase, category_group_enabled, name, id,
-        num_args, arg_names, arg_types, arg_values,
-        flags);
+  EventCallback event_callback = reinterpret_cast<EventCallback>(
+      subtle::NoBarrier_Load(&event_callback_));
+  if (event_callback) {
+    event_callback(phase, category_group_enabled, name, id,
+                   num_args, arg_names, arg_types, arg_values,
+                   flags);
+  }
+}
+
+void TraceLog::EnableEchoToConsole(bool enable) {
+  if (enable) {
+    EchoToConsoleContext* new_context = new EchoToConsoleContext();
+    if (subtle::NoBarrier_CompareAndSwap(
+            &echo_to_console_context_, 0,
+            reinterpret_cast<subtle::AtomicWord>(new_context))) {
+      // There is existing context, the new_context is not used.
+      delete new_context;
+    }
+  } else {
+    delete reinterpret_cast<EchoToConsoleContext*>(
+        subtle::NoBarrier_AtomicExchange(&echo_to_console_context_, 0));
   }
 }
 
@@ -1319,7 +1500,8 @@ void TraceLog::SetWatchEvent(const std::string& category_name,
   size_t notify_count = 0;
   {
     AutoLock lock(lock_);
-    watch_category_ = category;
+    subtle::NoBarrier_Store(&watch_category_,
+                            reinterpret_cast<subtle::AtomicWord>(category));
     watch_event_name_ = event_name;
 
     // First, search existing events for watch event because we want to catch
@@ -1339,7 +1521,7 @@ void TraceLog::SetWatchEvent(const std::string& category_name,
 
 void TraceLog::CancelWatchEvent() {
   AutoLock lock(lock_);
-  watch_category_ = NULL;
+  subtle::NoBarrier_Store(&watch_category_, 0);
   watch_event_name_ = "";
 }