chenBright commented on code in PR #3561:
URL: https://github.com/apache/brpc/pull/3561#discussion_r4073474895
##########
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:
Fixed in e4d8557ac9e9204df7ef8700ec4bc1740a4edee5 .
##########
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:
Fixed in e4d8557ac9e9204df7ef8700ec4bc1740a4edee5 .
##########
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:
Fixed in e4d8557ac9e9204df7ef8700ec4bc1740a4edee5 .
--
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]
