Revert "Move prepare_multiprocessing earlier"

This reverts commit e6a908a.

Author: sjmonson

src/guidellm/backend/backend.py

@@ -110,6 +110,14 @@ def info(self) -> dict[str, Any]:
         """
         ...
 
+    @abstractmethod
+    async def reset(self) -> None:
+        """
+        Reset the connection object. This is useful for backends that
+        reuse connections or have state that needs to be cleared.
+        """
+        ...
+
     async def validate(self):
         """
         Handle final setup and validate the backend is ready for use.
@@ -126,6 +134,8 @@ async def validate(self):
         ):  # type: ignore[attr-defined]
             pass
 
+        await self.reset()
+
     @abstractmethod
     async def check_setup(self):
         """

src/guidellm/backend/openai.py

@@ -157,6 +157,15 @@ def info(self) -> dict[str, Any]:
             "chat_completions_path": CHAT_COMPLETIONS_PATH,
         }
 
+    async def reset(self) -> None:
+        """
+        Reset the connection object. This is useful for backends that
+        reuse connections or have state that needs to be cleared.
+        For this backend, it closes the async client if it exists.
+        """
+        if self._async_client is not None:
+            await self._async_client.aclose()
+
     async def check_setup(self):
         """
         Check if the backend is setup correctly and can be used for requests.

src/guidellm/scheduler/scheduler.py

@@ -113,8 +113,6 @@ async def run(
         if max_duration is not None and max_duration < 0:
             raise ValueError(f"Invalid max_duration: {max_duration}")
 
-        # Must call before workers fork
-        await self.worker.prepare_multiprocessing()
         with (
             multiprocessing.Manager() as manager,
             ProcessPoolExecutor(
@@ -185,6 +183,7 @@ async def _start_processes(
         multiprocessing.Queue,
         multiprocessing.Queue,
     ]:
+        await self.worker.prepare_multiprocessing()
         requests_queue = manager.Queue(
             maxsize=scheduling_strategy.queued_requests_limit
         )