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.

Discussion:

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"