This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new bc010e7c Support higher performance histogram (#3561)
bc010e7c is described below
commit bc010e7c89560cdd1d671e35686b4e49be541cb3
Author: Bright Chen <[email protected]>
AuthorDate: Wed Sep 23 10:54:10 2026 +0800
Support higher performance histogram (#3561)
* Support higher performance histogram
* Fix some issues
---
BUILD.bazel | 5 +-
src/bvar/detail/sampler.h | 35 ++++----
src/bvar/histogram.cpp | 16 ++++
src/bvar/histogram.h | 155 +++++++++++++++++++++++++++++++++---
src/bvar/passive_status.h | 7 +-
test/bvar_histogram_unittest.cpp | 167 +++++++++++++++++++++++++++++++++++++++
6 files changed, 353 insertions(+), 32 deletions(-)
diff --git a/BUILD.bazel b/BUILD.bazel
index db73605e..23e433c7 100644
--- a/BUILD.bazel
+++ b/BUILD.bazel
@@ -424,7 +424,10 @@ cc_library(
deps = [
":butil",
] + select({
- "//bazel/config:with_babylon_counter":
["@babylon//:concurrent_counter"],
+ "//bazel/config:with_babylon_counter": [
+ "@babylon//:concurrent_counter",
+ "@babylon//:concurrent_thread_local",
+ ],
"//conditions:default": [],
}),
)
diff --git a/src/bvar/detail/sampler.h b/src/bvar/detail/sampler.h
index 06533e4e..d120458d 100644
--- a/src/bvar/detail/sampler.h
+++ b/src/bvar/detail/sampler.h
@@ -198,7 +198,7 @@ public:
// would not be ignored
take_sample();
}
- ~ReducerSampler() {}
+ ~ReducerSampler() override = default;
void take_sample() override {
// Make _q ready.
@@ -222,21 +222,7 @@ public:
}
Sample<T> latest;
- if (butil::is_same<InvOp, VoidOp>::value) {
- // The operator can't be inversed.
- // We reset the reducer and save the result as a sample.
- // Suming up samples gives the result within a window.
- // In this case, get_value() of _reducer gives wrong answer and
- // should not be called.
- latest.data = _source.reset();
- } else {
- // The operator can be inversed.
- // We save the result as a sample.
- // Inversed operation between latest and oldest sample within a
- // window gives result.
- // get_value() of _reducer can still be called.
- latest.data = _source.get_value();
- }
+ latest.data = take_sample_of(butil::is_same<InvOp, VoidOp>());
latest.time_us = butil::cpuwide_time_us();
_q.elim_push(latest);
}
@@ -313,6 +299,23 @@ public:
}
private:
+ // Tag dispatch instead of a runtime branch on is_same<InvOp, VoidOp>, so
+ // that only the taken branch is instantiated.
+
+ // The operator can't be inversed.
+ // We reset the reducer and save the result as a sample.
+ // Summing up samples gives the result within a window.
+ // In this case, get_value() of `_source` gives wrong answer and
+ // should not be called.
+ T take_sample_of(butil::true_type) { return _source.reset(); }
+
+ // The operator can be inversed.
+ // We save the result as a sample.
+ // Inversed operation between latest and oldest sample within a
+ // window gives result.
+ // get_value() of `_source` can still be called.
+ T take_sample_of(butil::false_type) { return _source.get_value(); }
+
source_type _source;
time_t _window_size;
butil::BoundedQueue<Sample<T> > _q;
diff --git a/src/bvar/histogram.cpp b/src/bvar/histogram.cpp
index 2288bd24..af4fcb69 100644
--- a/src/bvar/histogram.cpp
+++ b/src/bvar/histogram.cpp
@@ -100,10 +100,14 @@ std::ostream& operator<<(std::ostream& os, const
Histogram::Value& v) {
Histogram::Histogram(const BucketSchema& schema)
: _schema(schema)
+#if WITH_BABYLON_COUNTER
+ ,
_storage(std::make_shared<detail::HistogramStorage>(schema.num_buckets()))
+#else
// Both identities carry `num_buckets` so that a value combined out of no
// agent at all still knows how wide it is.
,
_combiner(std::make_shared<combiner_type>(value_type(schema.num_buckets()),
value_type(schema.num_buckets())))
+#endif // WITH_BABYLON_COUNTER
, _sampler(nullptr) {
}
@@ -133,6 +137,9 @@ Histogram& Histogram::operator<<(double value) {
<< " recorded into Histogram(" << name() <<
')';
return *this;
}
+#if WITH_BABYLON_COUNTER
+ _storage->add(_schema.index_of(value), value);
+#else
agent_type* agent = _combiner->get_or_create_tls_agent();
if (BAIDU_UNLIKELY(agent == nullptr)) {
LOG(FATAL) << "Fail to create agent";
@@ -140,9 +147,18 @@ Histogram& Histogram::operator<<(double value) {
}
// `_schema` outlives the call, the op only borrows it to find the bucket.
agent->element.modify(detail::AddSampleToHistogram(&_schema), value);
+#endif // WITH_BABYLON_COUNTER
return *this;
}
+Histogram::value_type Histogram::get_value() const {
+#if WITH_BABYLON_COUNTER
+ return _storage->combine_agents();
+#else
+ return _combiner->combine_agents();
+#endif // WITH_BABYLON_COUNTER
+}
+
Histogram::sampler_type* Histogram::get_sampler() {
if (_sampler == nullptr) {
_sampler = new sampler_type(this);
diff --git a/src/bvar/histogram.h b/src/bvar/histogram.h
index d2623841..468805f4 100644
--- a/src/bvar/histogram.h
+++ b/src/bvar/histogram.h
@@ -19,15 +19,22 @@
#define BVAR_HISTOGRAM_H
#include <stdint.h> // int64_t, uint64_t
-#include <algorithm> // std::lower_bound
+#include <algorithm> // std::max
#include <initializer_list> // std::initializer_list
+#include <memory> // std::shared_ptr
#include <string> // std::string
#include <vector> // std::vector
#include "butil/strings/string_piece.h" // butil::StringPiece
#include "bvar/variable.h" // Variable
-#include "bvar/detail/combiner.h" // AgentCombiner
#include "bvar/detail/sampler.h" // ReducerSampler
#include "bvar/detail/series.h" // HasPlottableSeries
+#if WITH_BABYLON_COUNTER
+#include "babylon/concurrent/thread_local.h" // EnumerableThreadLocal
+#include "butil/atomicops.h" // butil::atomic
+#include "butil/synchronization/seqlock.h" // butil::Seqlock
+#else
+#include "bvar/detail/combiner.h" // AgentCombiner
+#endif // WITH_BABYLON_COUNTER
namespace bvar {
@@ -36,6 +43,13 @@ namespace bvar {
// bounds.
static const size_t MAX_HISTOGRAM_BUCKETS = 32;
+#if WITH_BABYLON_COUNTER
+namespace detail {
+// Defined below, once Histogram::Value is complete.
+class HistogramStorage;
+} // namespace detail
+#endif // WITH_BABYLON_COUNTER
+
// Bucketed distribution of the recorded values.
//
// Each bucket count accumulates since construction and never decreases. The
@@ -79,9 +93,17 @@ public:
BucketSchema(std::initializer_list<double> bounds);
explicit BucketSchema(const std::vector<double>& bounds);
+ // Linear rather than a std::lower_bound: a schema holds at most
+ // MAX_HISTOGRAM_BUCKETS - 1 bounds, which is a couple of cache lines
+ // scanned straight through instead of jumped around in, and the
+ // branch of the scan predicts far better than the one of a binary
+ // search, whose direction is a coin flip at every step.
size_t index_of(double value) const {
- return std::lower_bound(_bounds.begin(), _bounds.end(), value) -
- _bounds.begin();
+ size_t index = 0;
+ while (index < _bounds.size() && _bounds[index] < value) {
+ ++index;
+ }
+ return index;
}
size_t num_buckets() const { return _bounds.size() + 1; }
@@ -148,9 +170,13 @@ public:
typedef Value value_type;
typedef detail::ReducerSampler<Histogram, value_type, Op, InvOp>
sampler_type;
+#if WITH_BABYLON_COUNTER
+ typedef std::shared_ptr<detail::HistogramStorage> shared_combiner_type;
+#else
typedef detail::AgentCombiner<value_type, value_type, Op> combiner_type;
typedef combiner_type::self_shared_type shared_combiner_type;
typedef combiner_type::Agent agent_type;
+#endif // WITH_BABYLON_COUNTER
explicit Histogram(const BucketSchema& schema);
Histogram(const butil::StringPiece& name, const BucketSchema& schema);
@@ -168,7 +194,11 @@ public:
const BucketSchema& schema() const { return _schema; }
bool valid() const {
+#if WITH_BABYLON_COUNTER
+ return _storage != nullptr;
+#else
return _combiner != nullptr && _combiner->valid();
+#endif // WITH_BABYLON_COUNTER
}
void describe(std::ostream& os, bool quote_string) const override;
@@ -189,10 +219,16 @@ public:
// The contract of Window<>/ReducerSampler
Op op() const { return Op(); }
InvOp inv_op() const { return InvOp(); }
- // Expose the shared data carrier, so that ReducerSampler holds it instead
- // of `this`. Sampling then keeps reading valid memory even if this
- // Percentile is destructed before the sampler is recycled.
- shared_combiner_type share_combiner() const { return _combiner; }
+ // Expose the shared data carrier, so that ReducerSampler holds it
+ // instead of `this`. Sampling then keeps reading valid memory even
+ // if this Histogram is destructed before the sampler is recycled.
+ shared_combiner_type share_combiner() const {
+#if WITH_BABYLON_COUNTER
+ return _storage;
+#else
+ return _combiner;
+#endif // WITH_BABYLON_COUNTER
+ }
sampler_type* get_sampler();
private:
@@ -202,10 +238,14 @@ private:
// Snapshot of all the values recorded so far. Walks through every thread
// that ever recorded into this Histogram.
- value_type get_value() const { return _combiner->combine_agents(); }
+ value_type get_value() const;
BucketSchema _schema;
+#if WITH_BABYLON_COUNTER
+ shared_combiner_type _storage;
+#else
shared_combiner_type _combiner;
+#endif // WITH_BABYLON_COUNTER
sampler_type* _sampler;
};
@@ -219,6 +259,101 @@ namespace detail {
template <>
struct HasPlottableSeries<Histogram::Value> : butil::false_type {};
+#if WITH_BABYLON_COUNTER
+
+// One thread's slice of a Histogram.
+//
+// Only the thread owning the slot writes it, so the counters are updated with
+// a relaxed load plus a relaxed store rather than an atomic read-modify-write.
+// They are atomic all the same because the sampling thread reads them while
+// they are being written, which the seqlock allows but does not by itself make
+// race free. The seqlock is what keeps the buckets, the sum and the count of
one
+// slot mutually consistent.
+class HistogramSlot {
+public:
+ HistogramSlot() {
+ for (size_t i = 0; i < MAX_HISTOGRAM_BUCKETS; ++i) {
+ _counts[i].store(0, butil::memory_order_relaxed);
+ }
+ }
+
+ DISALLOW_COPY_AND_ASSIGN(HistogramSlot);
+
+ void add(size_t bucket_index, double value) {
+ _seqlock.store([&] {
+ relaxed_add(&_counts[bucket_index], (uint64_t)1);
+ relaxed_add(&_sum, value);
+ relaxed_add(&_num, (int64_t)1);
+ });
+ }
+
+ Histogram::Value load(size_t num_buckets) const {
+ return _seqlock.load([&] {
+ Histogram::Value v(num_buckets);
+ for (size_t i = 0; i < num_buckets; ++i) {
+ v.counts[i] = _counts[i].load(butil::memory_order_relaxed);
+ }
+ v.sum = _sum.load(butil::memory_order_relaxed);
+ v.num = _num.load(butil::memory_order_relaxed);
+ return v;
+ });
+ }
+
+private:
+ template <typename T, typename U>
+ static void relaxed_add(butil::atomic<T>* target, U delta) {
+ target->store(target->load(butil::memory_order_relaxed) + delta,
+ butil::memory_order_relaxed);
+ }
+
+ butil::Seqlock<> _seqlock;
+ butil::atomic<uint64_t> _counts[MAX_HISTOGRAM_BUCKETS];
+ butil::atomic<double> _sum{0.0};
+ butil::atomic<int64_t> _num{0};
+};
+
+// The per thread slices of one Histogram and their aggregation.
+//
+// babylon's EnumerableThreadLocal hands out one slot per thread and walks
every
+// slot any thread ever took, which is how a thread that has exited keeps
+// contributing what it recorded, the way AgentCombiner commits a dying agent
+// into its global result. babylon recycles the thread id of an exited thread,
+// so a later thread inherits the slot and accumulates on top of it: correct
+// here because a Histogram only ever adds to its buckets and never clears one.
+class HistogramStorage {
+public:
+ explicit HistogramStorage(size_t num_buckets) : _num_buckets(num_buckets)
{}
+
+ DISALLOW_COPY_AND_ASSIGN(HistogramStorage);
+
+ // Records one value into the slot of the calling thread.
+ void add(size_t bucket_index, double value) {
+ _slots.local().add(bucket_index, value);
+ }
+
+ // [Threadsafe] Everything recorded so far by every thread that ever
+ // recorded into this Histogram. Named after
AgentCombiner::combine_agents()
+ // so that detail::CombinerSampleSource fits either backend.
+ Histogram::Value combine_agents() const {
+ Histogram::Value result(_num_buckets);
+ _slots.for_each([&](const HistogramSlot* iter, const HistogramSlot*
end) {
+ for (; iter != end; ++iter) {
+ result += iter->load(_num_buckets);
+ }
+ });
+ return result;
+ }
+
+private:
+ // Leaky: the id allocator behind the thread ids is never destroyed, which
+ // is what a Histogram of static storage duration needs. The thread ids
+ // themselves are still recycled when a thread exits.
+ babylon::EnumerableThreadLocal<HistogramSlot, true> _slots;
+ size_t _num_buckets;
+};
+
+#else
+
// The op of the writing path takes a recorded value rather than another
// Histogram::Value, and needs the schema to find its bucket.
struct AddSampleToHistogram {
@@ -231,6 +366,8 @@ struct AddSampleToHistogram {
const Histogram::BucketSchema* schema;
};
+#endif // WITH_BABYLON_COUNTER
+
} // namespace detail
} // namespace bvar
diff --git a/src/bvar/passive_status.h b/src/bvar/passive_status.h
index 7122e741..c6252c71 100644
--- a/src/bvar/passive_status.h
+++ b/src/bvar/passive_status.h
@@ -102,7 +102,7 @@ public:
, _series_sampler(nullptr) {
}
- ~PassiveStatus() {
+ ~PassiveStatus() override {
hide();
if (_sampler) {
_sampler->destroy();
@@ -162,11 +162,6 @@ public:
return 0;
}
- Tp reset() {
- CHECK(false) << "PassiveStatus::reset() should never be called, abort";
- abort();
- }
-
protected:
int expose_impl(const butil::StringPiece& prefix,
const butil::StringPiece& n,
diff --git a/test/bvar_histogram_unittest.cpp b/test/bvar_histogram_unittest.cpp
index 80b53c75..f5a83c5d 100644
--- a/test/bvar_histogram_unittest.cpp
+++ b/test/bvar_histogram_unittest.cpp
@@ -21,6 +21,7 @@
#include <stdio.h> // snprintf
#include <string.h> // memset
#include <algorithm> // std::max
+#include <iomanip> // std::setprecision
#include <limits>
#include <map>
#include <memory> // std::make_shared
@@ -30,6 +31,7 @@
#include <gtest/gtest.h>
#include <butil/atomicops.h>
#include <butil/float_util.h>
+#include <butil/logging.h>
#include <butil/strings/string_number_conversions.h>
#include <butil/time.h>
#include "bvar/bvar.h"
@@ -44,6 +46,54 @@ namespace {
// below call private members such as Histogram::get_value() directly.
class HistogramTest : public testing::Test {};
+#if WITH_BABYLON_COUNTER
+
+// One thread's slice of a Histogram, the babylon backed replacement of the
+// ElementContainer tested below.
+TEST_F(HistogramTest, histogram_slot) {
+ bvar::Histogram::BucketSchema schema({10, 20, 30});
+ bvar::detail::HistogramSlot slot;
+ // Freshly constructed, before any add().
+ bvar::Histogram::Value v = slot.load(schema.num_buckets());
+ ASSERT_EQ(0, v.num);
+ ASSERT_DOUBLE_EQ(0.0, v.sum);
+ ASSERT_EQ(0u, v.counts[0]);
+
+ slot.add(schema.index_of(5.25), 5.25);
+ slot.add(schema.index_of(25.5), 25.5);
+ slot.add(schema.index_of(1000.75), 1000.75);
+ v = slot.load(schema.num_buckets());
+ ASSERT_EQ(3, v.num);
+ ASSERT_DOUBLE_EQ(1031.5, v.sum);
+ ASSERT_EQ(4u, v.num_buckets);
+ ASSERT_EQ(1u, v.counts[0]); // 5.25 -> (-inf, 10]
+ ASSERT_EQ(0u, v.counts[1]);
+ ASSERT_EQ(1u, v.counts[2]); // 25.5 -> (20, 30]
+ ASSERT_EQ(1u, v.counts[3]); // 1000.75 -> +Inf
+}
+
+// A slot is taken lazily, on the first record of a thread, and the storage
+// aggregates every slot ever taken.
+TEST_F(HistogramTest, histogram_storage) {
+ bvar::Histogram::BucketSchema schema({10, 20, 30});
+ bvar::detail::HistogramStorage storage(schema.num_buckets());
+ // No thread has recorded anything, yet the combined value already knows
+ // how wide the histogram is.
+ bvar::Histogram::Value v = storage.combine_agents();
+ ASSERT_EQ(0, v.num);
+ ASSERT_EQ(4u, v.num_buckets);
+
+ storage.add(schema.index_of(5.25), 5.25);
+ storage.add(schema.index_of(25.5), 25.5);
+ v = storage.combine_agents();
+ ASSERT_EQ(2, v.num);
+ ASSERT_DOUBLE_EQ(30.75, v.sum);
+ ASSERT_EQ(1u, v.counts[0]);
+ ASSERT_EQ(1u, v.counts[2]);
+}
+
+#else
+
// The element container of a Histogram::Value, which is the generic mutex one:
// the value is far too wide to be atomical.
TEST_F(HistogramTest, element_container) {
@@ -76,6 +126,8 @@ TEST_F(HistogramTest, element_container) {
ASSERT_EQ(0u, v.counts[3]);
}
+#endif // WITH_BABYLON_COUNTER
+
// Fixed workload for export and performance tests, not a library default.
static bvar::Histogram::BucketSchema test_latency_schema() {
return {10, 20, 40, 80, 160, 320, 640, 1280, 2560, 5120,
@@ -724,4 +776,119 @@ TEST_F(HistogramTest, multithreaded) {
}
ASSERT_DOUBLE_EQ((double)expected_sum * nrecords / 40, v.sum);
}
+
+// What a thread recorded outlives it: the AgentCombiner backend commits a
+// dying agent into its global result, the babylon one keeps the slot of an
+// exited thread around for whichever thread inherits its id later.
+TEST_F(HistogramTest, values_of_dead_threads_are_kept) {
+ int64_t nvalues = 100;
+ bvar::Histogram h(bvar::Histogram::BucketSchema({10, 20, 30}));
+ AddArgs args = {&h, nvalues};
+ for (int64_t round = 1; round <= 3; ++round) {
+ pthread_t th;
+ ASSERT_EQ(0, pthread_create(&th, nullptr, add_values, &args));
+ ASSERT_EQ(0, pthread_join(th, nullptr));
+ // Every round runs on a thread which is gone by the time count() reads
+ // what it recorded, and the next one may well reuse its slot.
+ ASSERT_EQ(nvalues * round, h.count());
+ }
+}
+
+static void check_snapshots_while_recording(bvar::Histogram* h,
+ int64_t nrecords) {
+ bvar::Histogram::Value v = h->get_value();
+ std::vector<uint64_t> last_counts(v.num_buckets, 0);
+ for (int i = 0; i < 1000 || v.num < nrecords; ++i) {
+ v = h->get_value();
+ uint64_t total = 0;
+ for (size_t b = 0; b < last_counts.size(); ++b) {
+ total += v.counts[b];
+ // Every sample is taken after the previous one returned and a
+ // bucket count only ever grows, so this snapshot cannot hold less
+ // than the last one did.
+ ASSERT_LE(last_counts[b], v.counts[b]) << "i=" << i << " bucket="
<< b;
+ last_counts[b] = v.counts[b];
+ }
+ ASSERT_EQ(v.num, (int64_t)total) << "i=" << i;
+ }
+}
+
+// The invariant the per thread seqlock buys, the mutex of the ElementContainer
+// without WITH_BABYLON_COUNTER: a snapshot never mixes a bucket that has
+// already been incremented with a `num` that has not. It holds for the slice
+// of one thread, and summing consistent slices keeps it, so the whole snapshot
+// still satisfies the `+Inf bucket == _count` rule of the prometheus format
+// while other threads are recording.
+TEST_F(HistogramTest, snapshot_is_self_consistent_under_contention) {
+ int64_t nvalues = 50000;
+ bvar::Histogram h(bvar::Histogram::BucketSchema({10, 20, 30}));
+
+ pthread_t threads[4];
+ AddArgs args = {&h, nvalues};
+ for (size_t i = 0; i < arraysize(threads); ++i) {
+ ASSERT_EQ(0, pthread_create(&threads[i], nullptr, add_values, &args));
+ }
+
+ int64_t nrecords = (int64_t)arraysize(threads) * nvalues;
+ check_snapshots_while_recording(&h, nrecords);
+
+ for (size_t i = 0; i < arraysize(threads); ++i) {
+ ASSERT_EQ(0, pthread_join(threads[i], nullptr));
+ }
+ ASSERT_EQ(nrecords, h.count());
+}
+
+static const size_t PERF_OPS_PER_THREAD = 500000;
+
+struct PerfArgs {
+ bvar::Histogram* h;
+ int64_t elapsed_ns;
+};
+
+static void* record_into_histogram(void* arg) {
+ PerfArgs* args = (PerfArgs*)arg;
+ butil::Timer timer;
+ timer.start();
+ for (size_t i = 0; i < PERF_OPS_PER_THREAD; ++i) {
+ *args->h << (double)(i % 40);
+ }
+ timer.stop();
+ args->elapsed_ns = timer.n_elapsed();
+ return nullptr;
+}
+
+static double time_records(bvar::Histogram* h, size_t nthread) {
+ PerfArgs proto = {h, 0};
+ std::vector<PerfArgs> args(nthread, proto);
+ std::vector<pthread_t> threads(nthread);
+ for (size_t i = 0; i < nthread; ++i) {
+ EXPECT_EQ(0, pthread_create(&threads[i], nullptr,
+ record_into_histogram, &args[i]));
+ }
+ int64_t total_ns = 0;
+ for (size_t i = 0; i < nthread; ++i) {
+ EXPECT_EQ(0, pthread_join(threads[i], nullptr));
+ total_ns += args[i].elapsed_ns;
+ }
+ return (double)total_ns / (double)(PERF_OPS_PER_THREAD * nthread);
+}
+
+TEST_F(HistogramTest, write_perf) {
+#if WITH_BABYLON_COUNTER
+ const char* backend = "babylon";
+#else
+ const char* backend = "combiner";
+#endif // WITH_BABYLON_COUNTER
+ std::ostringstream oss;
+ oss << std::fixed << std::setprecision(2)
+ << "threads\t" << backend << " (ns per record)\n";
+ for (size_t nthread = 1; nthread <= 8; nthread *= 2) {
+ bvar::Histogram h(bvar::Histogram::BucketSchema({10, 20, 30}));
+ double ns = time_records(&h, nthread);
+ ASSERT_EQ((int64_t)(PERF_OPS_PER_THREAD * nthread), h.count());
+ oss << nthread << '\t' << ns << '\n';
+ }
+ LOG(INFO) << "Histogram write performance:\n" << oss.str();
+}
+
} // namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
