Implement scalable counter

Implementation is based on chapter 5.2.2 of
Paul E. McKenney (2017), "Is Parallel Programming
Hard, And, If So, What Can You Do About It?"

Author: jerryct

core/benchmarks/counter_bench.cc

@@ -10,10 +10,26 @@ static void BM_Counter_Increment(benchmark::State& state) {
       BuildCounter().Name("benchmark_counter").Help("").Register(registry);
   auto& counter = counter_family.Add({});
 
-  while (state.KeepRunning()) counter.Increment();
+  while (state.KeepRunning()) {
+    counter.Increment();
+  }
 }
 BENCHMARK(BM_Counter_Increment);
 
+class BM_Counter : public benchmark::Fixture {
+ protected:
+  BM_Counter() { this->ThreadPerCpu(); }
+
+  prometheus::Counter counter{};
+};
+
+BENCHMARK_F(BM_Counter, ConcurrentIncrement)
+(benchmark::State& state) {
+  for (auto _ : state) {
+    counter.Increment();
+  }
+}
+
 static void BM_Counter_Collect(benchmark::State& state) {
   using prometheus::BuildCounter;
   using prometheus::Counter;

core/benchmarks/gauge_bench.cc

@@ -14,6 +14,20 @@ static void BM_Gauge_Increment(benchmark::State& state) {
 }
 BENCHMARK(BM_Gauge_Increment);
 
+class BM_Gauge : public benchmark::Fixture {
+ protected:
+  BM_Gauge() { this->ThreadPerCpu(); }
+
+  prometheus::Gauge gauge{};
+};
+
+BENCHMARK_F(BM_Gauge, ConcurrentIncrement)
+(benchmark::State& state) {
+  for (auto _ : state) {
+    gauge.Increment();
+  }
+}
+
 static void BM_Gauge_Decrement(benchmark::State& state) {
   using prometheus::BuildGauge;
   using prometheus::Gauge;

core/include/prometheus/counter.h

@@ -1,7 +1,10 @@
 #pragma once
 
+#include <array>
+#include <atomic>
+#include <exception>
+
 #include "prometheus/client_metric.h"
-#include "prometheus/gauge.h"
 #include "prometheus/metric_type.h"
 
 namespace prometheus {
@@ -17,7 +20,17 @@ namespace prometheus {
 /// - errors
 ///
 /// Do not use a counter to expose a value that can decrease - instead use a
-/// Gauge.
+/// Gauge. If an montonically increasing counter is applicable a counter shall
+/// be prefered to a Gauge because of a better update performance.
+///
+/// The implementation exhibits a performance which is near a sequential
+/// implementation and scales linearly with increasing number of updater threads
+/// in a multi-threaded environment invoking Increment(). However, this
+/// excellent update-side scalability comes at read-side expense invoking
+/// Collect(). Increment() can therefor be used in the fast-path of the code,
+/// where the count is updated extremely frequently. The Collect() function on
+/// the other hand shall read the counter at a low sample rate, e.g.
+/// milliseconds.
 ///
 /// The class is thread-safe. No concurrent call to any API of this type causes
 /// a data race.
@@ -29,20 +42,55 @@ class Counter {
   Counter() = default;
 
   /// \brief Increment the counter by 1.
-  void Increment();
+  void Increment() { IncrementUnchecked(1.0); }
 
   /// \brief Increment the counter by a given amount.
   ///
   /// The counter will not change if the given amount is negative.
-  void Increment(double);
+  void Increment(const double value) {
+    if (value < 0.0) {
+      return;
+    }
+    IncrementUnchecked(value);
+  }
 
   /// \brief Get the current value of the counter.
   double Value() const;
 
   ClientMetric Collect() const;
 
  private:
-  Gauge gauge_{0.0};
+  int ThreadId() {
+    thread_local int id{-1};
+
+    if (id == -1) {
+      id = AssignThreadId();
+    }
+    return id;
+  }
+
+  int AssignThreadId() {
+    const int id{count_.fetch_add(1)};
+
+    if (id >= per_thread_counter_.size()) {
+      std::terminate();
+    }
+
+    return id;
+  }
+
+  void IncrementUnchecked(const double v) {
+    CacheLine& c = per_thread_counter_[ThreadId()];
+    const double new_value{c.v.load(std::memory_order_relaxed) + v};
+    c.v.store(new_value, std::memory_order_relaxed);
+  }
+
+  struct alignas(128) CacheLine {
+    std::atomic<double> v{0.0};
+  };
+
+  std::atomic<int> count_{0};
+  std::array<CacheLine, 256> per_thread_counter_{};
 };
 
 }  // namespace prometheus

core/include/prometheus/gauge.h

@@ -17,7 +17,11 @@ namespace prometheus {
 /// memory usage, but also "counts" that can go up and down, like the number of
 /// running processes.
 ///
-/// The class is thread-safe. No concurrent call to any API of this type causes a data race.
+/// If an montonically increasing counter is applicable a Counter shall be
+/// prefered to a Gauge because of a better update performance.
+///
+/// The class is thread-safe. No concurrent call to any API of this type causes
+/// a data race.
 class Gauge {
  public:
   static const MetricType metric_type{MetricType::Gauge};

core/src/counter.cc

@@ -1,12 +1,14 @@
 #include "prometheus/counter.h"
 
-namespace prometheus {
-
-void Counter::Increment() { gauge_.Increment(); }
+#include <numeric>
 
-void Counter::Increment(const double val) { gauge_.Increment(val); }
+namespace prometheus {
 
-double Counter::Value() const { return gauge_.Value(); }
+double Counter::Value() const {
+  return std::accumulate(
+      std::begin(per_thread_counter_), std::end(per_thread_counter_), 0.0,
+      [](const double a, const CacheLine& b) { return a + b.v; });
+}
 
 ClientMetric Counter::Collect() const {
   ClientMetric metric;

core/tests/counter_test.cc

@@ -2,6 +2,11 @@
 
 #include <gmock/gmock.h>
 
+#include <thread>
+#include <vector>
+
+#include <prometheus/counter.h>
+
 namespace prometheus {
 namespace {
 
@@ -37,5 +42,46 @@ TEST(CounterTest, inc_negative_value) {
   EXPECT_EQ(counter.Value(), 5.0);
 }
 
+TEST(CounterTest, concurrent_writes) {
+  Counter counter;
+  std::vector<std::thread> threads(std::thread::hardware_concurrency());
+
+  for (auto& thread : threads) {
+    thread = std::thread{[&counter]() {
+      for (int i{0}; i < 100000; ++i) {
+        counter.Increment();
+      }
+    }};
+  }
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+
+  EXPECT_EQ(100000 * threads.size(), counter.Value());
+}
+
+TEST(CounterTest, concurrent_read_write) {
+  Counter counter;
+  std::vector<double> values;
+  values.reserve(100000);
+
+  std::thread reader{[&counter, &values]() {
+    for (int i{0}; i < 100000; ++i) {
+      values.push_back(counter.Value());
+    }
+  }};
+  std::thread writer{[&counter]() {
+    for (int i{0}; i < 100000; ++i) {
+      counter.Increment();
+    }
+  }};
+
+  reader.join();
+  writer.join();
+
+  EXPECT_TRUE(std::is_sorted(std::begin(values), std::end(values)));
+}
+
 }  // namespace
 }  // namespace prometheus