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/include/prometheus/counter.h

@@ -1,24 +1,50 @@
 #pragma once
 
+#include <array>
 #include <atomic>
+#include <exception>
 
 #include "prometheus/client_metric.h"
-#include "prometheus/gauge.h"
 
 namespace prometheus {
 
 class Counter {
  public:
   static const MetricType metric_type = MetricType::Counter;
 
-  void Increment();
-  void Increment(double);
+  void Increment() { Increment(1.0); }
+  void Increment(double val) { per_thread_counter_[ThreadId()].v += val; }
+
   double Value() const;
 
-  ClientMetric Collect();
+  ClientMetric Collect() const;
 
  private:
-  Gauge gauge_;
+  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_;
+  }
+
+  struct alignas(128) cacheline {
+    double v{0.0};
+  };
+
+  std::atomic<int> count_{0};
+  std::array<cacheline, 256> per_thread_counter_{};
 };
 
 }  // namespace prometheus

core/src/counter.cc

@@ -1,14 +1,17 @@
-#include "prometheus/counter.h"
 
-namespace prometheus {
+#include "prometheus/counter.h"
 
-void Counter::Increment() { gauge_.Increment(); }
+#include <algorithm>
 
-void Counter::Increment(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() {
+ClientMetric Counter::Collect() const {
   ClientMetric metric;
   metric.counter.value = Value();
   return metric;

core/tests/benchmark/counter_bench.cc

@@ -10,10 +10,32 @@ 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::Registry registry;
+  prometheus::Family<prometheus::Counter>& counter_family =
+      prometheus::BuildCounter()
+          .Name("benchmark_counter")
+          .Help("")
+          .Register(registry);
+  prometheus::Counter& counter = counter_family.Add({});
+};
+
+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/tests/benchmark/gauge_bench.cc

@@ -14,6 +14,26 @@ static void BM_Gauge_Increment(benchmark::State& state) {
 }
 BENCHMARK(BM_Gauge_Increment);
 
+class BM_Gauge : public benchmark::Fixture {
+ protected:
+  BM_Gauge() { this->ThreadPerCpu(); }
+
+  prometheus::Registry registry;
+  prometheus::Family<prometheus::Gauge>& gauge_family{
+      prometheus::BuildGauge()
+          .Name("benchmark_gauge")
+          .Help("")
+          .Register(registry)};
+  prometheus::Gauge& gauge{gauge_family.Add({})};
+};
+
+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/tests/counter_test.cc

@@ -1,5 +1,8 @@
 #include "gmock/gmock.h"
 
+#include <thread>
+#include <vector>
+
 #include <prometheus/counter.h>
 
 using namespace testing;
@@ -31,3 +34,22 @@ TEST_F(CounterTest, inc_multiple) {
   counter.Increment(5);
   EXPECT_EQ(counter.Value(), 7.0);
 }
+
+TEST_F(CounterTest, concurrent) {
+  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());
+}