Implement worker group and worker tests, fix any bugs/issues that popped up

Author: markurtz

pyproject.toml

@@ -46,6 +46,7 @@ dependencies = [
     "click>=8.0.0,<8.2.0",
     "culsans~=0.9.0",
     "datasets",
+    "eval_type_backport",
     "ftfy>=6.0.0",
     "httpx[http2]<1.0.0",
     "loguru",

src/guidellm/benchmark/entrypoints.py

@@ -160,7 +160,12 @@ async def benchmark_generative_text(
         instances=progress or [], enabled=progress is not None
     )
 
-    async for aggregator_update, benchmark, strategy, scheduler_state in progress_group(
+    async for (
+        _aggregator_update,
+        benchmark,
+        _strategy,
+        _scheduler_state,
+    ) in progress_group(
         profile,
         Benchmarker[
             GenerativeBenchmark,

src/guidellm/config.py

@@ -134,6 +134,7 @@ class Settings(BaseSettings):
     max_worker_processes: int = 10
     max_add_requests_per_loop: int = 20
     scheduler_start_delay_non_distributed: float = 0.1
+    scheduler_poll_interval: float = 0.05
 
     # Data settings
     dataset: DatasetSettings = DatasetSettings()

src/guidellm/scheduler/__init__.py

@@ -19,6 +19,7 @@
     BackendT,
     MeasuredRequestTimings,
     MeasuredRequestTimingsT,
+    MultiTurnRequestT,
     RequestSchedulerTimings,
     RequestT,
     ResponseT,
@@ -70,6 +71,7 @@
     "MaxNumberConstraintInitializer",
     "MeasuredRequestTimings",
     "MeasuredRequestTimingsT",
+    "MultiTurnRequestT",
     "NoDelayRequestTimings",
     "NonDistributedEnvironment",
     "PoissonRateRequestTimings",

src/guidellm/scheduler/constraints.py

@@ -22,8 +22,10 @@
     ConstraintInitializer: Function signature for constraint factory.
 """
 
+from __future__ import annotations
+
 import time
-from typing import Any, Optional, Protocol, Union, runtime_checkable
+from typing import Any, Protocol, runtime_checkable
 
 from pydantic import Field
 
@@ -32,6 +34,7 @@
     ScheduledRequestInfo,
     SchedulerState,
     SchedulerUpdateAction,
+    SchedulerUpdateActionProgress,
 )
 from guidellm.utils import RegistryMixin
 
@@ -127,7 +130,7 @@ def resolve(
         cls,
         initializers: dict[
             str,
-            Union[Any, dict[str, Any], Constraint, ConstraintInitializer],
+            Any | dict[str, Any] | Constraint | ConstraintInitializer,
         ],
     ) -> dict[str, Constraint]:
         """
@@ -154,7 +157,7 @@ def resolve(
     @classmethod
     def resolve_constraints(
         cls,
-        constraints: dict[str, Union[Any, dict[str, Any], Constraint]],
+        constraints: dict[str, Any | dict[str, Any] | Constraint],
     ) -> dict[str, Constraint]:
         """
         Resolve constraints from mixed constraint specifications.
@@ -177,9 +180,7 @@ def resolve_constraints(
 
 
 class _MaxNumberBase(StandardBaseModel):
-    max_num: Union[int, float] = Field(
-        gt=0, description="Maximum number of requests allowed"
-    )
+    max_num: int | float = Field(gt=0, description="Maximum number of requests allowed")
 
 
 class MaxNumberConstraint(_MaxNumberBase):
@@ -208,12 +209,12 @@ def __call__(
                 "created_requests": state.created_requests,
                 "processed_requests": state.processed_requests,
             },
-            progress={
-                "remaining_fraction": max(
+            progress=SchedulerUpdateActionProgress(
+                remaining_fraction=max(
                     0.0, 1.0 - state.processed_requests / float(self.max_num)
                 ),
-                "remaining_requests": max(0, self.max_num - state.processed_requests),
-            },
+                remaining_requests=max(0, self.max_num - state.processed_requests),
+            ),
         )
 
 
@@ -222,9 +223,7 @@ class MaxNumberConstraintInitializer(_MaxNumberBase):
     """Factory for creating MaxNumberConstraint instances."""
 
     @classmethod
-    def from_simple_value(
-        cls, value: Union[int, float]
-    ) -> "MaxNumberConstraintInitializer":
+    def from_simple_value(cls, value: int | float) -> MaxNumberConstraintInitializer:
         """
         Create a MaxNumberConstraintInitializer from a simple scalar value.
 
@@ -246,9 +245,7 @@ def create_constraint(self, **_kwargs) -> Constraint:
 
 
 class _MaxDurationBase(StandardBaseModel):
-    max_duration: Union[int, float] = Field(
-        gt=0, description="Maximum duration in seconds"
-    )
+    max_duration: int | float = Field(gt=0, description="Maximum duration in seconds")
 
 
 class MaxDurationConstraint(_MaxDurationBase):
@@ -278,12 +275,10 @@ def __call__(
                 "start_time": state.start_time,
                 "current_time": current_time,
             },
-            progress={
-                "remaining_fraction": max(
-                    0.0, 1.0 - elapsed / float(self.max_duration)
-                ),
-                "remaining_duration": max(0.0, self.max_duration - elapsed),
-            },
+            progress=SchedulerUpdateActionProgress(
+                remaining_fraction=max(0.0, 1.0 - elapsed / float(self.max_duration)),
+                remaining_duration=max(0.0, self.max_duration - elapsed),
+            ),
         )
 
 
@@ -292,9 +287,7 @@ class MaxDurationConstraintInitializer(_MaxDurationBase):
     """Factory for creating MaxDurationConstraint instances."""
 
     @classmethod
-    def from_simple_value(
-        cls, value: Union[int, float]
-    ) -> "MaxDurationConstraintInitializer":
+    def from_simple_value(cls, value: int | float) -> MaxDurationConstraintInitializer:
         """
         Create a MaxDurationConstraintInitializer from a simple scalar value.
 
@@ -316,7 +309,7 @@ def create_constraint(self, **_kwargs) -> Constraint:
 
 
 class _MaxErrorsBase(StandardBaseModel):
-    max_errors: Union[int, float] = Field(
+    max_errors: int | float = Field(
         gt=0, description="Maximum number of errors allowed"
     )
 
@@ -352,9 +345,7 @@ class MaxErrorsConstraintInitializer(_MaxErrorsBase):
     """Factory for creating MaxErrorsConstraint instances."""
 
     @classmethod
-    def from_simple_value(
-        cls, value: Union[int, float]
-    ) -> "MaxErrorsConstraintInitializer":
+    def from_simple_value(cls, value: int | float) -> MaxErrorsConstraintInitializer:
         """
         Create a MaxErrorsConstraintInitializer from a simple scalar value.
 
@@ -376,10 +367,10 @@ def create_constraint(self, **_kwargs) -> Constraint:
 
 
 class _MaxErrorRateBase(StandardBaseModel):
-    max_error_rate: Union[int, float] = Field(
+    max_error_rate: int | float = Field(
         gt=0, le=1, description="Maximum error rate allowed (0.0 to 1.0)"
     )
-    window_size: Union[int, float] = Field(
+    window_size: int | float = Field(
         default=50,
         gt=0,
         description="Size of sliding window for calculating error rate",
@@ -444,9 +435,7 @@ class MaxErrorRateConstraintInitializer(_MaxErrorRateBase):
     """Factory for creating MaxErrorRateConstraint instances."""
 
     @classmethod
-    def from_simple_value(
-        cls, value: Union[int, float]
-    ) -> "MaxErrorRateConstraintInitializer":
+    def from_simple_value(cls, value: int | float) -> MaxErrorRateConstraintInitializer:
         """
         Create a MaxErrorRateConstraintInitializer from a simple scalar value.
 
@@ -469,10 +458,10 @@ def create_constraint(self, **_kwargs) -> Constraint:
 
 
 class _MaxGlobalErrorRateBase(StandardBaseModel):
-    max_error_rate: Union[int, float] = Field(
+    max_error_rate: int | float = Field(
         gt=0, le=1, description="Maximum error rate allowed (0.0 to 1.0)"
     )
-    min_processed: Optional[Union[int, float]] = Field(
+    min_processed: int | float | None = Field(
         default=50,
         gt=30,
         description=(
@@ -524,8 +513,8 @@ class MaxGlobalErrorRateConstraintInitializer(_MaxGlobalErrorRateBase):
 
     @classmethod
     def from_simple_value(
-        cls, value: Union[int, float]
-    ) -> "MaxGlobalErrorRateConstraintInitializer":
+        cls, value: int | float
+    ) -> MaxGlobalErrorRateConstraintInitializer:
         """
         Create a MaxGlobalErrorRateConstraintInitializer from a simple scalar value.
 

src/guidellm/scheduler/environment.py

@@ -10,12 +10,13 @@
     NonDistributedEnvironment: Single-node implementation with minimal overhead.
 """
 
+from __future__ import annotations
+
 import time
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterator, Iterable
 from typing import (
     Generic,
-    Optional,
 )
 
 from guidellm.config import settings
@@ -82,7 +83,7 @@ async def sync_run_start(self) -> float:
     @abstractmethod
     async def update_run_iteration(
         self,
-        response: Optional[ResponseT],
+        response: ResponseT | None,
         request: RequestT,
         request_info: ScheduledRequestInfo[MeasuredRequestTimingsT],
     ):
@@ -177,7 +178,7 @@ async def sync_run_start(self) -> float:
 
     async def update_run_iteration(
         self,
-        response: Optional[ResponseT],
+        response: ResponseT | None,
         request: RequestT,
         request_info: ScheduledRequestInfo[MeasuredRequestTimingsT],
     ):