Linear ramping and probabilistic ramping

source/common/BUILD

@@ -67,6 +67,7 @@ envoy_cc_library(
         "//api/client:grpc_service_lib",
         "//include/nighthawk/client:client_includes",
         "//include/nighthawk/common:base_includes",
+        "@com_google_absl//absl/random",
         "@dep_hdrhistogram_c//:hdrhistogram_c",
         "@envoy//source/common/common:assert_lib_with_external_headers",
         "@envoy//source/common/common:lock_guard_lib_with_external_headers",

source/common/rate_limiter_impl.cc

@@ -78,6 +78,43 @@ void LinearRateLimiter::releaseOne() {
   acquired_count_--;
 }
 
+LinearRampingRateLimiterImpl::LinearRampingRateLimiterImpl(Envoy::TimeSource& time_source,
+                                                           const std::chrono::nanoseconds ramp_time,
+                                                           const Frequency frequency)
+    : RateLimiterBaseImpl(time_source), ramp_time_(ramp_time), frequency_(frequency) {
+  if (frequency_.value() <= 0) {
+    throw NighthawkException("frequency must be > 0");
+  }
+  if (ramp_time <= 0ns) {
+    throw NighthawkException("duration must be positive");
+  }
+}
+
+bool LinearRampingRateLimiterImpl::tryAcquireOne() {
+  if (acquireable_count_) {
+    acquired_count_++;
+    return acquireable_count_--;
+  }
+  const auto elapsed_time = elapsed() + 1ns;
+  double elapsed_fraction = 1.0;
+  if (elapsed_time < ramp_time_) {
+    elapsed_fraction -= static_cast<double>(ramp_time_.count() - elapsed_time.count()) /
+                        static_cast<double>(ramp_time_.count());
+  }
+
+  const double current_frequency = elapsed_fraction * (frequency_.value() * 1.0);
+  const double chrono_seconds =
+      std::chrono::duration_cast<std::chrono::duration<double>>(elapsed_time).count();
+  const double total = chrono_seconds * current_frequency / 2.0;
+  acquireable_count_ = std::round(total) - acquired_count_;
+  return acquireable_count_ > 0 ? tryAcquireOne() : false;
+}
+
+void LinearRampingRateLimiterImpl::releaseOne() {
+  acquireable_count_++;
+  acquired_count_--;
+}
+
 DelegatingRateLimiterImpl::DelegatingRateLimiterImpl(
     RateLimiterPtr&& rate_limiter, RateLimiterDelegate random_distribution_generator)
     : ForwardingRateLimiterImpl(std::move(rate_limiter)),
@@ -118,4 +155,27 @@ bool FilteringRateLimiterImpl::tryAcquireOne() {
   return rate_limiter_->tryAcquireOne() ? filter_() : false;
 }
 
+GraduallyOpeningRateLimiterFilter::GraduallyOpeningRateLimiterFilter(
+    const std::chrono::nanoseconds ramp_time, DiscreteNumericDistributionSamplerPtr&& provider,
+    RateLimiterPtr&& rate_limiter)
+    : FilteringRateLimiterImpl(
+          std::move(rate_limiter),
+          [this]() {
+            if (elapsed() < ramp_time_) {
+              const double chance_percentage =
+                  100.0 - (static_cast<double>(ramp_time_.count() - elapsed().count()) /
+                           (ramp_time_.count() * 1.0)) *
+                              100.0;
+              return std::round(provider_->getValue() / 10000.0) <= chance_percentage;
+            }
+            return true;
+          }),
+      provider_(std::move(provider)), ramp_time_(ramp_time) {}
+
+ZipfRateLimiterImpl::ZipfRateLimiterImpl(RateLimiterPtr&& rate_limiter, bool deterministic,
+                                         double q, double v)
+    : FilteringRateLimiterImpl(std::move(rate_limiter),
+                               [this]() { return deterministic_ ? dist_(mt_) : dist_(g_); }),
+      dist_(1, q, v), deterministic_(deterministic) {}
+
 } // namespace Nighthawk

source/common/rate_limiter_impl.h

@@ -10,6 +10,8 @@
 
 #include "common/frequency.h"
 
+#include "absl/random/random.h"
+#include "absl/random/zipf_distribution.h"
 #include "absl/types/optional.h"
 
 namespace Nighthawk {
@@ -55,6 +57,24 @@ class LinearRateLimiter : public RateLimiterBaseImpl,
   const Frequency frequency_;
 };
 
+/**
+ * A rate limiter which linearly ramps up to the desired frequency over the specified period.
+ */
+class LinearRampingRateLimiterImpl : public RateLimiterBaseImpl,
+                                     public Envoy::Logger::Loggable<Envoy::Logger::Id::main> {
+public:
+  LinearRampingRateLimiterImpl(Envoy::TimeSource& time_source,
+                               const std::chrono::nanoseconds ramp_time, const Frequency frequency);
+  bool tryAcquireOne() override;
+  void releaseOne() override;
+
+private:
+  int64_t acquireable_count_{0};
+  uint64_t acquired_count_{0};
+  const std::chrono::nanoseconds ramp_time_;
+  const Frequency frequency_;
+};
+
 /**
  * Base for a rate limiter which wraps another rate limiter, and forwards
  * some calls.
@@ -157,4 +177,44 @@ class FilteringRateLimiterImpl : public ForwardingRateLimiterImpl,
   const RateLimiterFilter filter_;
 };
 
+/**
+ * Takes a probabilistic approach to suppressing an arbitrary wrapper rate limiter.
+ */
+class GraduallyOpeningRateLimiterFilter : public FilteringRateLimiterImpl {
+public:
+  GraduallyOpeningRateLimiterFilter(const std::chrono::nanoseconds ramp_time,
+                                    DiscreteNumericDistributionSamplerPtr&& provider,
+                                    RateLimiterPtr&& rate_limiter);
+
+private:
+  DiscreteNumericDistributionSamplerPtr provider_;
+  const std::chrono::nanoseconds ramp_time_;
+};
+
+/**
+ * Thin wrapper around absl::zipf_distribution that will pull zeroes and ones from the distribution
+ * with the intent to probabilistically suppress the wrapped rate limiter.
+ * This may need further consideration, because it will shoot holes in the pacing, lowering the
+ * actual achieved frequency.
+ */
+class ZipfRateLimiterImpl : public FilteringRateLimiterImpl {
+public:
+  /**
+   * From the absl header associated to the zipf distribution:
+   * Preconditions: v > 0, q > 1
+   * The precondidtions are validated when NDEBUG is not defined via
+   * a pair of assert() directives.
+   * If NDEBUG is defined and either or both of these parameters take invalid
+   * values, the behavior of the class is undefined.
+   */
+  ZipfRateLimiterImpl(RateLimiterPtr&& rate_limiter, bool deterministic, double q = 2.0,
+                      double v = 1.0);
+
+private:
+  absl::zipf_distribution<uint64_t> dist_;
+  absl::InsecureBitGen g_;
+  std::mt19937_64 mt_;
+  bool deterministic_;
+};
+
 } // namespace Nighthawk

test/rate_limiter_test.cc

@@ -172,4 +172,123 @@ TEST_F(RateLimiterTest, DistributionSamplingRateLimiterImplSchedulingTest) {
   EXPECT_TRUE(rate_limiter->tryAcquireOne());
 }
 
+// TODO(oschaaf): once we have hr sleep, test at a higher res.
+class LinearRampingRateLimiterImplTest : public Test {
+public:
+  std::vector<int64_t> getAcquisitionTimings(const Frequency frequency,
+                                             const std::chrono::seconds duration) {
+    Envoy::Event::SimulatedTimeSystem time_system;
+    std::vector<int64_t> aquisition_timings;
+    LinearRampingRateLimiterImpl rate_limiter(time_system, duration, frequency);
+    auto total_ms_elapsed = 0ms;
+    const auto clock_tick = 1ms;
+    auto last_acquisition_timestamp = 0ms;
+    EXPECT_FALSE(rate_limiter.tryAcquireOne());
+    do {
+      if (rate_limiter.tryAcquireOne()) {
+        EXPECT_FALSE(rate_limiter.tryAcquireOne());
+        aquisition_timings.push_back(total_ms_elapsed.count());
+        last_acquisition_timestamp = total_ms_elapsed;
+      }
+      time_system.sleep(clock_tick);
+      total_ms_elapsed += clock_tick;
+    } while (total_ms_elapsed <= duration);
+    return aquisition_timings;
+  }
+};
+
+TEST_F(RateLimiterTest, LinearRampingRateLimiterImplInvalidArgumentTest) {
+  Envoy::Event::SimulatedTimeSystem time_system;
+  // bad frequency
+  EXPECT_THROW(LinearRampingRateLimiterImpl rate_limiter(time_system, 1s, 0_Hz);
+               , NighthawkException);
+  // bad ramp duration
+  EXPECT_THROW(LinearRampingRateLimiterImpl rate_limiter(time_system, 0s, 1_Hz);
+               , NighthawkException);
+  EXPECT_THROW(LinearRampingRateLimiterImpl rate_limiter(time_system, -1s, 1_Hz);
+               , NighthawkException);
+}
+
+TEST_F(LinearRampingRateLimiterImplTest, TimingVerificationTest) {
+  EXPECT_EQ(getAcquisitionTimings(5_Hz, 5s),
+            std::vector<int64_t>(
+                {1000, 1733, 2237, 2646, 3000, 3317, 3606, 3873, 4124, 4359, 4583, 4796, 5000}));
+  EXPECT_EQ(getAcquisitionTimings(4_Hz, 2s), std::vector<int64_t>({708, 1225, 1582, 1871}));
+}
+
+class GraduallyOpeningRateLimiterFilterTest : public Test {
+public:
+  std::vector<int64_t> getAcquisitionTimings(const Frequency frequency,
+                                             const std::chrono::seconds duration) {
+    Envoy::Event::SimulatedTimeSystem time_system;
+    std::vector<int64_t> aquisition_timings;
+    auto* unsafe_discrete_numeric_distribution_sampler =
+        new MockDiscreteNumericDistributionSampler();
+    std::mt19937_64 mt(1243);
+    std::uniform_int_distribution<uint64_t> dist(1, 1000000);
+    EXPECT_CALL(*unsafe_discrete_numeric_distribution_sampler, getValue)
+        .Times(AtLeast(1))
+        .WillRepeatedly(Invoke([&dist, &mt]() { return dist(mt); }));
+    RateLimiterPtr rate_limiter = std::make_unique<GraduallyOpeningRateLimiterFilter>(
+        duration,
+        std::unique_ptr<DiscreteNumericDistributionSampler>(
+            unsafe_discrete_numeric_distribution_sampler),
+        std::make_unique<LinearRateLimiter>(time_system, frequency));
+    auto total_ms_elapsed = 0ms;
+    auto clock_tick = 1ms;
+    EXPECT_FALSE(rate_limiter->tryAcquireOne());
+
+    do {
+      if (rate_limiter->tryAcquireOne()) {
+        aquisition_timings.push_back(total_ms_elapsed.count());
+        EXPECT_FALSE(rate_limiter->tryAcquireOne());
+      }
+      time_system.sleep(clock_tick);
+      total_ms_elapsed += clock_tick;
+    } while (total_ms_elapsed <= duration);
+
+    EXPECT_FALSE(rate_limiter->tryAcquireOne());
+    time_system.sleep(1s);
+    // Verify that after the rampup the expected constant pacing is maintained.
+    // Calls should be forwarded to the regular linear rate limiter algorithm with its
+    // corrective behavior so we can expect to acquire a series with that.
+    for (uint64_t i = 0; i < frequency.value(); i++) {
+      EXPECT_TRUE(rate_limiter->tryAcquireOne());
+    }
+    // Verify we acquired everything.
+    EXPECT_FALSE(rate_limiter->tryAcquireOne());
+    return aquisition_timings;
+  }
+};
+
+TEST_F(GraduallyOpeningRateLimiterFilterTest, TimingVerificationTest) {
+  EXPECT_EQ(getAcquisitionTimings(50_Hz, 1s),
+            std::vector<int64_t>({120, 320, 380, 560, 580, 600, 620, 640, 660, 680, 700, 740,
+                                  760, 780, 840, 860, 880, 900, 920, 940, 960, 980, 1000}));
+}
+
+class ZipfRateLimiterImplTest : public Test {};
+
+TEST_F(ZipfRateLimiterImplTest, TimingVerificationTest) {
+  Envoy::Event::SimulatedTimeSystem time_system;
+  auto rate_limiter = std::make_unique<ZipfRateLimiterImpl>(
+      std::make_unique<LinearRateLimiter>(time_system, 10_Hz), true);
+  const std::chrono::seconds duration = 15s;
+  std::vector<int64_t> aquisition_timings;
+  auto total_ms_elapsed = 0ms;
+  auto clock_tick = 1ms;
+
+  do {
+    if (rate_limiter->tryAcquireOne()) {
+      aquisition_timings.push_back(total_ms_elapsed.count());
+    }
+    time_system.sleep(clock_tick);
+    total_ms_elapsed += clock_tick;
+  } while (total_ms_elapsed <= duration);
+  EXPECT_EQ(aquisition_timings,
+            std::vector<int64_t>({500,   800,   1300,  2400,  2900,  3900,  4200,  4400,  4500,
+                                  5800,  6000,  6400,  7900,  8400,  8600,  9900,  10200, 10500,
+                                  10600, 12000, 12300, 12600, 13300, 13600, 13700, 13800, 13900}));
+}
+
 } // namespace Nighthawk