DevTools: Parallelize trace messages serialization.

base/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 "base/debug/trace_event_impl.h"
 
 #include <algorithm>
 
 #include "base/base_switches.h"
 #include "base/bind.h"
@@ skipping 14 matching lines @@
 #include "base/strings/string_tokenizer.h"
 #include "base/strings/string_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/strings/utf_string_conversions.h"
 #include "base/synchronization/cancellation_flag.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/sys_info.h"
 #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/worker_pool.h"
 #include "base/time/time.h"
 
 #if defined(OS_WIN)
 #include "base/debug/trace_event_win.h"
 #endif
 
 class DeleteTraceLogForTesting {
  public:
   static void Delete() {
     Singleton<base::debug::TraceLog,
@@ skipping 21 matching lines @@
 const char kEnableSampling[] = "enable-sampling";
 const char kEnableSystrace[] = "enable-systrace";
 
 // Controls the number of trace events we will buffer in-memory
 // before throwing them away.
 const size_t kTraceBufferChunkSize = TraceBufferChunk::kTraceBufferChunkSize;
 const size_t kTraceEventVectorBigBufferChunks =
     512000000 / kTraceBufferChunkSize;
 const size_t kTraceEventVectorBufferChunks = 256000 / kTraceBufferChunkSize;
 const size_t kTraceEventRingBufferChunks = kTraceEventVectorBufferChunks / 4;
-const size_t kTraceEventBatchChunks = 1000 / kTraceBufferChunkSize;
+const size_t kTraceEventBufferSize = 10 * 1024 * 1024;
 // 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 int kThreadFlushTimeoutMs = 3000;
 
 #if !defined(OS_NACL)
 // These categories will cause deadlock when ECHO_TO_CONSOLE. crbug.com/325575.
 const char kEchoToConsoleCategoryFilter[] = "-ipc,-task";
@@ skipping 1645 matching lines @@
         FROM_HERE,
         Bind(&TraceLog::OnFlushTimeout, Unretained(this), generation),
         TimeDelta::FromMilliseconds(kThreadFlushTimeoutMs));
     return;
   }
 
   FinishFlush(generation);
 }
 
 void TraceLog::ConvertTraceEventsToTraceFormat(
-    scoped_ptr<TraceBuffer> logged_events,
     const TraceLog::OutputCallback& flush_output_callback) {
 
-  if (flush_output_callback.is_null())
-    return;
-
+  scoped_ptr<TraceBuffer> events_for_serialization;
+  {
+    AutoLock lock(lock_);
+    events_for_serialization.swap(previous_logged_events_);
+  }
   // 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();
+    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;
+    while (json_events_str_ptr->size() < kTraceEventBufferSize) {
+      const TraceBufferChunk* chunk = events_for_serialization->NextChunk();
+      has_more_events = chunk;
+      if (!chunk)
         break;
-      }
       for (size_t j = 0; j < chunk->size(); ++j) {
-        if (i > 0 || j > 0)
+        if (json_events_str_ptr->size())
           json_events_str_ptr->data().append(",");
         chunk->GetEventAt(j)->AppendAsJSON(&(json_events_str_ptr->data()));
       }
     }
-
     flush_output_callback.Run(json_events_str_ptr, has_more_events);
   } while (has_more_events);
 }
 
 void TraceLog::FinishFlush(int generation) {
-  scoped_ptr<TraceBuffer> previous_logged_events;
   OutputCallback flush_output_callback;
+  bool async_task_started = false;
 
   if (!CheckGeneration(generation))
     return;
 
   {
     AutoLock lock(lock_);
 
-    previous_logged_events.swap(logged_events_);
+    previous_logged_events_.swap(logged_events_);
     UseNextTraceBuffer();
     thread_message_loops_.clear();
 
     flush_message_loop_proxy_ = NULL;
     flush_output_callback = flush_output_callback_;
     flush_output_callback_.Reset();
+
+    async_task_started = WorkerPool::PostTask(

[Sami, 2014/12/05 16:33:01]

Is there any reason to expect this to fail in practice? That is, should we just
DCHECK that it worked?

[loislo, 2014/12/05 21:38:45]

As far as I see it has different behavior on the POSIX and non POSIX platforms.
See
https://code.google.com/p/chromium/codesearch#chromium/src/net/dns/serial_wor...

[Sami, 2014/12/08 14:35:36]

Interesting. Looks like QueueUserWorkItem can fail on Windows for unspecified
reasons. I guess we need to be prepared for that then.

+        FROM_HERE,
+        Bind(&TraceLog::ConvertTraceEventsToTraceFormat,
+             Unretained(this),
+             flush_output_callback),
+        true);
   }
-
-  ConvertTraceEventsToTraceFormat(previous_logged_events.Pass(),
-                                  flush_output_callback);
+  if (!async_task_started)
+    ConvertTraceEventsToTraceFormat(flush_output_callback);
 }
 
 // Run in each thread holding a local event buffer.
 void TraceLog::FlushCurrentThread(int generation) {
   {
     AutoLock lock(lock_);
     if (!CheckGeneration(generation) || !flush_message_loop_proxy_.get()) {
       // This is late. The corresponding flush has finished.
       return;
     }
@@ skipping 29 matching lines @@
          thread_message_loops_.begin();
          it != thread_message_loops_.end(); ++it) {
       LOG(WARNING) << "Thread: " << (*it)->thread_name();
     }
   }
   FinishFlush(generation);
 }
 
 void TraceLog::FlushButLeaveBufferIntact(
     const TraceLog::OutputCallback& flush_output_callback) {
-  scoped_ptr<TraceBuffer> previous_logged_events;
   {
     AutoLock lock(lock_);
     AddMetadataEventsWhileLocked();
     if (thread_shared_chunk_) {
       // Return the chunk to the main buffer to flush the sampling data.
       logged_events_->ReturnChunk(thread_shared_chunk_index_,
                                   thread_shared_chunk_.Pass());
     }
-    previous_logged_events = logged_events_->CloneForIteration().Pass();
+    previous_logged_events_ = logged_events_->CloneForIteration().Pass();
   }  // release lock
-
-  ConvertTraceEventsToTraceFormat(previous_logged_events.Pass(),
-                                  flush_output_callback);
+  ConvertTraceEventsToTraceFormat(flush_output_callback);
 }
 
 void TraceLog::UseNextTraceBuffer() {
   logged_events_.reset(CreateTraceBuffer());
   subtle::NoBarrier_AtomicIncrement(&generation_, 1);
   thread_shared_chunk_.reset();
   thread_shared_chunk_index_ = 0;
 }
 
 TraceEventHandle TraceLog::AddTraceEvent(
@@ skipping 706 matching lines @@
 }
 
 ScopedTraceBinaryEfficient::~ScopedTraceBinaryEfficient() {
   if (*category_group_enabled_) {
     TRACE_EVENT_API_UPDATE_TRACE_EVENT_DURATION(category_group_enabled_,
                                                 name_, event_handle_);
   }
 }
 
 }  // namespace trace_event_internal