Lint fixes

Author: markVaykhansky

src/guidellm/__main__.py

@@ -168,7 +168,8 @@ def cli():
     type=float,
     help=(
         "The maximum error rate after which a benchmark will stop. "
-        "Applicable only for finite deterministic scenarios i.e rate_type is 'constant' and 'max_seconds' exists OR "
+        "Applicable only for finite deterministic scenarios i.e "
+        "rate_type is 'constant' and 'max_seconds' exists OR "
         "'max_requests' exists OR the dataset is finite. "
         "If None or not applicable, benchmarks will continue regardless of error rate."
     ),

src/guidellm/benchmark/aggregator.py

@@ -599,8 +599,8 @@ def compile(self) -> GenerativeBenchmark:
         and return the compiled object.
         """
         successful, incomplete, errored = self._compile_results()
-        error_rate = self.requests_stats.totals.errored.total / \
-            (self.requests_stats.totals.successful.total + self.requests_stats.totals.errored.total)
+
+        error_rate = self._calculate_error_rate()
 
         return GenerativeBenchmark.from_stats(
             run_id=self.run_id,
@@ -634,6 +634,12 @@ def compile(self) -> GenerativeBenchmark:
             extras=self.extras,
         )
 
+    def _calculate_error_rate(self) -> float:
+        total_successful = self.requests_stats.totals.successful.total
+        total_errored = self.requests_stats.totals.errored.total
+        total_sent = total_errored + total_successful
+        return total_errored / total_sent
+
     def _compile_results(
         self,
     ) -> tuple[

src/guidellm/benchmark/benchmark.py

@@ -218,8 +218,10 @@ class BenchmarkRunStats(StandardBaseModel):
     )
     error_rate: float = Field(
         description=(
-            "The number of errored requests divided by the number of errored requests. This can be higher "
-            "than max_error_rate (if applicable) cause it does not take into account incomplete requests."
+            "The number of errored requests divided by the number "
+            "of errored requests. This can be higher than max_error_rate "
+            "(if applicable) cause it does not take into "
+            "account incomplete requests."
         )
     )
 

src/guidellm/request/__init__.py

@@ -1,6 +1,7 @@
 from .loader import (
     GenerativeRequestLoader,
     GenerativeRequestLoaderDescription,
+    GetInfiniteDatasetLengthError,
     RequestLoader,
     RequestLoaderDescription,
 )
@@ -10,6 +11,7 @@
     "GenerationRequest",
     "GenerativeRequestLoader",
     "GenerativeRequestLoaderDescription",
+    "GetInfiniteDatasetLengthError",
     "RequestLoader",
     "RequestLoaderDescription",
 ]

src/guidellm/request/loader.py

@@ -19,13 +19,13 @@
 __all__ = [
     "GenerativeRequestLoader",
     "GenerativeRequestLoaderDescription",
+    "GetInfiniteDatasetLengthError",
     "RequestLoader",
     "RequestLoaderDescription",
-    "InfiniteDatasetError"
 ]
 
 
-class InfiniteDatasetError(Exception):
+class GetInfiniteDatasetLengthError(Exception):
     pass
 
 
@@ -125,8 +125,11 @@ def __len__(self) -> int:
         if self.iter_type == "finite":
             return self.num_unique_items()
 
-        assert self.iter_type == "infinite"
-        raise InfiniteDatasetError(f"Dataset {self.data} is infinite and thus unable to determine length")
+        if self.iter_type != "infinite":
+            raise ValueError(f"Invalid iter_type {self.iter_type}")
+        raise GetInfiniteDatasetLengthError(f"Dataset {self.data} is "
+                                            f"infinite and thus "
+                                            f"unable to determine length")
 
     @property
     def description(self) -> GenerativeRequestLoaderDescription:

src/guidellm/scheduler/scheduler.py

@@ -15,7 +15,7 @@
 from loguru import logger
 
 from guidellm.config import settings
-from guidellm.request.loader import InfiniteDatasetError
+from guidellm.request.loader import GetInfiniteDatasetLengthError
 from guidellm.scheduler.result import (
     SchedulerRequestResult,
     SchedulerResult,
@@ -101,24 +101,15 @@ async def run(
         :param max_duration: The maximum duration for the scheduling run.
             If None, then no limit is set and either the iterator must be exhaustible
             or the max_number must be set.
-        :param max_error_rate: The maximum error rate after which the scheduler shuts down.
+        :param max_error_rate: The maximum error rate after which the
+            scheduler shuts down.
             Only applicable in benchmarks with finite deterministic number of requests.
             If None or not applicable then scheduler will continue regardless of errors.
         :return: An asynchronous generator that yields SchedulerResult objects.
             Each SchedulerResult object contains information about the request,
             the response, and the run information.
         """
-        if scheduling_strategy is None or not isinstance(
-            scheduling_strategy, SchedulingStrategy
-        ):
-            raise ValueError(f"Invalid scheduling strategy: {scheduling_strategy}")
-
-        if max_number is not None and max_number < 1:
-            raise ValueError(f"Invalid max_number: {max_number}")
-        if max_duration is not None and max_duration < 0:
-            raise ValueError(f"Invalid max_duration: {max_duration}")
-        if max_error_rate is not None and (max_error_rate < 0 or max_error_rate > 1):
-            raise ValueError(f"Invalid max_error_rate: {max_error_rate}")
+        self._validate_scheduler_params(scheduling_strategy, max_duration, max_error_rate, max_number)
 
         with (
             multiprocessing.Manager() as manager,
@@ -127,11 +118,13 @@ async def run(
             ) as executor,
         ):
             requests_iter: Optional[Iterator[Any]] = None
-            futures, requests_queue, responses_queue, shutdown_event = await self._start_processes(
-                manager, executor, scheduling_strategy, max_error_rate is not None
-            )
-            if shutdown_event:
-                assert not shutdown_event.is_set(),  "shutdown_event is set before starting scheduling"
+            futures, requests_queue, responses_queue, shutdown_event = \
+                await self._start_processes(
+                    manager, executor, scheduling_strategy, max_error_rate is not None)
+            if shutdown_event and shutdown_event.is_set():
+                raise RuntimeError(
+                    "shutdown_event is set before starting scheduling"
+                )
             run_info, requests_iter, times_iter = self._run_setup(
                 futures, scheduling_strategy, max_number, max_duration, max_error_rate
             )
@@ -169,17 +162,14 @@ async def run(
                         run_info,
                     )
                     if iter_result is not None:
-                        if iter_result.request_info.errored and not iter_result.request_info.canceled:
-                            current_error_rate = run_info.errored_requests / run_info.end_number
-                            is_over_max_error_rate = run_info.max_error_rate < current_error_rate
-
-                            if is_over_max_error_rate:
-                                shutdown_event.set()
-                                max_error_rate_reached = True
-                                logger.info(f"Max error rate of ({iter_result.run_info.max_error_rate}) "
-                                            f"reached, sending shutdown signal")
-                            else:
-                                logger.debug(f"Current error rate: {current_error_rate}")
+                        if iter_result.request_info.errored \
+                            and not iter_result.request_info.canceled \
+                                and self._is_max_error_rate_reached(iter_result.run_info):
+                            shutdown_event.set()
+                            max_error_rate_reached = True
+                            logger.info(f"Max error rate of "
+                                        f"({iter_result.run_info.max_error_rate}) "
+                                        f"reached, sending shutdown signal")
                         yield iter_result
 
                     # yield control to the event loop
@@ -194,6 +184,28 @@ async def run(
 
             await self._stop_processes(futures, requests_queue)
 
+    def _validate_scheduler_params(
+            self,
+            scheduling_strategy: SchedulingStrategy,
+            max_duration: Optional[float],
+            max_error_rate: Optional[float],
+            max_number: Optional[int]
+    ) -> None:
+        if scheduling_strategy is None or not isinstance(
+                scheduling_strategy, SchedulingStrategy
+        ):
+            raise ValueError(f"Invalid scheduling strategy: {scheduling_strategy}")
+        if max_number is not None and max_number < 1:
+            raise ValueError(f"Invalid max_number: {max_number}")
+        if max_duration is not None and max_duration < 0:
+            raise ValueError(f"Invalid max_duration: {max_duration}")
+        if max_error_rate is not None and (max_error_rate < 0 or max_error_rate > 1):
+            raise ValueError(f"Invalid max_error_rate: {max_error_rate}")
+
+    def _is_max_error_rate_reached(self, run_info) -> bool:
+        current_error_rate = run_info.errored_requests / run_info.end_number
+        return run_info.max_error_rate < current_error_rate
+
     async def _start_processes(
         self,
         manager,
@@ -282,10 +294,13 @@ def _run_setup(
         start_time = time.time()
         times_iter = iter(scheduling_strategy.request_times())
         end_time = time.time() + (max_duration or math.inf)
-        end_number = self._determine_total_requests_count(scheduling_strategy, max_duration, max_number)
+        end_number = self._determine_total_requests_count(
+            scheduling_strategy, max_duration, max_number
+        )
 
         if end_number == math.inf and max_error_rate is not None:
-            logger.warning("max_error_rate will be ignored because end_number can not be determined.")
+            logger.warning("max_error_rate will be ignored "
+                           "because end_number can not be determined.")
 
         if end_number == math.inf and end_time is None:
             logger.warning(
@@ -312,17 +327,19 @@ def _determine_total_requests_count(
     ) -> int:
         end_number = max_number or math.inf
         try:
-            # update end number if the request loader is finite and less than max
+            # update end_number if the request_loader is finite and less than max_number
             iter_length = len(self.request_loader)  # type: ignore[arg-type]
             if 0 < iter_length < end_number:
                 end_number = iter_length
-        except InfiniteDatasetError:
-            # Only when RPS is constant and duration is capped we can determine the total
-            # amount of requests that are supposed to be sent
+        except GetInfiniteDatasetLengthError:
+            # Only when RPS is constant and duration is
+            # capped we can determine the total amount of requests
+            # that are supposed to be sent
             if scheduling_strategy.type_ == "constant" and max_duration is not None:
-                total_requests_in_max_duration = int(scheduling_strategy.rate * max_duration)
-                if total_requests_in_max_duration < end_number:
-                    assert total_requests_in_max_duration > 0
+                total_requests_in_max_duration = int(
+                    scheduling_strategy.rate * max_duration
+                )
+                if 0 < total_requests_in_max_duration < end_number:
                     end_number = total_requests_in_max_duration
         except Exception:  # noqa: BLE001, S110
             pass

src/guidellm/scheduler/worker.py

@@ -128,18 +128,26 @@ async def get_request(
             process_id: Optional[int] = None,
     ) -> Optional[WorkerProcessRequest[RequestT]]:
         if shutdown_event is not None and process_id is None:
-            logger.warning("shutdown_event is not None and process_id is None which makes it hard to debug")
+            logger.warning("shutdown_event is not None and process_id "
+                           "is None which makes it hard to debug")
 
         def _get_queue_intermittently():
-            assert shutdown_event is not None
+            if shutdown_event is None:
+                raise ValueError("Shouldn't use _get_queue_intermittently "
+                                 "if there's no shutdown_even")
             while True:
                 try:
-                    return requests_queue.get(timeout=timedelta(seconds=1).total_seconds())
+                    get_timeout = timedelta(seconds=1).total_seconds()
+                    return requests_queue.get(timeout=get_timeout)
                 except queue.Empty:
                     if shutdown_event.is_set():
                         logger.info(f"Shutdown signal received in future {process_id}")
-                        return
-        return await asyncio.to_thread(_get_queue_intermittently if shutdown_event is not None else requests_queue.get)  # type: ignore[attr-defined]
+                        return None
+
+        get_method = _get_queue_intermittently \
+            if shutdown_event is not None \
+            else requests_queue.get
+        return await asyncio.to_thread(get_method)  # type: ignore[attr-defined]
 
     async def send_result(
         self,
@@ -165,7 +173,8 @@ async def resolve_scheduler_request(
             scheduled_time=time.time(),
             process_id=process_id,
         )
-        request_scheduled_result: WorkerProcessResult[RequestT, ResponseT] = WorkerProcessResult(
+        request_scheduled_result: WorkerProcessResult[RequestT, ResponseT] = \
+            WorkerProcessResult(
             type_="request_scheduled",
             request=request,
             response=None,