feat: add --fixed-schedule-speedup parameter for scaling loadgen

# Conflicts:
#	aiperf/timing/config.py

Author: ajcasagrande

aiperf/common/config/config_defaults.py

@@ -47,6 +47,7 @@ class InputDefaults:
     FIXED_SCHEDULE_AUTO_OFFSET = False
     FIXED_SCHEDULE_START_OFFSET = None
     FIXED_SCHEDULE_END_OFFSET = None
+    FIXED_SCHEDULE_SPEEDUP = None
     PUBLIC_DATASET = None
     CUSTOM_DATASET_TYPE = None
     RANDOM_SEED = None

aiperf/common/config/input_config.py

@@ -178,6 +178,21 @@ def validate_dataset_type(self) -> Self:
         ),
     ] = InputDefaults.FIXED_SCHEDULE_END_OFFSET
 
+    # NEW AIPerf Option
+    fixed_schedule_speedup: Annotated[
+        float | None,
+        Field(
+            gt=0,
+            description="Specifies a scaling factor to apply to the timestamps in the fixed schedule. "
+            "For example, a value of 2.0 will make the schedule run twice as fast, while a value of 0.5 "
+            "will make it run half as fast.",
+        ),
+        CLIParameter(
+            name=("--fixed-schedule-speedup",),
+            group=_CLI_GROUP,
+        ),
+    ] = InputDefaults.FIXED_SCHEDULE_SPEEDUP
+
     public_dataset: Annotated[
         PublicDatasetType | None,
         Field(description="The public dataset to use for the requests."),

aiperf/timing/config.py

@@ -25,6 +25,7 @@ class TimingManagerConfig(AIPerfBaseModel):
     auto_offset_timestamps: bool = InputDefaults.FIXED_SCHEDULE_AUTO_OFFSET
     fixed_schedule_start_offset: int | None = InputDefaults.FIXED_SCHEDULE_START_OFFSET
     fixed_schedule_end_offset: int | None = InputDefaults.FIXED_SCHEDULE_END_OFFSET
+    fixed_schedule_speedup: float | None = InputDefaults.FIXED_SCHEDULE_SPEEDUP
     request_cancellation_rate: float = LoadGeneratorDefaults.REQUEST_CANCELLATION_RATE
     request_cancellation_delay: float = LoadGeneratorDefaults.REQUEST_CANCELLATION_DELAY
 
@@ -45,6 +46,7 @@ def from_user_config(cls, user_config: UserConfig) -> "TimingManagerConfig":
             auto_offset_timestamps=user_config.input.fixed_schedule_auto_offset,
             fixed_schedule_start_offset=user_config.input.fixed_schedule_start_offset,
             fixed_schedule_end_offset=user_config.input.fixed_schedule_end_offset,
+            fixed_schedule_speedup=user_config.input.fixed_schedule_speedup,
             request_cancellation_rate=user_config.loadgen.request_cancellation_rate,
             request_cancellation_delay=user_config.loadgen.request_cancellation_delay,
         )

aiperf/timing/fixed_schedule_strategy.py

@@ -35,6 +35,7 @@ def __init__(
         self._auto_offset_timestamps = config.auto_offset_timestamps
         self._start_offset = config.fixed_schedule_start_offset
         self._end_offset = config.fixed_schedule_end_offset
+        self._time_scale = 1 / (config.fixed_schedule_speedup or 1.0)
         super().__init__(config=config, credit_manager=credit_manager)
 
     def _create_timestamp_groups(self) -> None:
@@ -90,8 +91,10 @@ async def _execute_single_phase(self, phase_stats: CreditPhaseStats) -> None:
 
             # Calculate the wait duration for this timestamp
             # (timestamp - schedule_zero_ms) is the offset of the conversation(s) from the start of the schedule
+            # Apply time scaling to the offset, then subtract elapsed time
             # (self._perf_counter_ms() - start_time_ms) is how much time has passed since we started dropping credits
-            wait_duration_ms = (timestamp - self._schedule_zero_ms) - (
+            scaled_offset_ms = (timestamp - self._schedule_zero_ms) * self._time_scale
+            wait_duration_ms = scaled_offset_ms - (
                 self._perf_counter_ms() - start_time_ms
             )
             wait_duration_sec = wait_duration_ms / MILLIS_PER_SECOND

tests/timing_manager/test_fixed_schedule_strategy.py

@@ -39,12 +39,14 @@ def _create_strategy(
         schedule: list[tuple[int, str]],
         auto_offset: bool = False,
         manual_offset: int | None = None,
+        speedup: float | None = None,
     ) -> tuple[FixedScheduleStrategy, CreditPhaseStats]:
         """Helper to create a strategy with optional config overrides."""
         config = TimingManagerConfig.model_construct(
             timing_mode=TimingMode.FIXED_SCHEDULE,
             auto_offset_timestamps=auto_offset,
             fixed_schedule_start_offset=manual_offset,
+            fixed_schedule_speedup=speedup,
         )
         return FixedScheduleStrategy(
             config=config,
@@ -194,3 +196,165 @@ async def test_execution_with_auto_offset(
         assert phase_stats.sent == 3
         expected_zero_ms = first_timestamp_ms if auto_offset else 0
         assert strategy._schedule_zero_ms == expected_zero_ms
+
+    @pytest.mark.parametrize(
+        "speedup,expected_time_scale",
+        [
+            (None, 1.0),   # Default behavior (no speedup)
+            (1.0, 1.0),    # No speedup
+            (2.0, 0.5),    # 2x faster
+            (0.5, 2.0),    # 2x slower
+            (10.0, 0.1),   # 10x faster
+            (0.1, 10.0),   # 10x slower
+        ],
+    )  # fmt: skip
+    def test_speedup_time_scale_calculation(
+        self,
+        simple_schedule: list[tuple[int, str]],
+        mock_credit_manager: MockCreditManager,
+        speedup: float | None,
+        expected_time_scale: float,
+    ):
+        """Test that speedup parameter correctly calculates time scale."""
+        strategy, _ = self._create_strategy(
+            mock_credit_manager, simple_schedule, speedup=speedup
+        )
+
+        assert strategy._time_scale == expected_time_scale
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "speedup,schedule",
+        [
+            # 2x faster - should take half the time
+            (2.0, [(0, "conv1"), (100, "conv2"), (200, "conv3")]),
+            # 4x faster - should take quarter the time
+            (4.0, [(0, "conv1"), (100, "conv2"), (200, "conv3")]),
+            # 2x slower - should take double the time
+            (0.5, [(0, "conv1"), (100, "conv2"), (200, "conv3")]),
+            # Different schedule with larger gaps
+            (2.0, [(0, "conv1"), (500, "conv2"), (1000, "conv3")]),
+            # Edge case: all at same timestamp should still be instant
+            (2.0, [(0, "conv1"), (0, "conv2"), (0, "conv3")]),
+            (0.5, [(0, "conv1"), (0, "conv2"), (0, "conv3")]),
+        ],
+    )  # fmt: skip
+    async def test_speedup_execution_timing(
+        self,
+        mock_credit_manager: MockCreditManager,
+        time_traveler: TimeTraveler,
+        speedup: float,
+        schedule: list[tuple[int, str]],
+    ):
+        """Test that speedup parameter affects actual execution timing."""
+        strategy, phase_stats = self._create_strategy(
+            mock_credit_manager, schedule, speedup=speedup
+        )
+
+        # Calculate expected duration: (last_timestamp - first_timestamp) / 1000 / speedup
+        # Since auto_offset is default True, we use the relative duration
+        first_timestamp_ms = schedule[0][0]
+        last_timestamp_ms = schedule[-1][0]
+        base_duration_sec = (last_timestamp_ms - first_timestamp_ms) / MILLIS_PER_SECOND
+        expected_duration = base_duration_sec / speedup
+
+        with time_traveler.sleeps_for(expected_duration):
+            await strategy._execute_single_phase(phase_stats)
+            await strategy.wait_for_tasks()
+
+        assert phase_stats.sent == len(schedule)
+        assert len(mock_credit_manager.dropped_credits) == len(schedule)
+
+    @pytest.mark.asyncio
+    async def test_speedup_with_auto_offset(
+        self,
+        mock_credit_manager: MockCreditManager,
+        time_traveler: TimeTraveler,
+    ):
+        """Test speedup works correctly with auto offset timestamps."""
+        # Schedule starts at 1000ms with 200ms total duration
+        schedule = [(1000, "conv1"), (1100, "conv2"), (1200, "conv3")]
+        speedup = 2.0  # 2x faster
+
+        strategy, phase_stats = self._create_strategy(
+            mock_credit_manager, schedule, auto_offset=True, speedup=speedup
+        )
+
+        # With auto offset, the duration should be (1200-1000)ms = 200ms
+        # At 2x speed: 200ms / 2 = 100ms = 0.1s
+        base_duration_sec = (1200 - 1000) / MILLIS_PER_SECOND  # 0.2s
+        expected_duration = base_duration_sec / speedup  # 0.1s
+
+        with time_traveler.sleeps_for(expected_duration):
+            await strategy._execute_single_phase(phase_stats)
+            await strategy.wait_for_tasks()
+
+        assert phase_stats.sent == 3
+        assert strategy._schedule_zero_ms == 1000  # First timestamp
+
+    @pytest.mark.asyncio
+    async def test_speedup_with_manual_offset(
+        self,
+        mock_credit_manager: MockCreditManager,
+        time_traveler: TimeTraveler,
+    ):
+        """Test speedup works correctly with manual offset."""
+        schedule = [(1000, "conv1"), (1100, "conv2"), (1200, "conv3")]
+        manual_offset = 500
+        speedup = 0.5  # 2x slower
+
+        strategy, phase_stats = self._create_strategy(
+            mock_credit_manager, schedule, manual_offset=manual_offset, speedup=speedup
+        )
+
+        # With manual offset of 500, effective duration is (1200-500) = 700ms
+        # At 0.5x speed: 700ms / 0.5 = 1400ms = 1.4s
+        base_duration_sec = (1200 - manual_offset) / MILLIS_PER_SECOND  # 0.7s
+        expected_duration = base_duration_sec / speedup  # 1.4s
+
+        with time_traveler.sleeps_for(expected_duration):
+            await strategy._execute_single_phase(phase_stats)
+            await strategy.wait_for_tasks()
+
+        assert phase_stats.sent == 3
+        assert strategy._schedule_zero_ms == manual_offset
+
+    @pytest.mark.asyncio
+    async def test_speedup_with_negative_timestamps(
+        self,
+        mock_credit_manager: MockCreditManager,
+        time_traveler: TimeTraveler,
+    ):
+        """Test speedup behavior with negative timestamps (past events)."""
+        # All timestamps are in the past, should execute immediately
+        schedule = [(-100, "conv1"), (-50, "conv2"), (0, "conv3")]
+        speedup = 2.0  # Even with speedup, past events should be immediate
+
+        strategy, phase_stats = self._create_strategy(
+            mock_credit_manager, schedule, speedup=speedup
+        )
+
+        # Should still take no time since all timestamps are <= 0
+        with time_traveler.sleeps_for(0.0):
+            await strategy._execute_single_phase(phase_stats)
+            await strategy.wait_for_tasks()
+
+        assert phase_stats.sent == 3
+
+    def test_speedup_edge_cases(
+        self,
+        simple_schedule: list[tuple[int, str]],
+        mock_credit_manager: MockCreditManager,
+    ):
+        """Test edge cases for speedup parameter."""
+        # Test very small speedup (very slow execution)
+        strategy_slow, _ = self._create_strategy(
+            mock_credit_manager, simple_schedule, speedup=0.001
+        )
+        assert strategy_slow._time_scale == 1000.0
+
+        # Test very large speedup (very fast execution)
+        strategy_fast, _ = self._create_strategy(
+            mock_credit_manager, simple_schedule, speedup=1000.0
+        )
+        assert strategy_fast._time_scale == 0.001