Throttle fixes (#554)

* `Strings` are UTF-8 encoded and use 1 byte for ascii chars

* Refill capacity based on elapsed time

* Wait based on refill per tick

* Clarify variable names, fix division by zero

Signed-off-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>

* Changelog formatting

* Add changelog entry

---------

Signed-off-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>
Co-authored-by: Brian L. Troutwine <brian.troutwine@datadoghq.com>

Author: GeorgeHahn

CHANGELOG.md

@@ -5,13 +5,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
-## Added
+### Added
 - Added target metrics support for Go expvars
 
+### Fixed
+- Fixed throttle behavior for generators that run very quickly
+
 ## [0.14.0]
-## Added
+### Added
 - Added the ability to configure details about DogStatsD payload
-## Changed
+### Changed
 - Datadog logs generation now much faster, relying on an experimentally
   determined 'encoded size' rather than searching at runtime for the same.
   (PR #564).
@@ -21,30 +24,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   of some memory during the generation process. (PR #547)
 
 ## [0.13.1]
-## Added
+### Added
 - Introduced Proportional Set Size (PSS) memory measurement under the
   `pss_bytes` metric.
 - Convert `Block` to use `Bytes` type instead of `Vec<u8>`.
 - Introduce Datadog trace-agent payload support in JSON and MsgPack serialization.
 
 ## [0.13.0]
-## Added
+### Added
 - Introduced automatic throttling into generators to search for stable target load.
-## Fixed
+### Fixed
 - Use saturating addition in observer stats gathering routine
 
 ## [0.12.0]
-## Added
+### Added
 - Added the ability for lading to back-off load generation based on RSS limits.
 - Process tree generator contributed by @safchain
 - Fixed: OpenTelemetry message length calculation corrected for some messages.
 - **Breaking change:** Split UDS support between explict datagram and stream modules.
 - Fixed: Corrected mistakes in the DogStatsD payload implementation.
-## Changed
+### Changed
 - Adjusted the cardinality of DogStatsD keys, values and names downward.
 - **Breaking change:** Added support for DogStatsD payload.
 - **Breaking change:** Support for Kafka generator is removed.
-## Fixed
+### Fixed
 - Lading's UDS will now re-attempt to connect to a UDS socket, rather than
   erroring.
 

src/payload/common.rs

@@ -27,8 +27,8 @@ impl Generator<String> for AsciiString {
         R: rand::Rng + ?Sized,
     {
         let len: usize = rng.gen_range(1..self.max_length) as usize;
-        let total_bytes = 4 * len; // max size of an `char` times length
-        let mut s = String::with_capacity(total_bytes);
+        let total_bytes = len;
+        let mut s = String::new();
         s.reserve(total_bytes);
         s.extend(
             CHARSET

src/throttle.rs

@@ -85,6 +85,7 @@ pub(crate) struct GenericThrottle<C>
 where
     C: Clock,
 {
+    last_tick: u64,
     spare_capacity: u64,
     /// The maximum capacity of `Throttle` past which no more capacity will be
     /// added.
@@ -118,13 +119,12 @@ where
         // 'interval' happens once every second. If we allow for the tick of
         // Throttle to be one per microsecond that's 1x10^6 ticks per interval.
 
-        let refill_per_tick = u64::from(maximum_capacity.get()) / INTERVAL_TICKS;
-
-        // let rate_limiter = RateLimiter::direct(Quota::per_second(maximum_capacity));
-        // let maximum_capacity = maximum_capacity.get();
-        // let interval_actual_budget = maximum_capacity / 10;
+        // We do not want a situation where refill never happens. If the maximum
+        // capacity is less than INTERVAL_TICKS we set the floor at 1.
+        let refill_per_tick = cmp::max(1, u64::from(maximum_capacity.get()) / INTERVAL_TICKS);
 
         Self {
+            last_tick: clock.ticks_elapsed(),
             maximum_capacity: u64::from(maximum_capacity.get()),
             refill_per_tick,
             requested_budget: 0,
@@ -206,9 +206,11 @@ where
         // called. Depending on how long ago this was we may have completely
         // filled up throttle capacity. Note we fill to the _projected_ budget
         // and not the maximum capacity.
-        let ticks_since = self.clock.ticks_elapsed();
+        let ticks_since_start = self.clock.ticks_elapsed();
+        let ticks_since_last_wait = ticks_since_start.saturating_sub(self.last_tick);
+        self.last_tick = ticks_since_start;
         let refilled_capacity: u64 = cmp::min(
-            ticks_since
+            ticks_since_last_wait
                 .saturating_mul(self.refill_per_tick)
                 .saturating_add(self.spare_capacity),
             self.projected_budget,
@@ -221,7 +223,7 @@ where
         // budget for those 'missing' intervals, although that would be
         // interesting to do maybe.
 
-        let current_interval = ticks_since / INTERVAL_TICKS;
+        let current_interval = ticks_since_start / INTERVAL_TICKS;
         if current_interval == self.interval {
             // Intentionally blank. There is nothing to do in the event we are
             // in the same interval, with regard to budgets.
@@ -264,7 +266,7 @@ where
             // need to pass before capacity is sufficient, force the client to
             // wait that amount of time.
             self.spare_capacity = 0;
-            let slop = capacity_request - refilled_capacity;
+            let slop = (capacity_request - refilled_capacity) / self.refill_per_tick;
             self.clock.wait(slop).await;
         }
         Ok(())