Copilot commented on code in PR #3561:
URL: https://github.com/apache/brpc/pull/3561#discussion_r4072916338
##########
src/bvar/passive_status.h:
##########
@@ -102,7 +102,7 @@ class PassiveStatus : public Variable {
, _series_sampler(nullptr) {
}
- ~PassiveStatus() {
+ ~PassiveStatus() override {
hide();
if (_sampler) {
_sampler->destroy();
Review Comment:
This PR removes `PassiveStatus::reset()` (which previously CHECK+abort'ed).
That changes the public behavior if any code calls `reset()` on a
`PassiveStatus` (it may now dispatch to a base implementation or become usable
unexpectedly). If `reset()` must remain unsupported, consider restoring an
override that hard-fails (or explicitly deleting it) to preserve the previous
contract.
##########
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;
+ uint64_t last_counts[4] = {};
+ for (int i = 0; i < 1000 || v.num < nrecords; ++i) {
+ v = h->get_value();
+ uint64_t total = 0;
+ for (size_t b = 0; b < arraysize(last_counts); ++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;
+ long elapsed_ns;
Review Comment:
`elapsed_ns` is stored in `long`, which can be 32-bit on some platforms and
may overflow for nanosecond timings. Prefer `int64_t` (or `long long`) for
`elapsed_ns` and for `total_ns` in `time_records()` to keep behavior correct
across architectures.
##########
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",
+ ],
Review Comment:
The PR description fields look incomplete/placeholder (e.g., 'Issue Number:
resolve', empty 'Problem Summary' / 'Changed'). Since this PR adds a new
backend and significant tests, please update the description to clearly state
the problem being solved, the intended behavior behind `WITH_BABYLON_COUNTER`,
and any expected side effects (perf/memory/behavioral differences).
##########
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;
+ uint64_t last_counts[4] = {};
+ for (int i = 0; i < 1000 || v.num < nrecords; ++i) {
+ v = h->get_value();
+ uint64_t total = 0;
+ for (size_t b = 0; b < arraysize(last_counts); ++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;
+ long 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]));
+ }
+ long 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;
+ }
Review Comment:
`total_ns` uses `long` and accumulates per-thread elapsed nanoseconds; on
32-bit `long` this can overflow even for modest runtimes. Use a 64-bit type
(e.g., `int64_t`) consistently for both the per-thread and accumulated values.
##########
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;
+ uint64_t last_counts[4] = {};
+ for (int i = 0; i < 1000 || v.num < nrecords; ++i) {
+ v = h->get_value();
+ uint64_t total = 0;
+ for (size_t b = 0; b < arraysize(last_counts); ++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;
+ long 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]));
+ }
+ long 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) {
Review Comment:
This adds a performance measurement as a regular unit test, with a fairly
large workload (`PERF_OPS_PER_THREAD=500000` across up to 8 threads). That can
significantly increase CI runtime and can be noisy (it always logs results).
Consider marking it as disabled-by-default (e.g., `DISABLED_write_perf`),
gating it behind an environment flag, or moving it to a benchmark/perf test
target so normal test runs remain fast and deterministic.
##########
src/bvar/detail/sampler.h:
##########
@@ -313,6 +299,23 @@ class ReducerSampler : public Sampler {
}
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.
+ // Suming 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
Review Comment:
Correct wording in comment: 'Inversed' should be 'Inverse' (e.g., 'Inverse
operation').
##########
src/bvar/detail/sampler.h:
##########
@@ -313,6 +299,23 @@ class ReducerSampler : public Sampler {
}
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.
+ // Suming up samples gives the result within a window.
Review Comment:
Correct spelling in comment: 'Suming' should be 'Summing'.
##########
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;
+ uint64_t last_counts[4] = {};
+ for (int i = 0; i < 1000 || v.num < nrecords; ++i) {
+ v = h->get_value();
+ uint64_t total = 0;
+ for (size_t b = 0; b < arraysize(last_counts); ++b) {
Review Comment:
`last_counts` is hard-coded to 4 buckets, which implicitly assumes the
schema `{10, 20, 30}` forever. If this helper is reused or the schema changes,
the test will silently become incorrect. Consider sizing `last_counts` from
`v.num_buckets` (e.g., a `std::vector<uint64_t>` initialized after the first
`get_value()`), and iterating to `v.num_buckets`.
##########
src/bvar/histogram.h:
##########
@@ -192,7 +222,13 @@ class Histogram : public Variable {
// 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.
Review Comment:
The comment refers to `Percentile`, but this is `Histogram`. Please update
the wording to avoid misleading future readers.
--
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]
