WIP

Author: markvaykhansky

.pre-commit-config.yaml

@@ -4,40 +4,40 @@ repos:
   hooks:
   - id: trailing-whitespace
   - id: end-of-file-fixer
-- repo: https://github.com/astral-sh/ruff-pre-commit
-  rev: v0.11.7
-  hooks:
-    - id: ruff
-- repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v1.15.0
-  hooks:
-  - id: mypy
-    args: [--check-untyped-defs]
-    additional_dependencies:
-      [
-        # main dependencies
-        click,
-        datasets,
-        ftfy,
-        loguru,
-        numpy,
-        pillow,
-        pydantic,
-        pydantic_settings,
-        pyyaml,
-        respx,
-        rich,
-        setuptools,
-        setuptools-git-versioning,
-        transformers,
-
-        # dev dependencies
-        pytest,
-        pydantic_settings,
-
-        # types
-        types-click,
-        types-PyYAML,
-        types-requests,
-        types-toml,
-      ]
+#- repo: https://github.com/astral-sh/ruff-pre-commit
+#  rev: v0.11.7
+#  hooks:
+#    - id: ruff
+#- repo: https://github.com/pre-commit/mirrors-mypy
+#  rev: v1.15.0
+#  hooks:
+#  - id: mypy
+#    args: [--check-untyped-defs]
+#    additional_dependencies:
+#      [
+#        # main dependencies
+#        click,
+#        datasets,
+#        ftfy,
+#        loguru,
+#        numpy,
+#        pillow,
+#        pydantic,
+#        pydantic_settings,
+#        pyyaml,
+#        respx,
+#        rich,
+#        setuptools,
+#        setuptools-git-versioning,
+#        transformers,
+#
+#        # dev dependencies
+#        pytest,
+#        pydantic_settings,
+#
+#        # types
+#        types-click,
+#        types-PyYAML,
+#        types-requests,
+#        types-toml,
+#      ]

src/guidellm/scheduler/scheduler.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 import asyncio
 import math
 import multiprocessing
@@ -269,27 +270,17 @@ async def _start_processes(
         futures = []
         loop = asyncio.get_event_loop()
         for id_, requests_limit in zip(process_ids, process_requests_limits):
-            if scheduling_strategy.processing_mode == "sync":
+            if scheduling_strategy.processing_mode in ["sync", "async"]:
                 futures.append(
                     loop.run_in_executor(
                         executor,
-                        self.worker.process_loop_synchronous,
+                        self.worker.run_process,
                         requests_queue,
                         responses_queue,
-                        id_,
                         shutdown_event,
-                    )
-                )
-            elif scheduling_strategy.processing_mode == "async":
-                futures.append(
-                    loop.run_in_executor(
-                        executor,
-                        self.worker.process_loop_asynchronous,
-                        requests_queue,
-                        responses_queue,
-                        requests_limit,
+                        timedelta(seconds=10).total_seconds(),
                         id_,
-                        shutdown_event,
+                        requests_limit,
                     )
                 )
             else:

src/guidellm/scheduler/worker.py

@@ -194,22 +194,25 @@ async def resolve_scheduler_request(
 
     def run_process(
             self,
-            type_: Literal["synchronous", "asynchronous"],
+            type_: Literal["sync", "async"],
             requests_queue: multiprocessing.Queue,
             results_queue: multiprocessing.Queue,
             shutdown_event: multiprocessing.Event,
-            shutdown_poll_interval: float,
+            shutdown_poll_interval_seconds: float,
             process_id: int,
-            max_concurrency: int,
+            max_concurrency: Optional[int] = None,
     ):
         async def _process_runner():
-            if type_ == "synchronous":
+            if type_ == "sync":
                 loop_task = asyncio.create_task(self._process_synchronous_requests_loop(
                     requests_queue=requests_queue,
                     results_queue=results_queue,
                     process_id=process_id,
                 ), name="request_loop_processor_task")
-            elif type_ == "asynchronous":
+            elif type_ == "async":
+                if max_concurrency is None:
+                    raise ValueError("max_concurrency must be set "
+                                     "for async processor")
                 loop_task = asyncio.create_task(self._process_asynchronous_requests_loop(
                     requests_queue=requests_queue,
                     results_queue=results_queue,
@@ -220,8 +223,8 @@ async def _process_runner():
                 raise ValueError(f"Invalid process type: {type_}")
 
             shutdown_task = asyncio.create_task(
-                self._wait_for_shutdown(shutdown_event, shutdown_poll_interval),
-                name="shutdown_task"
+                self._wait_for_shutdown(shutdown_event, shutdown_poll_interval_seconds),
+                name="shutdown_task",
             )
 
             done, pending = await asyncio.wait(