WIP: The tracing service prototype

services/resource_coordinator/tracing/coordinator_impl.cc

Index: services/resource_coordinator/tracing/coordinator_impl.cc
diff --git a/services/resource_coordinator/tracing/coordinator_impl.cc b/services/resource_coordinator/tracing/coordinator_impl.cc
new file mode 100644
index 0000000000000000000000000000000000000000..caf526cbb9f76826b2553f99d7efa073208c0be1
--- /dev/null
+++ b/services/resource_coordinator/tracing/coordinator_impl.cc
@@ -0,0 +1,249 @@
+// Copyright 2017 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 "services/resource_coordinator/tracing/coordinator_impl.h"
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/callback_forward.h"
+#include "base/memory/ptr_util.h"
+#include "base/task_scheduler/post_task.h"
+#include "base/task_scheduler/task_traits.h"
+#include "base/threading/thread_task_runner_handle.h"
+#include "base/trace_event/trace_config.h"
+#include "mojo/common/data_pipe_utils.h"
+#include "services/resource_coordinator/public/cpp/tracing/chrome_agent.h"
+#include "services/resource_coordinator/tracing/agent_set_impl.h"
+#include "services/resource_coordinator/tracing/recorder_impl.h"
+#include "services/service_manager/public/cpp/bind_source_info.h"
+
+namespace {
+
+resource_coordinator::tracing::CoordinatorImpl* g_coordinator_impl;
+
+}  // namespace
+
+namespace resource_coordinator {
+namespace tracing {
+
+const char CoordinatorImpl::kRequestBufferUsageType[] = "RequestBufferUsage";
+
+// static
+CoordinatorImpl* CoordinatorImpl::GetInstance() {
+  return g_coordinator_impl;
+}
+
+CoordinatorImpl::CoordinatorImpl()
+    : binding_(this), task_runner_(base::ThreadTaskRunnerHandle::Get()) {
+  DCHECK(!g_coordinator_impl);
+  g_coordinator_impl = this;
+  background_task_runner_ = base::CreateSequencedTaskRunnerWithTraits(
+      base::TaskTraits().MayBlock().WithBaseSyncPrimitives().WithPriority(
+          base::TaskPriority::BACKGROUND));
+  agent_set_ = base::MakeUnique<AgentSetImpl>();
+  mojom::AgentSetPtr agent_set_interface_ptr;
+  agent_set_->BindAgentSetRequest(service_manager::BindSourceInfo(),
+                                  mojo::MakeRequest(&agent_set_interface_ptr));
+  resource_coordinator::tracing::ChromeAgent::InitializeIfNeeded(
+      std::move(agent_set_interface_ptr));
+}
+
+CoordinatorImpl::~CoordinatorImpl() {
+  g_coordinator_impl = nullptr;
+}
+
+void CoordinatorImpl::BindCoordinatorRequest(
+    mojom::CoordinatorRequest request) {
+  binding_.Bind(std::move(request));
+}
+
+void CoordinatorImpl::StartTracing(mojo::ScopedDataPipeProducerHandle stream,
+                                   const std::string& config) {
+  DCHECK(!stream_.is_valid());
+  config_ = config;
+  stream_ = std::move(stream);
+  first_recorder_update_ = true;
+  recording_label_.clear();
+  agent_set_->SetAgentProcessingCallback(base::Bind(
+      &CoordinatorImpl::SendStartTracingToAgent, base::Unretained(this)));
+}
+
+void CoordinatorImpl::StopAndFlush() {
+  DCHECK(IsTracing());
+  agent_set_->ResetAgentProcessingCallback();
+  agent_set_->ForAllAgents(
+      [](AgentSetImpl::Entry* entry) { entry->agent()->StopAndFlush(); });
+  if (recorders_.size() == 0)
+    OnFlushDone();
+}
+
+void CoordinatorImpl::IsTracing(const IsTracingCallback& callback) {
+  callback.Run(IsTracing());
+}
+
+bool CoordinatorImpl::IsTracing() {
+  return stream_.is_valid() || !get_categories_callback_.is_null();
+}
+
+void CoordinatorImpl::RequestBufferUsage(
+    const RequestBufferUsageCallback& callback) {
+  if (!request_buffer_usage_callback_.is_null()) {
+    callback.Run(false, 0, 0);
+    return;
+  }
+
+  maximum_trace_buffer_usage_ = 0;
+  approximate_event_count_ = 0;
+  request_buffer_usage_callback_ = callback;
+  pending_request_buffer_status_count_ = 0;
+  agent_set_->ForAllAgents([this](AgentSetImpl::Entry* entry) {
+    entry->AddDisconnectClosure(
+        kRequestBufferUsageType,
+        base::BindOnce(&CoordinatorImpl::OnRequestBufferStatusResponse,
+                       base::Unretained(this), entry, 0, 0));
+    entry->agent()->RequestBufferStatus(
+        base::BindRepeating(&CoordinatorImpl::OnRequestBufferStatusResponse,
+                            base::Unretained(this), entry));
+    pending_request_buffer_status_count_++;
+  });
+}
+
+void CoordinatorImpl::OnRequestBufferStatusResponse(AgentSetImpl::Entry* entry,
+                                                    uint32_t capacity,
+                                                    uint32_t count) {
+  if (!entry->RemoveDisconnectClosure(kRequestBufferUsageType))
+    return;
+
+  DCHECK(pending_request_buffer_status_count_);
+  if (capacity > 0) {
+    float percent_full =
+        static_cast<float>(static_cast<double>(count) / capacity);
+    maximum_trace_buffer_usage_ =
+        std::max(maximum_trace_buffer_usage_, percent_full);
+    approximate_event_count_ += count;
+  }
+
+  if (--pending_request_buffer_status_count_ == 0) {
+    request_buffer_usage_callback_.Run(true, maximum_trace_buffer_usage_,
+                                       approximate_event_count_);
+    request_buffer_usage_callback_.Reset();
+  }
+}
+
+void CoordinatorImpl::SendStartTracingToAgent(AgentSetImpl::Entry* entry) {
+  bool is_array = entry->type() == mojom::TraceDataType::ARRAY;
+  mojom::RecorderPtr ptr;
+  recorders_[entry->label()].insert(base::MakeUnique<RecorderImpl>(
+      MakeRequest(&ptr), is_array,
+      base::BindRepeating(&CoordinatorImpl::OnRecorderUpdated,
+                          base::Unretained(this), entry->label()),
+      background_task_runner_));
+  DCHECK(is_array || recorders_[entry->label()].size() == 1);
+  entry->agent()->StartTracing(config_, std::move(ptr),
+                               !get_categories_callback_.is_null(),
+                               base::Closure());
+}
+
+void CoordinatorImpl::OnFlushDone() {
+  if (!get_categories_callback_.is_null()) {
+    std::set<std::string> category_set;
+    for (const auto& key_value : recorders_) {
+      for (const auto& recorder : key_value.second) {
+        const auto& recorder_categories = recorder->category_set();
+        category_set.insert(recorder_categories.begin(),
+                            recorder_categories.end());
+      }
+    }
+
+    std::string categories_str;
+    bool first = true;
+    for (const auto& category : category_set) {
+      if (!first)
+        categories_str += ",";
+      first = false;
+      categories_str += category;
+    }
+    get_categories_callback_.Run(categories_str);
+    get_categories_callback_.Reset();
+  }
+
+  recorders_.clear();
+  stream_.reset();
+}
+
+void CoordinatorImpl::OnRecorderUpdated(const std::string& label) {
+  DCHECK(background_task_runner_->RunsTasksOnCurrentThread());
+  if (!recording_label_.empty() && recording_label_ != label)
+    return;
+  while (true) {
+    if (!recording_label_.empty() && ReadFromRecorders())
+      return;
+    recording_label_.clear();
+    bool all_finished = true;
+    for (const auto& key_value : recorders_) {
+      for (const auto& recorder : key_value.second) {
+        all_finished &= !recorder->is_recording();
+        if (recorder->data().size() > 0) {
+          recording_label_ = key_value.first;
+          write_label_ = true;
+          break;
+        }
+      }
+      if (recording_label_.size() != 0)
+        break;
+    }
+    if (recording_label_.size() == 0) {
+      // No recorder has any data for us, right now.
+      if (all_finished) {
+        if (!first_recorder_update_)
+          mojo::common::BlockingCopyFromString("}", stream_);
+        // Recorder connections should be closed on their binding thread.
+        task_runner_->PostTask(
+            FROM_HERE,
+            base::Bind(&CoordinatorImpl::OnFlushDone, base::Unretained(this)));
+      }
+      return;
+    }
+  }
+}
+
+bool CoordinatorImpl::ReadFromRecorders() {
+  DCHECK(background_task_runner_->RunsTasksOnCurrentThread());
+  bool all_finished = true;
+  bool is_array = (*recorders_[recording_label_].begin())->is_array();
+  for (const auto& recorder : recorders_[recording_label_]) {
+    all_finished &= !recorder->is_recording();
+    if (recorder->data().size() == 0)
+      continue;
+    DCHECK(stream_.is_valid());
+    if (write_label_) {
+      std::string prefix = first_recorder_update_ ? "{\"" : "\"";
+      prefix += recording_label_ + "\":" + (is_array ? "[" : "\"");
+      mojo::common::BlockingCopyFromString(prefix, stream_);
+      write_label_ = false;
+      first_recorder_update_ = false;
+    } else {
+      if (recorder->is_array())
+        mojo::common::BlockingCopyFromString(",", stream_);
+    }
+    mojo::common::BlockingCopyFromString(recorder->data(), stream_);
+    recorder->clear_data();
+  }
+  if (all_finished) {
+    if (!write_label_)
+      mojo::common::BlockingCopyFromString(is_array ? "]" : "\"", stream_);
+    return false;
+  }
+  return true;
+}
+
+void CoordinatorImpl::GetCategories(const GetCategoriesCallback& callback) {
+  get_categories_callback_ = callback;
+  StartTracing(mojo::ScopedDataPipeProducerHandle(),
+               base::trace_event::TraceConfig("*", "").ToString());
+  StopAndFlush();
+}
+
+}  // namespace tracing
+}  // namespace resource_coordinator