SingletonRelaxedCounter

Summary: [Folly] `SingletonRelaxedCounter`, a singleton-per-tag relaxed counter.

Differential Revision: D13149336

fbshipit-source-id: 7cf0144758e9595e188465137a336d712c5d9a76

folly/experimental/SingletonRelaxedCounter.h

+/*
+ * Copyright 2019-present Facebook, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <array>
+#include <atomic>
+#include <type_traits>
+#include <unordered_set>
+
+#include <folly/Likely.h>
+#include <folly/Portability.h>
+#include <folly/Synchronized.h>
+#include <folly/Utility.h>
+#include <folly/detail/StaticSingletonManager.h>
+
+namespace folly {
+
+//  SingletonRelaxedCounter
+//
+//  A singleton-per-tag relaxed counter. Optimized for increment/decrement
+//  runtime performance under contention and inlined fast path code size.
+//
+//  The cost of computing the value of the counter is linear in the number of
+//  threads which perform increments/decrements, and computing the value of the
+//  counter is exclusive with thread exit and dlclose. The result of this
+//  computation is not a point-in-time snapshot of increments and decrements
+//  summed, but is an approximation which may exclude any subset of increments
+//  and decrements that do not happen before the start of the computation.
+//
+//  Templated over the integral types. When templated over an unsigned integral
+//  type, it is assumed that decrements do not exceed increments, and if within
+//  computation of the value of the counter more decrements are observed to
+//  exceed increments then the excess decrements are ignored. This avoids the
+//  scenario of incrementing and decrementing once each in different threads,
+//  and concurrently observing a computed value of the counter of 2^64 - 1.
+//
+//  Templated over the tag types. Each unique pair of integral type and tag type
+//  is a different counter.
+//
+//  Implementation:
+//  Uses a thread-local counter when possible to avoid contention, and a global
+//  counter as a fallback. The total count at any given time is computed by
+//  summing over the global counter plus all of the thread-local counters; since
+//  the total sum is not a snapshot of the value at any given point in time, it
+//  is a relaxed sum; when the system quiesces (i.e., when no concurrent
+//  increments or decrements are happening and no threads are going through
+//  thread exit phase), the sum is exact.
+template <typename Int, typename Tag>
+class SingletonRelaxedCounter {
+ public:
+  static void add(Int value) {
+    mutate(+to_signed(value));
+  }
+  static void sub(Int value) {
+    mutate(-to_signed(value));
+  }
+
+  static Int count() {
+    auto const& global = Global::instance();
+    auto count = global.fallback.load(std::memory_order_relaxed);
+    auto const tracking = global.tracking.rlock();
+    for (auto const kvp : tracking->locals) {
+      count += kvp.first->load(std::memory_order_relaxed);
+    }
+    return std::is_unsigned<Int>::value
+        ? to_unsigned(std::max(Signed(0), count))
+        : count;
+  }
+
+ private:
+  using Signed = std::make_signed_t<Int>;
+  using Counter = std::atomic<Signed>;
+
+  struct CounterAndCache {
+    Counter counter; // valid during LocalLifetime object lifetime
+    Counter* cache; // points to counter when counter is valid
+  };
+
+  struct CounterRefAndLocal {
+    Counter* counter; // refers either to local counter or to global counter
+    bool local; // if true, definitely local; if false, could be global
+  };
+
+  struct LocalLifetime;
+
+  struct Global {
+    struct Tracking {
+      using CounterSet = std::unordered_set<Counter*>;
+      std::unordered_map<Counter*, size_t> locals; // for summing
+      std::unordered_map<LocalLifetime*, CounterSet> lifetimes;
+    };
+
+    Counter fallback; // used instead of local during thread destruction
+    folly::Synchronized<Tracking> tracking;
+
+    static Global& instance() {
+      return folly::detail::createGlobal<Global, Tag>();
+    }
+  };
+
+  // manages local().cache, global().tracking, and moving outstanding counts
+  // from local().counter to global().counter during thread destruction
+  //
+  // the counter-set is within Global to reduce per-thread overhead for threads
+  // which do not participate in counter mutations, rather than being a member
+  // field of LocalLifetime; this comes at the cost of the slow path always
+  // acquiring a unique lock on the global mutex
+  struct LocalLifetime {
+    ~LocalLifetime() {
+      auto& global = Global::instance();
+      auto const tracking = global.tracking.wlock();
+      auto& lifetimes = tracking->lifetimes[this];
+      for (auto ctr : lifetimes) {
+        auto const it = tracking->locals.find(ctr);
+        if (!--it->second) {
+          tracking->locals.erase(it);
+          auto const current = ctr->load(std::memory_order_relaxed);
+          global.fallback.fetch_add(current, std::memory_order_relaxed);
+        }
+      }
+    }
+
+    void track(CounterAndCache& state) {
+      auto& global = Global::instance();
+      state.cache = &state.counter;
+      auto const tracking = global.tracking.wlock();
+      auto const inserted = tracking->lifetimes[this].insert(&state.counter);
+      tracking->locals[&state.counter] += inserted.second;
+    }
+  };
+
+  FOLLY_ALWAYS_INLINE static void mutate(Signed v) {
+    auto cl = counter();
+    auto& c = *cl.counter;
+    if (cl.local) {
+      // splitting load/store on the local counter is faster than fetch-and-add
+      c.store(c.load(std::memory_order_relaxed) + v, std::memory_order_relaxed);
+    } else {
+      // but is not allowed on the global counter because mutations may be lost
+      c.fetch_add(v, std::memory_order_relaxed);
+    }
+  }
+
+  FOLLY_EXPORT FOLLY_ALWAYS_INLINE static CounterAndCache& local() {
+    // this is a member function local instead of a class member because of
+    // https://gcc.gnu.org/bugzilla/show_bug.cgi?id=66944
+    static thread_local CounterAndCache instance;
+    return instance;
+  }
+
+  FOLLY_EXPORT FOLLY_ALWAYS_INLINE static LocalLifetime& lifetime() {
+    static thread_local LocalLifetime lifetime;
+    return lifetime;
+  }
+
+  FOLLY_NOINLINE static Counter* counterSlow(CounterAndCache& state) {
+    lifetime().track(state); // idempotent
+    auto const cache = state.cache;
+    return FOLLY_LIKELY(!!cache) ? cache : &Global::instance().fallback;
+  }
+
+  FOLLY_ALWAYS_INLINE static CounterRefAndLocal counter() {
+    auto& state = local();
+    auto const cache = state.cache; // a copy! null before/after LocalLifetime
+    auto const counter = FOLLY_LIKELY(!!cache) ? cache : counterSlow(state);
+    // cache is a stale nullptr after the first call to counterSlow(); this is
+    // intentional for the side-effect of shrinking the inline fast path
+    return CounterRefAndLocal{counter, !!cache};
+  }
+};
+
+} // namespace folly

folly/experimental/test/SingletonRelaxedCounterBench.cpp

+/*
+ * Copyright 2019-present Facebook, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <folly/experimental/SingletonRelaxedCounter.h>
+
+#include <cstddef>
+#include <thread>
+
+#include <boost/thread/barrier.hpp>
+
+#include <folly/Benchmark.h>
+#include <folly/init/Init.h>
+#include <folly/portability/GFlags.h>
+
+DEFINE_int32(num_threads, 4, "No. of threads allocting the objects");
+
+namespace folly {
+
+namespace {
+struct PrivateTag {};
+} // namespace
+
+using Counter = SingletonRelaxedCounter<size_t, PrivateTag>;
+
+// small wrappers around the functions being benchmarked
+// useful for looking at the the inlined native code of the fast path
+extern "C" void check() noexcept {
+  Counter::add(1);
+}
+
+// a noop which the compiler cannot see through
+// used to prevent some compiler optimizations which could skew benchmarks
+[[gnu::weak]] void noop() noexcept {}
+
+BENCHMARK(MultiThreadPerformance, iters) {
+  BenchmarkSuspender braces;
+
+  boost::barrier barrier(1 + FLAGS_num_threads);
+
+  std::vector<std::thread> threads(FLAGS_num_threads);
+
+  for (auto& thread : threads) {
+    thread = std::thread([&] {
+      auto once = [] {
+        noop();
+        Counter::add(1);
+        noop();
+        Counter::sub(1);
+      };
+
+      // in case the first call does expensive setup, such as take a global lock
+      once();
+
+      barrier.wait(); // A - wait for thread start
+
+      barrier.wait(); // B - init the work
+
+      for (auto i = 0u; i < iters; ++i) {
+        once();
+      }
+
+      barrier.wait(); // C - join the work
+    });
+  }
+
+  barrier.wait(); // A - wait for thread start
+
+  // we want to exclude thread start/stop operations from measurement
+  braces.dismissing([&] {
+    barrier.wait(); // B - init the work
+    barrier.wait(); // C - join the work
+  });
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+}
+} // namespace folly
+
+int main(int argc, char** argv) {
+  folly::Init init(&argc, &argv);
+  folly::runBenchmarks();
+  return 0;
+}

folly/experimental/test/SingletonRelaxedCounterTest.cpp

+/*
+ * Copyright 2019-present Facebook, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <folly/experimental/SingletonRelaxedCounter.h>
+
+#include <cstddef>
+#include <thread>
+
+#include <boost/thread/barrier.hpp>
+
+#include <folly/portability/GTest.h>
+
+namespace folly {
+
+namespace {
+struct PrivateTag {};
+} // namespace
+
+using Counter = SingletonRelaxedCounter<size_t, PrivateTag>;
+
+class SingletonRelaxedCounterTest : public testing::Test {};
+
+TEST_F(SingletonRelaxedCounterTest, Basic) {
+  EXPECT_EQ(0, Counter::count());
+  Counter::add(3);
+  EXPECT_EQ(3, Counter::count());
+  Counter::sub(3);
+  EXPECT_EQ(0, Counter::count());
+}
+
+TEST_F(SingletonRelaxedCounterTest, MultithreadCorrectness) {
+  static constexpr size_t const kPerThreadIncrements = 1000000;
+  static constexpr size_t const kNumThreads = 24;
+
+  std::vector<std::thread> threads(kNumThreads);
+
+  boost::barrier barrier{kNumThreads + 1};
+
+  for (auto& thread : threads) {
+    thread = std::thread([&] {
+      barrier.wait(); // A
+      barrier.wait(); // B
+      for (size_t i = 0; i < kPerThreadIncrements; ++i) {
+        Counter::add(1);
+      }
+      barrier.wait(); // C
+      barrier.wait(); // D
+      for (size_t i = 0; i < kPerThreadIncrements; ++i) {
+        Counter::sub(1);
+      }
+      barrier.wait(); // E
+      barrier.wait(); // F
+    });
+  }
+
+  barrier.wait(); // A
+  EXPECT_EQ(0, Counter::count());
+
+  barrier.wait(); // B
+  barrier.wait(); // C
+  EXPECT_EQ(kPerThreadIncrements * kNumThreads, Counter::count());
+
+  barrier.wait(); // D
+  barrier.wait(); // E
+  EXPECT_EQ(0, Counter::count());
+
+  barrier.wait(); // F
+
+  for (auto& thread : threads) {
+    thread.join();
+  }
+}
+} // namespace folly