diff --git a/src/bvar/detail/sampler.h b/src/bvar/detail/sampler.h index d120458d18..2a9815c044 100644 --- a/src/bvar/detail/sampler.h +++ b/src/bvar/detail/sampler.h @@ -182,6 +182,7 @@ class CombinerSampleSource { // sampling immune to R being destructed first. template class ReducerSampler : public Sampler { + friend class LatencyRecorderTestPeer; typedef typename butil::conditional< HasShareCombiner::value, CombinerSampleSource, diff --git a/src/bvar/latency_recorder.cpp b/src/bvar/latency_recorder.cpp index 8c9379d8c2..6db4764b15 100644 --- a/src/bvar/latency_recorder.cpp +++ b/src/bvar/latency_recorder.cpp @@ -113,10 +113,10 @@ int CDF::describe_series( return 0; } -// Return random int value with expectation = `dval' -static int64_t double_to_random_int(double dval) { +// Return random int value with expectation = `dval'. +int64_t double_to_random_int(double dval, double random) { int64_t ival = static_cast(dval); - if (dval > ival + butil::fast_rand_double()) { + if (dval > ival + random) { ival += 1; } return ival; @@ -129,7 +129,8 @@ static int64_t get_window_recorder_qps(void* arg) { if (s.time_us <= 0) { return 0; } - return double_to_random_int(s.data.num * 1000000.0 / s.time_us); + return double_to_random_int( + s.data.num * 1000000.0 / s.time_us, butil::fast_rand_double()); } static int64_t get_recorder_count(void* arg) { @@ -209,7 +210,8 @@ int64_t LatencyRecorder::qps(time_t window_size) const { if (s.time_us <= 0) { return 0; } - return detail::double_to_random_int(s.data.num * 1000000.0 / s.time_us); + return detail::double_to_random_int( + s.data.num * 1000000.0 / s.time_us, butil::fast_rand_double()); } int LatencyRecorder::expose(const butil::StringPiece& prefix1, diff --git a/src/bvar/latency_recorder.h b/src/bvar/latency_recorder.h index 1cc09f51e3..8c820600b9 100644 --- a/src/bvar/latency_recorder.h +++ b/src/bvar/latency_recorder.h @@ -33,6 +33,10 @@ typedef Window RecorderWindow; typedef Window, SERIES_IN_SECOND> MaxWindow; typedef Window PercentileWindow; +// Round |dval| up with probability equal to its fractional part. |random| +// must be in [0, 1); keeping it explicit makes the rule deterministic to test. +int64_t double_to_random_int(double dval, double random); + // NOTE: Always use int64_t in the interfaces no matter what the impl. is. class CDF : public Variable { diff --git a/src/bvar/window.h b/src/bvar/window.h index d701dcc08d..b1adb1ad75 100644 --- a/src/bvar/window.h +++ b/src/bvar/window.h @@ -43,6 +43,7 @@ namespace detail { // Just for constructor reusing of Window<> template class WindowBase : public Variable { + friend class LatencyRecorderTestPeer; public: typedef typename R::value_type value_type; typedef typename R::sampler_type sampler_type; diff --git a/test/bvar_recorder_unittest.cpp b/test/bvar_recorder_unittest.cpp index d9d3fd7fb3..bf6544632d 100644 --- a/test/bvar_recorder_unittest.cpp +++ b/test/bvar_recorder_unittest.cpp @@ -19,15 +19,63 @@ #include // pthread_* +#include #include #include #include +#include #include "butil/time.h" #include "butil/macros.h" #include "bvar/recorder.h" #include "bvar/latency_recorder.h" #include +namespace bvar { +namespace detail { + +// Give the real QPS APIs a fixed history without racing the sampling thread. +// The recorder's original sampler remains registered and is restored before +// destruction. Only this replacement sampler's queue is modified by the test. +class LatencyRecorderTestPeer { +public: + LatencyRecorderTestPeer(RecorderWindow& window, IntRecorder& recorder, + int count) + : _window(window), _original(window._sampler), + _sampler(new IntRecorder::sampler_type(&recorder)) { + CHECK_EQ(0, _sampler->set_window_size(4)); + _sampler->take_sample(); // Allocate room for five snapshots. + _sampler->_q.clear(); + for (int i = 0; i <= 4; ++i) { + CHECK(_sampler->_q.push(Sample( + i == 4 ? Stat(count, count) : Stat(), i * 1000000L))); + } + _sampler->add_borrower(); + _window._sampler = _sampler.get(); + // Keep the original borrow alive until it is restored. + } + + ~LatencyRecorderTestPeer() { + _window._sampler = _original; + _sampler->remove_borrower(); + } + + void advance() { + Sample next = *_sampler->_q.bottom(); + next.time_us += 1000000L; + _sampler->_q.elim_push(next); + } + +private: + LatencyRecorderTestPeer(const LatencyRecorderTestPeer&) = delete; + LatencyRecorderTestPeer& operator=(const LatencyRecorderTestPeer&) = delete; + RecorderWindow& _window; + IntRecorder::sampler_type* _original; + std::unique_ptr _sampler; +}; + +} // namespace detail +} // namespace bvar + namespace { #if !WITH_BABYLON_COUNTER TEST(RecorderTest, test_complement) { @@ -222,6 +270,16 @@ class LatencyRecorderWithWindowSample : public bvar::LatencyRecorder { bool get_latency_sample(bvar::detail::Sample* sample) const { return _latency_window.get_span(sample); } + + bool get_latency_sample(time_t window_size, + bvar::detail::Sample* sample) const { + return _latency_window.get_span(window_size, sample); + } + + std::unique_ptr fixed_history(int count) { + return std::unique_ptr( + new bvar::detail::LatencyRecorderTestPeer(_latency_window, _latency, count)); + } }; TEST(RecorderTest, latency_recorder_window_statistics) { @@ -276,44 +334,52 @@ TEST(RecorderTest, latency_recorder_window_statistics) { } TEST(RecorderTest, latency_recorder_qps_accuracy) { - bvar::LatencyRecorder lr1(2); // set windows size to 2s - bvar::LatencyRecorder lr2(2); - bvar::LatencyRecorder lr3(2); - bvar::LatencyRecorder lr4(2); - usleep(3000000); // wait sampler to sample 3 times - - auto write = [](bvar::LatencyRecorder& lr, int times) { - for (int i = 0; i < times; ++i) { - lr << 1; - } - }; - write(lr1, 10); - write(lr2, 11); - write(lr3, 3); - write(lr4, 1); - usleep(1000000); // wait sampler to sample 1 time - - auto read = [](bvar::LatencyRecorder& lr, double exp_qps, int window_size = 0) { - int64_t qps_sum = 0; - int64_t exp_qps_int = (int64_t)exp_qps; - for (int i = 0; i < 1000; ++i) { - int64_t qps = window_size ? lr.qps(window_size): lr.qps(); - EXPECT_GE(qps, exp_qps_int - 1); - EXPECT_LE(qps, exp_qps_int + 1); - qps_sum += qps; + for (int count : {10, 11, 3, 1}) { + for (time_t window : {2, 3}) { + SCOPED_TRACE(::testing::Message() << "count=" << count + << " window=" << window); + // Default qps() uses two seconds; explicit qps(3) must override + // a four-second default. + LatencyRecorderWithWindowSample lr(window == 2 ? 2 : 4); + auto history = lr.fixed_history(count); + bvar::detail::Sample sample; + ASSERT_TRUE(lr.get_latency_sample(window, &sample)); + ASSERT_EQ(count, sample.data.num); + ASSERT_EQ(window * 1000000L, sample.time_us); + + const double expected = count / static_cast(window); + int64_t total = 0; + const int trials = 10000; + for (int i = 0; i < trials; ++i) { + const int64_t value = window == 2 ? lr.qps() : lr.qps(window); + ASSERT_GE(value, static_cast(std::floor(expected))); + ASSERT_LE(value, static_cast(std::ceil(expected))); + total += value; + } + // Exercise rounding through the public APIs, including low QPS. + // For independent uniform draws, Hoeffding bounds the chance of + // this tolerance failing at less than 2*exp(-50) per case. + EXPECT_NEAR(expected, total / static_cast(trials), 0.05); + + // Move the batch out of the requested window. Duration stays the + // same while count becomes zero, reproducing the rollover case. + for (time_t i = 0; i < window; ++i) { + history->advance(); + } + ASSERT_TRUE(lr.get_latency_sample(window, &sample)); + EXPECT_EQ(0, sample.data.num); + EXPECT_EQ(window * 1000000L, sample.time_us); + EXPECT_EQ(0, window == 2 ? lr.qps() : lr.qps(window)); } - double err = fabs(qps_sum / 1000.0 - exp_qps); - return err; - }; - ASSERT_GT(0.1, read(lr1, 10/2.0)); - ASSERT_GT(0.1, read(lr2, 11/2.0)); - ASSERT_GT(0.1, read(lr3, 3/2.0)); - ASSERT_GT(0.1, read(lr4, 1/2.0)); - - ASSERT_GT(0.1, read(lr1, 10/3.0, 3)); - ASSERT_GT(0.1, read(lr2, 11/3.0, 3)); - ASSERT_GT(0.1, read(lr3, 3/3.0, 3)); - ASSERT_GT(0.1, read(lr4, 1/3.0, 3)); + } +} + +TEST(RecorderTest, latency_recorder_qps_rounding) { + // A random value equal to the fractional part rounds down. + EXPECT_EQ(10, bvar::detail::double_to_random_int(10.25, 0.25)); + EXPECT_EQ(11, bvar::detail::double_to_random_int(10.25, 0.24)); + EXPECT_EQ(10, bvar::detail::double_to_random_int(10.0, 0.0)); + EXPECT_EQ(11, bvar::detail::double_to_random_int(10.75, 0.1)); } } // namespace