Add more performant manual recorder

CMakeLists.txt

@@ -195,6 +195,7 @@ include(LightStepTracerCommon)
 include(LightStepTracerConfiguration)
 
 set(LIGHTSTEP_SRCS src/common/utility.cpp
+                   src/common/buffer_chain.cpp
                    src/common/fragment_input_stream.cpp
                    src/common/fragment_array_input_stream.cpp
                    src/common/protobuf.cpp
@@ -204,13 +205,20 @@ set(LIGHTSTEP_SRCS src/common/utility.cpp
                    src/common/random.cpp
                    src/common/random_traverser.cpp
                    src/common/serialization.cpp
-                   src/common/serialization_chain.cpp
+                   src/common/chunked_http_framing.cpp
+                   src/common/report_request_framing.cpp
+                   src/common/composable_fragment_input_stream.cpp
+                   src/common/chained_stream.cpp
                    src/common/timestamp.cpp
                    src/recorder/report_builder.cpp
                    src/recorder/auto_recorder.cpp
                    src/recorder/fork_aware_recorder.cpp
+                   src/recorder/legacy_manual_recorder.cpp
                    src/recorder/manual_recorder.cpp
                    src/recorder/transporter.cpp
+                   src/recorder/serialization/report_request.cpp
+                   src/recorder/serialization/report_request_header.cpp
+                   src/recorder/serialization/embedded_metrics_message.cpp
                    src/tracer/binary_carrier.cpp
                    src/tracer/json_options.cpp
                    src/tracer/propagation/b3_propagator.cpp
@@ -266,10 +274,9 @@ if (WITH_LIBEVENT)
                              src/recorder/stream_recorder/satellite_connection.cpp
                              src/recorder/stream_recorder/satellite_streamer.cpp
                              src/recorder/stream_recorder/span_stream.cpp
-                             src/recorder/stream_recorder/stream_recorder_metrics.cpp
+                             src/recorder/metrics_tracker.cpp
                              src/recorder/stream_recorder/connection_stream.cpp
                              src/recorder/stream_recorder/host_header.cpp
-                             src/recorder/stream_recorder/embedded_metrics_message.cpp
                              src/recorder/stream_recorder/status_line_parser.cpp
                              src/recorder/stream_recorder/utility.cpp
                              src/network/event.cpp

benchmark/BUILD

@@ -14,3 +14,12 @@ lightstep_google_benchmark(
     ],
 )
 
+lightstep_google_benchmark(
+    name = "manual_tracer_benchmark",
+    srcs = [
+        "manual_tracer_benchmark.cpp",
+    ],
+    deps = [
+        "//:tracer_lib",
+    ],
+)

benchmark/manual_tracer_benchmark.cpp

@@ -0,0 +1,156 @@
+#include <cassert>
+#include <memory>
+
+#include "lightstep/tracer.h"
+
+#include "benchmark/benchmark.h"
+
+int MaxBufferedSpans = 500;
+
+//--------------------------------------------------------------------------------------------------
+// LegacyNullTransport
+//--------------------------------------------------------------------------------------------------
+namespace {
+class LegacyNullTransporter final : public lightstep::LegacyAsyncTransporter {
+ public:
+  // LegacyAsyncTransporter
+  void Send(const google::protobuf::Message& request,
+            google::protobuf::Message& /*response*/,
+            Callback& callback) override {
+    using google::protobuf::uint8;
+    auto size = request.ByteSizeLong();
+    std::unique_ptr<uint8> buffer{new uint8[size]};
+    request.SerializeWithCachedSizesToArray(buffer.get());
+    callback.OnSuccess();
+  }
+};
+}  // namespace
+
+//--------------------------------------------------------------------------------------------------
+// NullTransporter
+//--------------------------------------------------------------------------------------------------
+namespace {
+class NullTransporter final : public lightstep::AsyncTransporter {
+ public:
+  // AsyncTransporter
+  void Send(std::unique_ptr<lightstep::BufferChain>&& message,
+            Callback& callback) noexcept override {
+    auto size = message->num_bytes();
+    std::unique_ptr<char> buffer{new char[size]};
+    message->CopyOut(buffer.get(), size);
+    callback.OnSuccess(*message);
+  }
+};
+}  // namespace
+
+//--------------------------------------------------------------------------------------------------
+// MakeLegacyManualTracer
+//--------------------------------------------------------------------------------------------------
+static std::shared_ptr<lightstep::LightStepTracer> MakeLegacyManualTracer() {
+  lightstep::LightStepTracerOptions tracer_options;
+  tracer_options.transporter.reset(new LegacyNullTransporter{});
+  tracer_options.use_thread = false;
+  tracer_options.component_name = "abc";
+  tracer_options.access_token = "123";
+  tracer_options.max_buffered_spans = static_cast<size_t>(MaxBufferedSpans);
+
+  return lightstep::MakeLightStepTracer(std::move(tracer_options));
+}
+
+//--------------------------------------------------------------------------------------------------
+// MakeManualTracer
+//--------------------------------------------------------------------------------------------------
+static std::shared_ptr<lightstep::LightStepTracer> MakeManualTracer() {
+  lightstep::LightStepTracerOptions tracer_options;
+  tracer_options.transporter.reset(new NullTransporter{});
+  tracer_options.use_thread = false;
+  tracer_options.component_name = "abc";
+  tracer_options.access_token = "123";
+  tracer_options.max_buffered_spans = static_cast<size_t>(MaxBufferedSpans);
+
+  return lightstep::MakeLightStepTracer(std::move(tracer_options));
+}
+
+//--------------------------------------------------------------------------------------------------
+// MakeTracer
+//--------------------------------------------------------------------------------------------------
+static std::shared_ptr<lightstep::LightStepTracer> MakeTracer(
+    opentracing::string_view tracer_type = "legacy_manual") {
+  if (tracer_type == "legacy_manual") {
+    return MakeLegacyManualTracer();
+  }
+  if (tracer_type == "manual") {
+    return MakeManualTracer();
+  }
+  std::cerr << "Unknown tracer type: " << tracer_type << "\n";
+  std::terminate();
+}
+
+//--------------------------------------------------------------------------------------------------
+// BM_SmallSpanReport
+//--------------------------------------------------------------------------------------------------
+static void BM_SmallSpanReport(benchmark::State& state,
+                               const char* tracer_type) {
+  auto tracer = MakeTracer(tracer_type);
+  assert(tracer != nullptr);
+  for (auto _ : state) {
+    for (int i = 0; i < MaxBufferedSpans; ++i) {
+      auto span = tracer->StartSpan("abc123");
+    }
+    tracer->Flush();
+  }
+}
+BENCHMARK_CAPTURE(BM_SmallSpanReport, legacy_manual, "legacy_manual");
+BENCHMARK_CAPTURE(BM_SmallSpanReport, manual, "manual");
+
+//--------------------------------------------------------------------------------------------------
+// BM_TaggedSpanReport
+//--------------------------------------------------------------------------------------------------
+static void BM_TaggedSpanReport(benchmark::State& state,
+                                const char* tracer_type) {
+  auto tracer = MakeTracer(tracer_type);
+  assert(tracer != nullptr);
+  for (auto _ : state) {
+    for (int i = 0; i < MaxBufferedSpans; ++i) {
+      auto span = tracer->StartSpan("abc123");
+      char key[5];
+      key[0] = 'a';
+      key[1] = 'b';
+      key[2] = 'c';
+      key[3] = '0';
+      key[4] = '\0';
+      for (int i = 0; i < 10; ++i) {
+        span->SetTag(opentracing::string_view{key, 4}, "123");
+        ++key[3];
+      }
+    }
+    tracer->Flush();
+  }
+}
+BENCHMARK_CAPTURE(BM_TaggedSpanReport, legacy_manual, "legacy_manual");
+BENCHMARK_CAPTURE(BM_TaggedSpanReport, manual, "manual");
+
+//--------------------------------------------------------------------------------------------------
+// BM_LoggedSpanReport
+//--------------------------------------------------------------------------------------------------
+static void BM_LoggedSpanReport(benchmark::State& state,
+                                const char* tracer_type) {
+  auto tracer = MakeTracer(tracer_type);
+  assert(tracer != nullptr);
+  for (auto _ : state) {
+    for (int i = 0; i < MaxBufferedSpans; ++i) {
+      auto span = tracer->StartSpan("abc123");
+      for (int i = 0; i < 10; ++i) {
+        span->Log({{"abc", 123}});
+      }
+    }
+    tracer->Flush();
+  }
+}
+BENCHMARK_CAPTURE(BM_LoggedSpanReport, legacy_manual, "legacy_manual");
+BENCHMARK_CAPTURE(BM_LoggedSpanReport, manual, "manual");
+
+//--------------------------------------------------------------------------------------------------
+// BENCHMARK_MAIN
+//--------------------------------------------------------------------------------------------------
+BENCHMARK_MAIN();

include/lightstep/BUILD

@@ -19,12 +19,20 @@ lightstep_cc_library(
     ],
 )
 
+lightstep_cc_library(
+    name = "buffer_chain_interface",
+    hdrs = [
+        "buffer_chain.h",
+    ],
+)
+
 lightstep_cc_library(
     name = "transporter_interface",
     hdrs = [
         "transporter.h",
     ],
     external_deps = [
+        ":buffer_chain_interface",
         "@com_google_protobuf//:protobuf",
         "@io_opentracing_cpp//:opentracing",
     ],

include/lightstep/buffer_chain.h

@@ -0,0 +1,42 @@
+#pragma once
+
+#include <cstddef>
+
+namespace lightstep {
+/**
+ * BufferChain provides an interface to access a chained sequence of
+ * contiguous memory buffers.
+ */
+class BufferChain {
+ public:
+  using FragmentCallback = bool (*)(void*, const void*, size_t);
+
+  virtual ~BufferChain() = default;
+
+  /**
+   * @return the number of fragments in chain
+   */
+  virtual size_t num_fragments() const noexcept = 0;
+
+  /**
+   * @return the total number of bytes in all fragments
+   */
+  virtual size_t num_bytes() const noexcept = 0;
+
+  /**
+   * Iterate over each fragment in the buffer chain
+   * @param callback the callback to call for each fragment
+   * @param context a pointer to pass to the callback.
+   */
+  virtual bool ForEachFragment(FragmentCallback callback,
+                               void* context) const = 0;
+
+  /**
+   * Copy out all the fragments into an area of contiguous memory
+   * @param data the location to start copying
+   * @param length the size of the buffer destination. Length must be at least
+   * num_bytes.
+   */
+  void CopyOut(char* data, size_t length) const noexcept;
+};
+}  // namespace lightstep

include/lightstep/tracer.h

@@ -144,7 +144,7 @@ struct LightStepTracerOptions {
   // default transporter is used.
   //
   // If `use_thread` is false, `transporter` should be derived from
-  // AsyncTransporter; otherwise, it must be derived from SyncTransporter.
+  // LegacyAsyncTransporter; otherwise, it must be derived from SyncTransporter.
   std::unique_ptr<Transporter> transporter;
 
   // `metrics_observer` can be optionally provided to track LightStep tracer

include/lightstep/transporter.h

@@ -1,12 +1,13 @@
 #pragma once
 
 #include <google/protobuf/message.h>
+#include <lightstep/buffer_chain.h>
 #include <opentracing/string_view.h>
 #include <opentracing/util.h>
 
 namespace lightstep {
 // Transporter is the abstract base class for SyncTransporter and
-// AsyncTransporter.
+// LegacyAsyncTransporter.
 class Transporter {
  public:
   Transporter() noexcept = default;
@@ -36,8 +37,10 @@ class SyncTransporter : public Transporter {
       google::protobuf::Message& response) = 0;
 };
 
-// AsyncTransporter customizes how asynchronous tracing reports are sent.
-class AsyncTransporter : public Transporter {
+// LegacyAsyncTransporter customizes how asynchronous tracing reports are sent.
+//
+// Deprecated: Use AsyncTransporter
+class LegacyAsyncTransporter : public Transporter {
  public:
   // Callback interface used by Send.
   class Callback {
@@ -66,4 +69,52 @@ class AsyncTransporter : public Transporter {
                     google::protobuf::Message& response,
                     Callback& callback) = 0;
 };
+
+/**
+ * Provides a hook to customize how ReportRequests are transported.
+ */
+class AsyncTransporter : public Transporter {
+ public:
+  /**
+   * A Callback to be invoked after transporting a ReportRequest
+   */
+  class Callback {
+   public:
+    Callback() noexcept = default;
+    Callback(const Callback&) noexcept = default;
+    Callback(Callback&&) noexcept = default;
+
+    virtual ~Callback() = default;
+
+    Callback& operator=(const Callback&) noexcept = default;
+    Callback& operator=(Callback&&) noexcept = default;
+
+    /**
+     * Call when a ReportRequest was successfully transported.
+     * @param message the ReportRequest
+     */
+    virtual void OnSuccess(BufferChain& message) noexcept = 0;
+
+    /**
+     * Call when a ReportRequest not successfully transported.
+     * @param message the ReportRequest
+     */
+    virtual void OnFailure(BufferChain& message) noexcept = 0;
+  };
+
+  /**
+   * Called when the tracer's span buffer is full. Implementors can either flush
+   * the pending spans early (so that we don't drop anything) or do nothing and
+   * let subsequent finished spans drop.
+   */
+  virtual void OnSpanBufferFull() noexcept {}
+
+  /**
+   * Send a ReportRequest
+   * @param message a BufferChain representing the ReportRequest's serialization
+   * @param callback a Callback to be called after message is transported
+   */
+  virtual void Send(std::unique_ptr<BufferChain>&& message,
+                    Callback& callback) noexcept = 0;
+};
 }  // namespace lightstep