Delay creation of exporter threads until a handler is registered, and and do not include the time it takes to export in the export interval. (#20)

Author: isturdy

opencensus/stats/internal/stats_exporter.cc

@@ -16,6 +16,7 @@
 
 #include <thread>  // NOLINT
 
+#include "absl/memory/memory.h"
 #include "absl/synchronization/mutex.h"
 #include "absl/time/clock.h"
 #include "absl/time/time.h"
@@ -41,9 +42,15 @@ class StatsExporterImpl {
     views_.erase(std::string(name));
   }
 
+  // Adds a handler, which cannot be subsequently removed (except by
+  // ClearHandlersForTesting()). The background thread is started when the
+  // first handler is registered.
   void RegisterHandler(std::unique_ptr<StatsExporter::Handler> handler) {
     absl::MutexLock l(&mu_);
     handlers_.push_back(std::move(handler));
+    if (!thread_started_) {
+      StartExportThread();
+    }
   }
 
   void Export() {
@@ -59,7 +66,7 @@ class StatsExporterImpl {
   }
 
  private:
-  StatsExporterImpl() : t_(&StatsExporterImpl::RunWorkerLoop, this) {}
+  StatsExporterImpl() {}
 
   void SendToHandlers(const ViewDescriptor& descriptor, const ViewData& data)
       SHARED_LOCKS_REQUIRED(mu_) {
@@ -68,10 +75,20 @@ class StatsExporterImpl {
     }
   }
 
+  void StartExportThread() EXCLUSIVE_LOCKS_REQUIRED(mu_) {
+    t_ = std::thread(&StatsExporterImpl::RunWorkerLoop, this);
+    thread_started_ = true;
+  }
+
   // Loops forever, calling Export() every export_interval_.
   void RunWorkerLoop() {
+    absl::Time next_export_time = absl::Now() + export_interval_;
     while (true) {
-      absl::SleepFor(export_interval_);
+      // SleepFor() returns immediately when given a negative duration.
+      absl::SleepFor(next_export_time - absl::Now());
+      // In case the last export took longer than the export interval, we
+      // calculate the next time from now.
+      next_export_time = absl::Now() + export_interval_;
       Export();
     }
   }
@@ -83,7 +100,9 @@ class StatsExporterImpl {
   std::vector<std::unique_ptr<StatsExporter::Handler>> handlers_
       GUARDED_BY(mu_);
   std::unordered_map<std::string, std::unique_ptr<View>> views_ GUARDED_BY(mu_);
-  std::thread t_;
+
+  bool thread_started_ GUARDED_BY(mu_) = false;
+  std::thread t_ GUARDED_BY(mu_);
 };
 
 void StatsExporter::AddView(const ViewDescriptor& view) {

opencensus/stats/internal/stats_exporter_test.cc

@@ -58,11 +58,19 @@ class MockExporter : public StatsExporter::Handler {
   std::vector<ViewDescriptor> actual_descriptors_;
 };
 
-constexpr char kMeasureId[] = "first_measure_id";
+constexpr char kMeasureId[] = "test_measure_id";
+
+MeasureDouble TestMeasure() {
+  static MeasureDouble measure =
+      MeasureRegistry::RegisterDouble(kMeasureId, "ops", "");
+  return measure;
+}
 
 class StatsExporterTest : public ::testing::Test {
  protected:
   void SetUp() {
+    // Access the measure to ensure it has been registered.
+    TestMeasure();
     descriptor1_.set_name("id1");
     descriptor1_.set_measure(kMeasureId);
     descriptor1_.set_aggregation(Aggregation::Count());
@@ -87,9 +95,6 @@ class StatsExporterTest : public ::testing::Test {
 
   static void Export() { StatsExporter::ExportForTesting(); }
 
-  MeasureDouble first_measure_ = MeasureRegistry::RegisterDouble(
-      kMeasureId, "ops", "Usage of resource 1.");
-
   ViewDescriptor descriptor1_;
   ViewDescriptor descriptor1_edited_;
   ViewDescriptor descriptor2_;
@@ -128,7 +133,7 @@ TEST_F(StatsExporterTest, MultipleExporters) {
 TEST_F(StatsExporterTest, TimedExport) {
   MockExporter::Register({descriptor1_});
   StatsExporter::AddView(descriptor1_);
-  absl::SleepFor(absl::Seconds(15));
+  absl::SleepFor(absl::Seconds(11));
 }
 
 }  // namespace stats

opencensus/trace/internal/span_exporter_impl.cc

@@ -35,15 +35,15 @@ SpanExporterImpl* SpanExporterImpl::Get() {
 // Create detached worker thread
 SpanExporterImpl::SpanExporterImpl(uint32_t buffer_size,
                                    absl::Duration interval)
-    : buffer_size_(buffer_size),
-      interval_(interval),
-      size_(0),
-      t_(&SpanExporterImpl::RunWorkerLoop, this) {}
+    : buffer_size_(buffer_size), interval_(interval), size_(0) {}
 
 void SpanExporterImpl::RegisterHandler(
     std::unique_ptr<SpanExporter::Handler> handler) {
   absl::MutexLock l(&handler_mu_);
   handlers_.emplace_back(std::move(handler));
+  if (!thread_started_) {
+    StartExportThread();
+  }
 }
 
 void SpanExporterImpl::AddSpan(
@@ -53,23 +53,30 @@ void SpanExporterImpl::AddSpan(
   size_.fetch_add(1, std::memory_order_acq_rel);
 }
 
+void SpanExporterImpl::StartExportThread() {
+  t_ = std::thread(&SpanExporterImpl::RunWorkerLoop, this);
+  thread_started_ = true;
+}
+
 void SpanExporterImpl::RunWorkerLoop() {
   std::vector<opencensus::trace::exporter::SpanData> span_data_;
   std::vector<std::shared_ptr<opencensus::trace::SpanImpl>> spans_copy_;
   // Thread loops forever.
   // TODO: Add in shutdown mechanism.
+  absl::Time next_forced_export_time = absl::Now() + interval_;
   while (true) {
     {
       absl::MutexLock l(&span_mu_);
       // Wait until batch is full or interval time has been exceeded.
-      span_mu_.AwaitWithTimeout(
+      span_mu_.AwaitWithDeadline(
           absl::Condition(
               +[](SpanExporterImpl* ptr) {
                 return (ptr->size_.load(std::memory_order_acquire) >=
                         ptr->buffer_size_);
               },
               this),
-          interval_);
+          next_forced_export_time);
+      next_forced_export_time = absl::Now() + interval_;
       if (spans_.empty()) {
         continue;
       }

opencensus/trace/internal/span_exporter_impl.h

@@ -65,6 +65,7 @@ class SpanExporterImpl {
   SpanExporterImpl& operator=(SpanExporterImpl&&) = delete;
   friend class Span;
 
+  void StartExportThread() EXCLUSIVE_LOCKS_REQUIRED(handler_mu_);
   void RunWorkerLoop();
   // Calls all registered handlers and exports the spans contained in span_data.
   void Export(const std::vector<SpanData>& span_data);
@@ -83,6 +84,7 @@ class SpanExporterImpl {
       GUARDED_BY(span_mu_);
   std::vector<std::unique_ptr<SpanExporter::Handler>> handlers_
       GUARDED_BY(handler_mu_);
+  bool thread_started_ GUARDED_BY(handler_mu_) = false;
   std::thread t_;
 };