Re-land issue 750183008 (Parallelize trace messages serialization)

This re-lands https://codereview.chromium.org/750183008 (originally
by loislo@) with the output chunk size reduced down from 10M to 100K
(close to what it used to be originally). The reason for telemetry
failures of original change on Windows was the bug in websocket-client
(https://github.com/liris/websocket-client/issues/163) that caused
it to fragment Python heap when receiving a large frame.

Original issue description:
Move serialization into a worker thread. As a result
IO thread will be able to send messages to the browser.

The original implementation did serialization on IO thread
and was not able to send the messages because ipc had
is_blocked_on_write_ = true and had no chance to check
the actual state of the channel. So the messages were
collected in output_queue. Also the messages could be quite
big and could block the IO thread for a long time.

BUG=463572
TBR=dsinclair
(as he already l-g-t-m-d the original patch)

Review URL: https://codereview.chromium.org/1035783002

Cr-Commit-Position: refs/heads/master@{#322391}

base/trace_event/trace_event_impl.cc

@@ -30,6 +30,7 @@
 #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"
 #include "base/trace_event/trace_event.h"
 #include "base/trace_event/trace_event_synthetic_delay.h"
@@ -73,7 +74,7 @@ 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 kTraceEventBufferSizeInBytes = 100 * 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.
@@ -1208,7 +1209,8 @@ TraceLog::TraceLog()
       event_callback_category_filter_(
           CategoryFilter::kDefaultCategoryFilterString),
       thread_shared_chunk_index_(0),
-      generation_(0) {
+      generation_(0),
+      use_worker_thread_(false) {
   // 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
@@ -1681,7 +1683,9 @@ void TraceLog::SetEventCallbackDisabled() {
 //    - The message loop will be removed from thread_message_loops_;
 //    If this is the last message loop, finish the flush;
 // 4. If any thread hasn't finish its flush in time, finish the flush.
-void TraceLog::Flush(const TraceLog::OutputCallback& cb) {
+void TraceLog::Flush(const TraceLog::OutputCallback& cb,
+                     bool use_worker_thread) {
+  use_worker_thread_ = use_worker_thread;
   if (IsEnabled()) {
     // Can't flush when tracing is enabled because otherwise PostTask would
     // - generate more trace events;
@@ -1735,6 +1739,7 @@ void TraceLog::Flush(const TraceLog::OutputCallback& cb) {
   FinishFlush(generation);
 }
 
+// Usually it runs on a different thread.
 void TraceLog::ConvertTraceEventsToTraceFormat(
     scoped_ptr<TraceBuffer> logged_events,
     const TraceLog::OutputCallback& flush_output_callback) {
@@ -1749,19 +1754,17 @@ void TraceLog::ConvertTraceEventsToTraceFormat(
     scoped_refptr<RefCountedString> json_events_str_ptr =
         new RefCountedString();
 
-    for (size_t i = 0; i < kTraceEventBatchChunks; ++i) {
+    while (json_events_str_ptr->size() < kTraceEventBufferSizeInBytes) {
       const TraceBufferChunk* chunk = logged_events->NextChunk();
-      if (!chunk) {
-        has_more_events = false;
+      has_more_events = chunk != NULL;
+      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(",\n");
         chunk->GetEventAt(j)->AppendAsJSON(&(json_events_str_ptr->data()));
       }
     }
-
     flush_output_callback.Run(json_events_str_ptr, has_more_events);
   } while (has_more_events);
 }
@@ -1785,6 +1788,16 @@ void TraceLog::FinishFlush(int generation) {
     flush_output_callback_.Reset();
   }
 
+  if (use_worker_thread_ &&
+      WorkerPool::PostTask(
+          FROM_HERE,
+          Bind(&TraceLog::ConvertTraceEventsToTraceFormat,
+               Passed(&previous_logged_events),
+               flush_output_callback),
+          true)) {
+    return;
+  }
+
   ConvertTraceEventsToTraceFormat(previous_logged_events.Pass(),
                                   flush_output_callback);
 }

base/trace_event/trace_event_impl.h

@@ -538,10 +538,11 @@ class BASE_EXPORT TraceLog {
   // Due to the implementation of thread-local buffers, flush can't be
   // done when tracing is enabled. If called when tracing is enabled, the
   // callback will be called directly with (empty_string, false) to indicate
-  // the end of this unsuccessful flush.
+  // the end of this unsuccessful flush. Flush does the serialization
+  // on the same thread if the caller doesn't set use_worker_thread explicitly.
   typedef base::Callback<void(const scoped_refptr<base::RefCountedString>&,
                               bool has_more_events)> OutputCallback;
-  void Flush(const OutputCallback& cb);
+  void Flush(const OutputCallback& cb, bool use_worker_thread = false);
   void FlushButLeaveBufferIntact(const OutputCallback& flush_output_callback);
 
   // Called by TRACE_EVENT* macros, don't call this directly.
@@ -710,7 +711,9 @@ class BASE_EXPORT TraceLog {
   // |generation| is used in the following callbacks to check if the callback
   // is called for the flush of the current |logged_events_|.
   void FlushCurrentThread(int generation);
-  void ConvertTraceEventsToTraceFormat(scoped_ptr<TraceBuffer> logged_events,
+  // Usually it runs on a different thread.
+  static void ConvertTraceEventsToTraceFormat(
+      scoped_ptr<TraceBuffer> logged_events,
       const TraceLog::OutputCallback& flush_output_callback);
   void FinishFlush(int generation);
   void OnFlushTimeout(int generation);
@@ -803,6 +806,7 @@ class BASE_EXPORT TraceLog {
   OutputCallback flush_output_callback_;
   scoped_refptr<MessageLoopProxy> flush_message_loop_proxy_;
   subtle::AtomicWord generation_;
+  bool use_worker_thread_;
 
   DISALLOW_COPY_AND_ASSIGN(TraceLog);
 };

content/browser/tracing/tracing_controller_impl.cc

@@ -349,7 +349,8 @@ void TracingControllerImpl::OnDisableRecordingDone() {
     // Flush asynchronously now, because we don't have any children to wait for.
     TraceLog::GetInstance()->Flush(
         base::Bind(&TracingControllerImpl::OnLocalTraceDataCollected,
-                   base::Unretained(this)));
+                   base::Unretained(this)),
+        true);
   }
 
   // Notify all child processes.
@@ -682,7 +683,8 @@ void TracingControllerImpl::OnDisableRecordingAcked(
     // called with the last of the local trace data.
     TraceLog::GetInstance()->Flush(
         base::Bind(&TracingControllerImpl::OnLocalTraceDataCollected,
-                   base::Unretained(this)));
+                   base::Unretained(this)),
+        true);
     return;
   }