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,55 @@
 #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(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);
+  }
+
   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 {
+    std::atomic<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,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/tests/benchmark/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/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,44 @@ TEST_F(CounterTest, inc_multiple) {
   counter.Increment(5);
   EXPECT_EQ(counter.Value(), 7.0);
 }
+
+TEST_F(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_F(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)));
+}