Determine number of tasks on per-process basis

Author: sjmonson

src/guidellm/scheduler/scheduler.py

@@ -188,9 +188,6 @@ async def _start_processes(
             maxsize=scheduling_strategy.queued_requests_limit
         )
         responses_queue = manager.Queue()
-        per_process_requests_limit = scheduling_strategy.processing_requests_limit // (
-            scheduling_strategy.processes_limit
-        )
 
         futures = []
         loop = asyncio.get_event_loop()
@@ -206,16 +203,17 @@ async def _start_processes(
                     )
                 )
             elif scheduling_strategy.processing_mode == "async":
-                futures.append(
-                    loop.run_in_executor(
-                        executor,
-                        self.worker.process_loop_asynchronous,
-                        requests_queue,
-                        responses_queue,
-                        per_process_requests_limit,
-                        process_id,
+                if scheduling_strategy.process_requests_limits[process_id]:
+                    futures.append(
+                        loop.run_in_executor(
+                            executor,
+                            self.worker.process_loop_asynchronous,
+                            requests_queue,
+                            responses_queue,
+                            scheduling_strategy.process_requests_limits[process_id],
+                            process_id,
+                        )
                     )
-                )
             else:
                 raise ValueError(
                     f"Invalid processing mode: {scheduling_strategy.processing_mode} "

src/guidellm/scheduler/strategy.py

@@ -94,6 +94,23 @@ def processing_requests_limit(self) -> int:
         """
         return settings.max_concurrency
 
+    @property
+    def process_requests_limits(self) -> list[int]:
+        """
+        The maximum number of requests per process for the scheduling strategy.
+        It determines how many requests can be processed by each worker process
+        for the scheduling strategy.
+
+        :return: A per-process list of the maximum number of requests per process.
+        """
+        split = self.processing_requests_limit // self.processes_limit
+        remain = self.processing_requests_limit % self.processes_limit
+
+        return [
+            split + 1 if i < remain else split
+            for i in range(self.processes_limit)
+        ]
+
     def request_times(self) -> Generator[float, None, None]:
         """
         A generator that yields timestamps for when requests should be sent.
@@ -168,6 +185,18 @@ def processing_requests_limit(self) -> int:
         """
         return 1
 
+    @property
+    def process_requests_limits(self) -> List[int]:
+        """
+        The maximum number of requests per process for the scheduling strategy.
+        It determines how many requests can be processed by each worker process
+        for the scheduling strategy.
+
+        :return: A per-process list of the maximum number of requests per process.
+        """
+
+        return [1]
+
     def request_times(self) -> Generator[float, None, None]:
         """
         A generator that yields time.time() so requests are sent immediately,

src/guidellm/scheduler/worker.py

@@ -224,7 +224,15 @@ def process_loop_asynchronous(
         process_id: int,
     ):
         async def _process_runner():
-            pending = asyncio.Semaphore(max_concurrency) if max_concurrency else None
+            if max_concurrency is not None:
+                if max_concurrency < 1:
+                    raise ValueError(
+                        f"max_concurrency must be greater than 0, got {max_concurrency}"
+                    )
+
+                pending = asyncio.Semaphore(max_concurrency)
+            else:
+                pending = None
 
             while (
                 process_request := await self.get_request(requests_queue)