Fix flaky latency recorder QPS test

src/bvar/detail/sampler.h

@@ -182,6 +182,7 @@ class CombinerSampleSource {
 // sampling immune to R being destructed first.
 template <typename R, typename T, typename Op, typename InvOp>
 class ReducerSampler : public Sampler {
+    friend class LatencyRecorderTestPeer;
     typedef typename butil::conditional<
         HasShareCombiner<R>::value,
         CombinerSampleSource<R, T, Op, InvOp>,

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<int64_t>(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,

src/bvar/latency_recorder.h

@@ -33,6 +33,10 @@ typedef Window<IntRecorder, SERIES_IN_SECOND> RecorderWindow;
 typedef Window<Maxer<int64_t>, SERIES_IN_SECOND> MaxWindow;
 typedef Window<Percentile, SERIES_IN_SECOND> 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 {

src/bvar/window.h

@@ -43,6 +43,7 @@ namespace detail {
 // Just for constructor reusing of Window<>
 template <typename R, SeriesFrequency series_freq>
 class WindowBase : public Variable {
+    friend class LatencyRecorderTestPeer;
 public:
     typedef typename R::value_type value_type;
     typedef typename R::sampler_type sampler_type;

test/bvar_recorder_unittest.cpp

@@ -19,15 +19,63 @@
 
 #include <pthread.h>                                // pthread_*
 
+#include <cmath>
 #include <cstddef>
 #include <memory>
 #include <iostream>
+#include <vector>
 #include "butil/time.h"
 #include "butil/macros.h"
 #include "bvar/recorder.h"
 #include "bvar/latency_recorder.h"
 #include <gtest/gtest.h>
 
+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<Stat>(
+                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<Stat> 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<IntRecorder::sampler_type> _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<bvar::Stat>* sample) const {
         return _latency_window.get_span(sample);
     }
+
+    bool get_latency_sample(time_t window_size,
+                            bvar::detail::Sample<bvar::Stat>* sample) const {
+        return _latency_window.get_span(window_size, sample);
+    }
+
+    std::unique_ptr<bvar::detail::LatencyRecorderTestPeer> fixed_history(int count) {
+        return std::unique_ptr<bvar::detail::LatencyRecorderTestPeer>(
+            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<bvar::Stat> 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<double>(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<int64_t>(std::floor(expected)));
+                ASSERT_LE(value, static_cast<int64_t>(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<double>(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