Tests and fixes for WorkerProcess and supporting changes

Author: markurtz

pyproject.toml

@@ -141,6 +141,7 @@ ignore_missing_imports=true
 
 
 [tool.ruff]
+target-version = "py39"
 line-length = 88
 indent-width = 4
 exclude = ["build", "dist", "env", ".venv"]

src/guidellm/backend/interface.py

@@ -3,23 +3,19 @@
 from typing import (
     Any,
     Generic,
-    Literal,
     Optional,
     TypeVar,
 )
 
-from pydantic import Field
-
-from guidellm.objects import StandardBaseModel
 from guidellm.scheduler import (
+    MeasuredRequestTimingsT,
     RequestT,
-    RequestTimingsT,
     ResponseT,
     ScheduledRequestInfo,
 )
 
 
-class BackendInterface(ABC, Generic[RequestT, RequestTimingsT, ResponseT]):
+class BackendInterface(ABC, Generic[RequestT, MeasuredRequestTimingsT, ResponseT]):
     """
     Abstract interface for request processing backends. Note: before process_startup
     is invoked, the implementation must ensure all properties are pickleable.
@@ -71,9 +67,9 @@ async def process_shutdown(self) -> None:
     async def resolve(
         self,
         request: RequestT,
-        request_info: ScheduledRequestInfo[RequestTimingsT],
+        request_info: ScheduledRequestInfo[MeasuredRequestTimingsT],
         history: Optional[list[tuple[RequestT, ResponseT]]] = None,
-    ) -> AsyncIterator[tuple[ResponseT, ScheduledRequestInfo[RequestTimingsT]]]:
+    ) -> AsyncIterator[tuple[ResponseT, ScheduledRequestInfo[MeasuredRequestTimingsT]]]:
         """
         Process a request and yield incremental response updates.
 

src/guidellm/backend/objects.py

@@ -12,7 +12,7 @@
 from pydantic import Field
 
 from guidellm.objects.pydantic import StandardBaseModel
-from guidellm.scheduler import RequestTimings
+from guidellm.scheduler import MeasuredRequestTimings
 
 __all__ = [
     "GenerationRequest",
@@ -135,7 +135,7 @@ def preferred_output_tokens(
             return self.response_output_tokens or self.request_output_tokens
 
 
-class GenerationRequestTimings(RequestTimings):
+class GenerationRequestTimings(MeasuredRequestTimings):
     """Timing model for tracking generation request lifecycle events."""
 
     first_iteration: Optional[float] = Field(

src/guidellm/benchmark/aggregator.py

@@ -49,8 +49,8 @@
     StatusDistributionSummary,
 )
 from guidellm.scheduler import (
+    MeasuredRequestTimingsT,
     RequestT,
-    RequestTimingsT,
     ResponseT,
     ScheduledRequestInfo,
     SchedulerState,
@@ -67,7 +67,7 @@
 
 
 @runtime_checkable
-class Aggregator(Protocol[ResponseT, RequestT, RequestTimingsT]):
+class Aggregator(Protocol[ResponseT, RequestT, MeasuredRequestTimingsT]):
     """
     Protocol for processing benchmark data updates during execution.
 
@@ -81,7 +81,7 @@ def __call__(
         agg_state: dict[str, Any],
         response: Optional[ResponseT],
         request: RequestT,
-        request_info: ScheduledRequestInfo[RequestTimingsT],
+        request_info: ScheduledRequestInfo[MeasuredRequestTimingsT],
         scheduler_state: SchedulerState,
     ) -> Optional[dict[str, Any]]:
         """
@@ -98,7 +98,7 @@ def __call__(
 
 
 @runtime_checkable
-class CompilableAggregator(Aggregator[ResponseT, RequestT, RequestTimingsT]):
+class CompilableAggregator(Aggregator[ResponseT, RequestT, MeasuredRequestTimingsT]):
     """
     Protocol for aggregators that compile final results from aggregated state.
 
@@ -146,7 +146,7 @@ def add_aggregate_metric(
 
 
 class SchedulerStatsAggregator(
-    CompilableAggregator[ResponseT, RequestT, RequestTimingsT]
+    CompilableAggregator[ResponseT, RequestT, MeasuredRequestTimingsT]
 ):
     """
     Aggregates scheduler timing and performance metrics.
@@ -160,7 +160,7 @@ def __call__(
         agg_state: dict[str, Any],
         response: Optional[ResponseT],
         request: RequestT,
-        request_info: ScheduledRequestInfo[RequestTimingsT],
+        request_info: ScheduledRequestInfo[MeasuredRequestTimingsT],
         scheduler_state: SchedulerState,
     ) -> Optional[dict[str, Any]]:
         """

src/guidellm/benchmark/benchmarker.py

@@ -32,8 +32,8 @@
     BackendT,
     Constraint,
     Environment,
+    MeasuredRequestTimingsT,
     RequestT,
-    RequestTimingsT,
     ResponseT,
     Scheduler,
     SchedulerState,
@@ -45,7 +45,7 @@
 
 
 class Benchmarker(
-    Generic[BenchmarkT, RequestT, RequestTimingsT, ResponseT],
+    Generic[BenchmarkT, RequestT, MeasuredRequestTimingsT, ResponseT],
     ABC,
     ThreadSafeSingletonMixin,
 ):
@@ -65,14 +65,14 @@ async def run(
         requests: Iterable[
             Union[RequestT, Iterable[Union[RequestT, tuple[RequestT, float]]]]
         ],
-        backend: BackendT[RequestT, RequestTimingsT, ResponseT],
+        backend: BackendT[RequestT, MeasuredRequestTimingsT, ResponseT],
         profile: Profile,
         environment: Environment,
         benchmark_aggregators: dict[
             str,
             Union[
-                Aggregator[ResponseT, RequestT, RequestTimingsT],
-                CompilableAggregator[ResponseT, RequestT, RequestTimingsT],
+                Aggregator[ResponseT, RequestT, MeasuredRequestTimingsT],
+                CompilableAggregator[ResponseT, RequestT, MeasuredRequestTimingsT],
             ],
         ],
         benchmark_class: type[BenchmarkT],
@@ -114,7 +114,9 @@ async def run(
                     request,
                     request_info,
                     scheduler_state,
-                ) in Scheduler[BackendT, RequestT, RequestTimingsT, ResponseT].run(
+                ) in Scheduler[
+                    BackendT, RequestT, MeasuredRequestTimingsT, ResponseT
+                ].run(
                     requests=requests,
                     backend=backend,
                     strategy=strategy,
@@ -161,13 +163,13 @@ def _compile_benchmark_kwargs(
         requests: Iterable[
             Union[RequestT, Iterable[Union[RequestT, tuple[RequestT, float]]]]
         ],
-        backend: BackendT[RequestT, RequestTimingsT, ResponseT],
+        backend: BackendT[RequestT, MeasuredRequestTimingsT, ResponseT],
         environment: Environment,
         aggregators: dict[
             str,
             Union[
-                Aggregator[ResponseT, RequestT, RequestTimingsT],
-                CompilableAggregator[ResponseT, RequestT, RequestTimingsT],
+                Aggregator[ResponseT, RequestT, MeasuredRequestTimingsT],
+                CompilableAggregator[ResponseT, RequestT, MeasuredRequestTimingsT],
             ],
         ],
         aggregators_state: dict[str, dict[str, Any]],

src/guidellm/scheduler/__init__.py

@@ -17,10 +17,10 @@
 from .objects import (
     BackendInterface,
     BackendT,
+    MeasuredRequestTimings,
+    MeasuredRequestTimingsT,
     RequestSchedulerTimings,
     RequestT,
-    RequestTimings,
-    RequestTimingsT,
     ResponseT,
     ScheduledRequestInfo,
     SchedulerState,
@@ -68,13 +68,13 @@
     "MaxGlobalErrorRateConstraintInitializer",
     "MaxNumberConstraint",
     "MaxNumberConstraintInitializer",
+    "MeasuredRequestTimings",
+    "MeasuredRequestTimingsT",
     "NoDelayRequestTimings",
     "NonDistributedEnvironment",
     "PoissonRateRequestTimings",
     "RequestSchedulerTimings",
     "RequestT",
-    "RequestTimings",
-    "RequestTimingsT",
     "ResponseT",
     "ScheduledRequestInfo",
     "ScheduledRequestTimings",

src/guidellm/scheduler/constraints.py

@@ -101,8 +101,11 @@ def create(cls, key: str, *args, **kwargs) -> ConstraintInitializer:
         initializer_class = cls.get_registered_object(key)
 
         # Handle simple scalar values by delegating to the initializer class
-        if (len(args) == 1 and not kwargs and
-            hasattr(initializer_class, "from_simple_value")):
+        if (
+            len(args) == 1
+            and not kwargs
+            and hasattr(initializer_class, "from_simple_value")
+        ):
             return initializer_class.from_simple_value(args[0])
 
         return initializer_class(*args, **kwargs)

src/guidellm/scheduler/environment.py

@@ -21,8 +21,8 @@
 from guidellm.config import settings
 from guidellm.scheduler.constraints import Constraint
 from guidellm.scheduler.objects import (
+    MeasuredRequestTimingsT,
     RequestT,
-    RequestTimingsT,
     ResponseT,
     ScheduledRequestInfo,
     SchedulerState,
@@ -84,7 +84,7 @@ async def update_run_iteration(
         self,
         response: Optional[ResponseT],
         request: RequestT,
-        request_info: ScheduledRequestInfo[RequestTimingsT],
+        request_info: ScheduledRequestInfo[MeasuredRequestTimingsT],
     ):
         """
         Update environment state with completed request iteration.
@@ -116,7 +116,10 @@ async def sync_run_end(
         self,
     ) -> AsyncIterator[
         tuple[
-            ResponseT, RequestT, ScheduledRequestInfo[RequestTimingsT], SchedulerState
+            ResponseT,
+            RequestT,
+            ScheduledRequestInfo[MeasuredRequestTimingsT],
+            SchedulerState,
         ]
     ]:
         """
@@ -176,7 +179,7 @@ async def update_run_iteration(
         self,
         response: Optional[ResponseT],
         request: RequestT,
-        request_info: ScheduledRequestInfo[RequestTimingsT],
+        request_info: ScheduledRequestInfo[MeasuredRequestTimingsT],
     ):
         """
         No-op for single-node execution.
@@ -198,7 +201,10 @@ async def sync_run_end(
         self,
     ) -> AsyncIterator[
         tuple[
-            ResponseT, RequestT, ScheduledRequestInfo[RequestTimingsT], SchedulerState
+            ResponseT,
+            RequestT,
+            ScheduledRequestInfo[MeasuredRequestTimingsT],
+            SchedulerState,
         ]
     ]:
         """

src/guidellm/scheduler/objects.py

@@ -28,19 +28,19 @@
     Optional,
     TypeVar,
 )
-from typing_extensions import TypedDict
 
-from pydantic import computed_field, Field
+from pydantic import Field, computed_field
+from typing_extensions import TypedDict
 
 from guidellm.objects import StandardBaseModel
 
 __all__ = [
     "BackendInterface",
     "BackendT",
+    "MeasuredRequestTimings",
+    "MeasuredRequestTimingsT",
     "RequestSchedulerTimings",
     "RequestT",
-    "RequestTimings",
-    "RequestTimingsT",
     "ResponseT",
     "ScheduledRequestInfo",
     "SchedulerState",
@@ -82,7 +82,7 @@ class RequestSchedulerTimings(StandardBaseModel):
     )
 
 
-class RequestTimings(StandardBaseModel):
+class MeasuredRequestTimings(StandardBaseModel):
     """Base timing measurements for backend request processing."""
 
     request_start: Optional[float] = Field(
@@ -93,10 +93,12 @@ class RequestTimings(StandardBaseModel):
     )
 
 
-RequestTimingsT = TypeVar("RequestTimingsT", bound=RequestTimings)
+MeasuredRequestTimingsT = TypeVar(
+    "MeasuredRequestTimingsT", bound=MeasuredRequestTimings
+)
 
 
-class ScheduledRequestInfo(StandardBaseModel, Generic[RequestTimingsT]):
+class ScheduledRequestInfo(StandardBaseModel, Generic[MeasuredRequestTimingsT]):
     """Complete request information including status, timings, and metadata."""
 
     request_id: str = Field(description="Unique identifier for the request")
@@ -120,7 +122,7 @@ class ScheduledRequestInfo(StandardBaseModel, Generic[RequestTimingsT]):
         default_factory=RequestSchedulerTimings,
         description="Scheduler-level timing measurements for request lifecycle",
     )
-    request_timings: Optional[RequestTimingsT] = Field(
+    request_timings: Optional[MeasuredRequestTimingsT] = Field(
         default=None,
         description="Backend-specific timing measurements for request processing",
     )
@@ -152,7 +154,7 @@ def completed_at(self) -> Optional[float]:
         return request_end or self.scheduler_timings.resolve_end
 
 
-class BackendInterface(ABC, Generic[RequestT, RequestTimingsT, ResponseT]):
+class BackendInterface(ABC, Generic[RequestT, MeasuredRequestTimingsT, ResponseT]):
     """
     Abstract interface for request processing backends. Note: before process_startup
     is invoked, the implementation must ensure all properties are pickleable.
@@ -204,9 +206,9 @@ async def process_shutdown(self) -> None:
     async def resolve(
         self,
         request: RequestT,
-        request_info: ScheduledRequestInfo[RequestTimingsT],
+        request_info: ScheduledRequestInfo[MeasuredRequestTimingsT],
         history: Optional[list[tuple[RequestT, ResponseT]]] = None,
-    ) -> AsyncIterator[tuple[ResponseT, ScheduledRequestInfo[RequestTimingsT]]]:
+    ) -> AsyncIterator[tuple[ResponseT, ScheduledRequestInfo[MeasuredRequestTimingsT]]]:
         """
         Process a request and yield incremental response updates.