Type fixes, typos & bugfixes

Author: markVaykhansky

src/guidellm/__main__.py

@@ -179,7 +179,7 @@ def cli():
     type=float,
     default=None,
     help=(
-        "The percent of the benchmark (based on max-seconds, max-requets, "
+        "The percent of the benchmark (based on max-seconds, max-requests, "
         "or lenth of dataset) to run as a warmup and not include in the final results. "
         "Defaults to None."
     ),
@@ -188,7 +188,7 @@ def cli():
     "--cooldown-percent",
     type=float,
     help=(
-        "The percent of the benchmark (based on max-seconds, max-requets, or lenth "
+        "The percent of the benchmark (based on max-seconds, max-requests, or length "
         "of dataset) to run as a cooldown and not include in the final results. "
         "Defaults to None."
     ),

src/guidellm/scheduler/scheduler.py

@@ -5,6 +5,7 @@
 import time
 from collections.abc import AsyncGenerator, Iterable, Iterator
 from concurrent.futures import ProcessPoolExecutor
+from multiprocessing.synchronize import Event as MultiprocessingEvent
 from typing import (
     Any,
     Generic,
@@ -168,11 +169,15 @@ async def run(
                         if iter_result.request_info.errored \
                         and not iter_result.request_info.canceled \
                         and self._is_max_error_rate_reached(iter_result.run_info):
+                            if shutdown_event is None:
+                                raise RuntimeError("We've reached max_error_rate "
+                                                   "but shutdown_event is corrupt")
                             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")
+                        logger.info("Itter is not None")
                         yield iter_result
 
                     # yield control to the event loop
@@ -205,8 +210,12 @@ def _validate_scheduler_params(
         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:
+    def _is_max_error_rate_reached(self, run_info: SchedulerRunInfo) -> bool:
+        if run_info.max_error_rate is None:
+            return False
         current_error_rate = run_info.errored_requests / run_info.end_number
+        logger.info(f"Current error rate {current_error_rate} "
+                    f"i.e total_finished [success / error] / max total possible")
         return run_info.max_error_rate < current_error_rate
 
     async def _start_processes(
@@ -219,7 +228,7 @@ async def _start_processes(
         list[asyncio.Future],
         multiprocessing.Queue,
         multiprocessing.Queue,
-        Optional[multiprocessing.Event]
+        Optional[MultiprocessingEvent]
     ]:
         await self.worker.prepare_multiprocessing()
         shutdown_event = manager.Event() if create_shutdown_event else None
@@ -232,7 +241,6 @@ async def _start_processes(
             scheduling_strategy.processes_limit,
             scheduling_strategy.processing_requests_limit,
         )
-        num_processes = 1
         requests_limit_split = (
             scheduling_strategy.processing_requests_limit
             // scheduling_strategy.processes_limit
@@ -327,7 +335,7 @@ def _determine_total_requests_count(
             scheduling_strategy: SchedulingStrategy,
             max_duration: Optional[float],
             max_number: Optional[int],
-    ) -> int:
+    ) -> Union[int, float]:
         end_number = max_number or math.inf
         try:
             # update end_number if the request_loader is finite and less than max_number

src/guidellm/scheduler/worker.py

@@ -1,13 +1,13 @@
 import asyncio
 import math
-import multiprocessing
 import multiprocessing.queues
 import queue
 import time
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator
 from dataclasses import dataclass
 from datetime import timedelta
+from multiprocessing.synchronize import Event as MultiprocessingEvent
 from typing import (
     Any,
     Generic,
@@ -124,7 +124,7 @@ async def resolve(
 
     async def get_request(
         self, requests_queue: multiprocessing.Queue,
-            shutdown_event: Optional[multiprocessing.Event] = None,
+            shutdown_event: Optional[MultiprocessingEvent] = None,
             process_id: Optional[int] = None,
     ) -> Optional[WorkerProcessRequest[RequestT]]:
         if shutdown_event is not None and process_id is None:
@@ -186,7 +186,8 @@ async def resolve_scheduler_request(
             await asyncio.sleep(wait_time)
 
         info.worker_start = time.time()
-        request_start_result = WorkerProcessResult(
+        request_start_result: WorkerProcessResult[RequestT, ResponseT] = \
+            WorkerProcessResult(
             type_="request_start",
             request=request,
             response=None,
@@ -215,7 +216,7 @@ def process_loop_synchronous(
         requests_queue: multiprocessing.Queue,
         results_queue: multiprocessing.Queue,
         process_id: int,
-        shutdown_event: Optional[multiprocessing.Event] = None,
+        shutdown_event: Optional[MultiprocessingEvent] = None,
     ):
         async def _process_runner():
             while (
@@ -256,7 +257,7 @@ def process_loop_asynchronous(
         results_queue: multiprocessing.Queue,
         max_concurrency: int,
         process_id: int,
-        shutdown_event: Optional[multiprocessing.Event] = None,
+        shutdown_event: Optional[MultiprocessingEvent] = None,
     ):
         async def _process_runner():
             pending = asyncio.Semaphore(max_concurrency)
@@ -355,7 +356,7 @@ def process_loop_synchronous(
         requests_queue: multiprocessing.Queue,
         results_queue: multiprocessing.Queue,
         process_id: int,
-        shutdown_event: Optional[multiprocessing.Event] = None
+        shutdown_event: Optional[MultiprocessingEvent] = None
     ):
         asyncio.run(self.backend.validate())
         super().process_loop_synchronous(
@@ -371,7 +372,7 @@ def process_loop_asynchronous(
         results_queue: multiprocessing.Queue,
         max_concurrency: int,
         process_id: int,
-        shutdown_event: Optional[multiprocessing.Event] = None
+        shutdown_event: Optional[MultiprocessingEvent] = None
     ):
         asyncio.run(self.backend.validate())
         super().process_loop_asynchronous(