Copilot commented on code in PR #3561:
URL: https://github.com/apache/brpc/pull/3561#discussion_r4073670286
##########
test/bvar_histogram_unittest.cpp:
##########
@@ -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;
Review Comment:
This adds a heavy performance-style test (up to ~4M records plus threading)
that will run unconditionally in CI and also emits INFO logs. Consider making
it opt-in (e.g., `DISABLED_write_perf`, or gated on an env var/flag) or moving
it to a benchmark target, to avoid slowing and destabilizing the unit test
suite.
##########
test/bvar_histogram_unittest.cpp:
##########
@@ -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();
+}
Review Comment:
This adds a heavy performance-style test (up to ~4M records plus threading)
that will run unconditionally in CI and also emits INFO logs. Consider making
it opt-in (e.g., `DISABLED_write_perf`, or gated on an env var/flag) or moving
it to a benchmark target, to avoid slowing and destabilizing the unit test
suite.
##########
src/bvar/histogram.h:
##########
@@ -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) {
Review Comment:
For integral counters (`_counts`, `_num`), using `fetch_add(...,
memory_order_relaxed)` would more directly express intent and avoids relying on
the single-writer assumption for correctness. If `butil::atomic` doesn’t
support `fetch_add` for `double`, you could still switch the integer fields to
`fetch_add` while keeping the double path as-is.
##########
test/bvar_histogram_unittest.cpp:
##########
@@ -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);
+}
Review Comment:
The reported metric is the average of per-thread wall-clock times divided by
ops, which is closer to 'avg per-thread ns/record' than global throughput (wall
time). Either clarify the label/output to match what's being measured or change
the measurement to use a single wall-clock timer around the whole threaded run
if you want aggregate throughput.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
