Repository: pinecone-io/VSB

Issue:
Add jitter to when each user starts
When running with multiple users (`--users=N`) and a fixed operation rate (`--requests_per_sec=R`), requests from different users will initially be corrdinated with each other - we spawn N greenlets, then each greenlet issues the first request as soon as possible, then sleeps for `1 / per_user_requests_per_sec`, then issues the next request. This results in N requests hitting the DB at the ~same time. This can result in artificially "bursty" requests to the DB, particulary when users is high compared to the requests_per_sec.

For example, a graph of when each request is issued by VSB within each 1 second window, for 10 users and 10 requests per second (so each user performing 1 request per second):

![Image](https://github.com/user-attachments/assets/4ab5023f-3ded-4693-aee4-2f6f06c8859d)

Observe that all requests (other than initial ramp up) occur at ~0.9s in each 1 second window.
 
We should avoid this so requests are scheduled to occur randomly distributed across the time interval, instead of all at the start of the interval.

Hints from issue comments:

Solution patch:
diff --git a/vsb/randomized_pacer.py b/vsb/randomized_pacer.py
new file mode 100644
index 0000000..6633af6
--- /dev/null
+++ b/vsb/randomized_pacer.py
@@ -0,0 +1,58 @@
+import time
+import random
+
+
+class RandomizedPacer:
+    """
+    Maintains a consistent average request rate while randomizing exact request timing
+    to avoid coordination between multiple instances.
+    """
+
+    def __init__(self, target_rps):
+        """
+        Initialize pacing controller with target requests per second.
+        """
+        self.target_rps = target_rps
+        self.next_expected_time = time.time()
+
+    def wait_time(self):
+        """
+        Calculate wait time for next request, introducing randomization while
+        maintaining the target throughput over time.
+
+        Returns:
+            float: Time to wait in seconds before next request
+        """
+        now = time.time()
+
+        # Calculate ideal time between requests
+        interval = 1.0 / self.target_rps if self.target_rps > 0 else 0
+
+        # If we're behind schedule, don't wait
+        if now >= self.next_expected_time:
+            # Schedule the next request one interval from now
+            self.next_expected_time = now + interval
+            return 0
+
+        # Calculate wait time to next expected request (assuming exact target
+        # rate)
+        wait = self.next_expected_time - now
+
+        # Introduce randomization: instead of waiting the full time,
+        # we'll wait a random amount between 0.5x - 1.5x of wait - i.e. on
+        # average we wait for `wait` seconds, thus maintaining the target rate
+        # over time.
+        # (0.5x - 1.5x is chosen instead of say 0 - 2.0x to reduce the variance
+        # in the request rate, particulary for small test durations / number of
+        # users - this should still be sufficiently uncorrelated for typical
+        # workloads).
+        random_factor = 0.5 + random.random()
+        randomized_wait = wait * random_factor
+
+        # Schedule the next expected request time:
+        # 1. Move forward by one full interval
+        # 2. Adjust for the randomization to maintain the average rate
+        adjustment = wait - randomized_wait
+        self.next_expected_time = self.next_expected_time + interval - adjustment
+
+        return max(randomized_wait, 0)
diff --git a/vsb/users.py b/vsb/users.py
index 09491e3..24ee12b 100644
--- a/vsb/users.py
+++ b/vsb/users.py
@@ -15,6 +15,7 @@
 from vsb import metrics, metrics_tracker
 from vsb.databases import DB
 import vsb.metrics_tracker
+from vsb.randomized_pacer import RandomizedPacer
 from vsb.vsb_types import (
     RecordList,
     SearchRequest,
@@ -243,9 +244,10 @@ def __init__(self, environment):
         self.workload = environment.workload_sequence[iteration]
         self.state = RunUser.State.Active
         opts = environment.parsed_options
-        self.target_throughput = opts.requests_per_sec / float(opts.num_users)
+        target_throughput = opts.requests_per_sec / float(opts.num_users)
+        self.pacer = RandomizedPacer(target_throughput)
         logger.debug(
-            f"Initialising RunUser id:{self.user_id}, target request/sec:{self.target_throughput}"
+            f"Initialising RunUser id:{self.user_id}, target request/sec:{target_throughput}"
         )
         self.query_iter = None
 
@@ -263,9 +265,7 @@ def wait_time(self):
         """Method called by locust to control how long this task should wait between
         executions.
         """
-        if self.target_throughput > 0:
-            return constant_throughput(self.target_throughput)(self)
-        return 0
+        return self.pacer.wait_time()
 
     def do_run(self):
         if not self.query_iter:


Test patch:
diff --git a/tests/unit/test_randomized_pacer.py b/tests/unit/test_randomized_pacer.py
new file mode 100644
index 0000000..5faff6a
--- /dev/null
+++ b/tests/unit/test_randomized_pacer.py
@@ -0,0 +1,117 @@
+import os
+import sys
+from unittest.mock import patch
+import statistics
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
+
+from vsb.randomized_pacer import RandomizedPacer
+import pytest
+
+
+class TestRandomizedPacer:
+    def test_init(self):
+        pacer = RandomizedPacer(1.0)
+        assert pacer.target_rps == 1.0
+
+    def test_zero_rps(self):
+        # When target_rps is 0, wait_time should return 0
+        pacer = RandomizedPacer(0.0)
+        assert pacer.wait_time() == 0
+
+    @patch("time.time")
+    def test_behind_schedule(self, mock_time):
+        # Test when we're behind schedule
+        mock_time.return_value = 100.0
+        pacer = RandomizedPacer(1.0)
+
+        # Now we're 2 seconds behind
+        mock_time.return_value = 102.0
+
+        with patch("random.random", return_value=0.5):
+            wait = pacer.wait_time()
+            assert wait == 0.0  # Should return 0 when behind schedule
+
+    def test_long_term_rate_stability(self):
+        # Test that over many calls, the average rate is maintained
+        # This test mocks time to make it deterministic
+        with patch("time.time") as mock_time:
+            current_time = 1000.0
+            mock_time.return_value = current_time
+
+            pacer = RandomizedPacer(10.0)
+            wait_times = []
+
+            # Simulate many requests with mocked time
+            samples = 1000  # Large sample for statistical confidence
+            for _ in range(samples):
+                wait = pacer.wait_time()
+                wait_times.append(wait)
+                current_time += wait
+                mock_time.return_value = current_time
+
+            # Calculate effective rate
+            total_time = sum(wait_times)
+            effective_rps = samples / total_time if total_time > 0 else float("inf")
+
+            # Rate should be close to target (allowing for some variation)
+            assert (
+                9.5 <= effective_rps <= 10.5
+            ), f"Effective rate {effective_rps} is too far from target 10.0"
+
+    def test_wait_time_distribution(self):
+        """
+        Test that wait times are properly distributed and maintain the target rate.
+
+        This test validates:
+        1. Wait times are properly randomized (not all the same)
+        2. The overall request rate matches the target rate
+        3. The distribution has expected statistical properties
+        """
+        with patch("time.time") as mock_time:
+            current_time = 1000.0
+            mock_time.return_value = current_time
+
+            # Create a pacer with 1 request per second
+            pacer = RandomizedPacer(1.0)
+
+            # Collect wait times and timestamps
+            wait_times = []
+            timestamps = [current_time]  # Start with initial time
+
+            for _ in range(1000):
+                wait = pacer.wait_time()
+                wait_times.append(wait)
+
+                # Simulate time passing exactly by the wait duration
+                current_time += wait
+                mock_time.return_value = current_time
+                timestamps.append(current_time)
+
+        # Verify wait times are non-negative
+        assert min(wait_times) >= 0, "Wait times should not be negative"
+
+        # Verify wait times are randomized (have variance)
+        assert statistics.stdev(wait_times) > 0.05, "Wait times should be distributed"
+
+        # Verify the mean wait time is close to the expected interval
+        # With 1000 samples, it should be very close to the target interval of 1 second
+        mean_wait = statistics.mean(wait_times)
+        assert (
+            0.95 <= mean_wait <= 1.05
+        ), f"Mean wait time {mean_wait} should be close to 1.0 second"
+
+        # Verify the request rate over the entire period matches the target
+        total_time = timestamps[-1] - timestamps[0]
+        req_per_sec = 1000 / total_time
+        assert (
+            0.95 <= req_per_sec <= 1.05
+        ), f"Request rate {req_per_sec} should be close to 1.0"
+
+        # Verify successive wait times aren't too strongly correlated
+        # (which would indicate poor randomization)
+        successive_diffs = [
+            abs(wait_times[i] - wait_times[i - 1]) for i in range(1, len(wait_times))
+        ]
+        avg_diff = statistics.mean(successive_diffs)
+        assert avg_diff > 0.1, "Wait times should vary between successive requests"