wip

Author: markVaykhansky

src/guidellm/scheduler/scheduler.py

@@ -64,7 +64,6 @@ def __init__(
 
         self.worker = worker
         self.request_loader = request_loader
-        self.error_rate: Optional[float] = None
 
     async def run(
         self,
@@ -118,16 +117,14 @@ async def run(
         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}")
 
-        shutdown_event = multiprocessing.Event()
-
         with (
             multiprocessing.Manager() as manager,
             ProcessPoolExecutor(
                 max_workers=scheduling_strategy.processes_limit
             ) as executor,
         ):
             requests_iter: Optional[Iterator[Any]] = None
-            futures, requests_queue, responses_queue = await self._start_processes(
+            futures, requests_queue, responses_queue, shutdown_event = await self._start_processes(
                 manager, executor, scheduling_strategy
             )
             run_info, requests_iter, times_iter = self._run_setup(
@@ -167,7 +164,9 @@ async def run(
                     )
                     if iter_result is not None:
                         if self._is_max_error_rate_reached(iter_result.run_info):
-                            logger.info(f"Max_error rate of ({iter_result.run_info.max_error_rate}) reached!")
+                            logger.info(f"Max_error rate of ({iter_result.run_info.max_error_rate}) reached, sending "
+                                        f"shutdown signal")
+                            shutdown_event.set()
                         yield iter_result
 
                     # yield control to the event loop
@@ -191,8 +190,10 @@ async def _start_processes(
         list[asyncio.Future],
         multiprocessing.Queue,
         multiprocessing.Queue,
+        multiprocessing.Event
     ]:
         await self.worker.prepare_multiprocessing()
+        shutdown_event = multiprocessing.Event()
         requests_queue = manager.Queue(
             maxsize=scheduling_strategy.queued_requests_limit
         )
@@ -229,6 +230,7 @@ async def _start_processes(
                         requests_queue,
                         responses_queue,
                         id_,
+                        shutdown_event,
                     )
                 )
             elif scheduling_strategy.processing_mode == "async":
@@ -240,6 +242,7 @@ async def _start_processes(
                         responses_queue,
                         requests_limit,
                         id_,
+                        shutdown_event,
                     )
                 )
             else:
@@ -250,7 +253,7 @@ async def _start_processes(
 
         await asyncio.sleep(0.1)  # give time for processes to start
 
-        return futures, requests_queue, responses_queue
+        return futures, requests_queue, responses_queue, shutdown_event
 
     def _run_setup(
         self,
@@ -385,8 +388,7 @@ def _check_result_ready(
             )
         raise ValueError(f"Invalid process response type: {process_response}")
 
-    @staticmethod
-    def _is_max_error_rate_reached(run_info: SchedulerRunInfo) -> bool:
+    def _is_max_error_rate_reached(self, run_info: SchedulerRunInfo) -> bool:
         current_error_rate = run_info.errored_requests / run_info.end_number
         return current_error_rate > run_info.max_error_rate
 

src/guidellm/scheduler/worker.py

@@ -121,9 +121,13 @@ async def resolve(
         ...
 
     async def get_request(
-        self, requests_queue: multiprocessing.Queue
+        self, requests_queue: multiprocessing.Queue, shutdown_event: multiprocessing.Event, shutdonen_check_
     ) -> Optional[WorkerProcessRequest[RequestT]]:
-        return await asyncio.to_thread(requests_queue.get)  # type: ignore[attr-defined]
+        def _get_queue_intermittently(request_queue: multiprocessing.Queue, shutdown_event):
+            try:
+                
+
+        return await asyncio.to_thread(_get_queue_intermittently())  # type: ignore[attr-defined]
 
     async def send_result(
         self,
@@ -222,6 +226,7 @@ def process_loop_asynchronous(
         results_queue: multiprocessing.Queue,
         max_concurrency: int,
         process_id: int,
+        shutdown_event: multiprocessing.Event,
     ):
         async def _process_runner():
             pending = asyncio.Semaphore(max_concurrency)