Drop times queue and handling timing in each worker

Author: sjmonson

src/guidellm/scheduler/result.py

@@ -17,7 +17,7 @@
     "SchedulerRequestResult",
     "SchedulerResult",
     "SchedulerRunInfo",
-    "WorkerProcessRequestTime",
+    "WorkerProcessRequest",
     "WorkerProcessResult",
 ]
 
@@ -147,8 +147,8 @@ class SchedulerRequestResult(
 
 
 @dataclass
-class WorkerProcessRequestTime:
-    start_time: float
+class WorkerProcessRequest(Generic[RequestT, ResponseT]):
+    session: RequestSession[RequestT, ResponseT]
     timeout_time: float
     queued_time: float
 
@@ -163,6 +163,5 @@ class WorkerProcessResult(Generic[RequestT, ResponseT]):
 
 @dataclass
 class MPQueues(Generic[RequestT, ResponseT]):
-    requests: Queue[RequestSession[RequestT, ResponseT]]
-    times: Queue[WorkerProcessRequestTime]
+    requests: Queue[WorkerProcessRequest[RequestT, ResponseT]]
     responses: Queue[WorkerProcessResult[RequestT, ResponseT]]

src/guidellm/scheduler/scheduler.py

@@ -17,7 +17,6 @@
 from loguru import logger
 
 from guidellm.config import settings
-from guidellm.request.session import RequestSession
 from guidellm.request.types import (
     RequestT,
     ResponseT,
@@ -27,7 +26,7 @@
     SchedulerRequestResult,
     SchedulerResult,
     SchedulerRunInfo,
-    WorkerProcessRequestTime,
+    WorkerProcessRequest,
     WorkerProcessResult,
 )
 from guidellm.scheduler.strategy import SchedulingStrategy
@@ -127,10 +126,14 @@ async def run(
             ) as executor,
         ):
             requests_iter: Optional[Iterator[Any]] = None
+            # TODO: Configurable delay and move somewhere more appropriate
+            scheduling_strategy.start_time = (
+                time.time()
+            )  # Add a small delay to allow processes to start
             futures, queues, stop_event = await self._start_processes(
                 manager, executor, scheduling_strategy
             )
-            run_info, requests_iter, times_iter = self._run_setup(
+            run_info, requests_iter = self._run_setup(
                 futures, scheduling_strategy, max_number, max_duration
             )
             yield SchedulerResult(
@@ -147,17 +150,16 @@ async def run(
 
                     if (
                         requests_iter is None
-                        and run_info.completed_requests >= run_info.created_requests
+                        # FIXME: Need new way to handle max requests
+                        # and run_info.completed_requests >= run_info.created_requests
                     ):
                         # we've exhausted all requests we've wanted to run
                         # and yielded all responses
                         break
 
                     requests_iter = self._add_requests(
                         requests_iter,
-                        times_iter,
                         queues.requests,
-                        queues.times,
                         run_info,
                     )
                     await asyncio.sleep(0)  # enable requests to start
@@ -196,7 +198,6 @@ async def _start_processes(
             requests=manager.Queue(
                 maxsize=scheduling_strategy.processing_requests_limit
             ),
-            times=manager.Queue(maxsize=scheduling_strategy.processing_requests_limit),
             responses=manager.Queue(),
         )
         stop_event = manager.Event()
@@ -229,10 +230,12 @@ async def _start_processes(
                     executor,
                     self.worker.process_loop_asynchronous,
                     queues,
+                    scheduling_strategy,
                     stop_event,
                     False,  # TODO: Make configurable
                     requests_limit,
                     id_,
+                    num_processes,
                 )
             )
 
@@ -246,11 +249,9 @@ def _run_setup(
         scheduling_strategy: SchedulingStrategy,
         max_number: Optional[int],
         max_duration: Optional[float],
-    ) -> tuple[SchedulerRunInfo, Iterator[Any], Iterator[float]]:
+    ) -> tuple[SchedulerRunInfo, Iterator[Any]]:
         requests_iter = iter(self.request_loader)
-        start_time = time.time()
-        times_iter = iter(scheduling_strategy.request_times())
-        end_time = time.time() + (max_duration or math.inf)
+        end_time = scheduling_strategy.start_time + (max_duration or math.inf)
         end_number = max_number or math.inf
 
         try:
@@ -268,27 +269,28 @@ def _run_setup(
             )
 
         info = SchedulerRunInfo(
-            start_time=start_time,
+            start_time=scheduling_strategy.start_time,
             end_time=end_time,
             end_number=end_number,
             processes=len(processes),
             strategy=scheduling_strategy,
         )
 
-        return info, requests_iter, times_iter
+        return info, requests_iter
 
     def _add_requests(
         self,
         requests_iter: Optional[Iterator[Any]],
-        times_iter: Iterator[float],
-        requests_queue: Queue[RequestSession[RequestT, ResponseT]],
-        times_queue: Queue[WorkerProcessRequestTime],
+        requests_queue: Queue[WorkerProcessRequest[RequestT, ResponseT]],
         run_info: SchedulerRunInfo,
     ) -> Optional[Iterator[Any]]:
         if requests_iter is not None:
             try:
                 added_count = 0
 
+                if time.time() >= run_info.end_time:
+                    raise StopIteration
+
                 while (
                     not requests_queue.full()
                     and added_count < settings.max_add_requests_per_loop
@@ -297,23 +299,16 @@ def _add_requests(
                         raise StopIteration
 
                     session = next(requests_iter)
-                    requests_queue.put(session)
-                    for _ in range(len(session)):
-                        if (
-                            request_time := next(times_iter)
-                        ) >= run_info.end_time or time.time() >= run_info.end_time:
-                            raise StopIteration
-
-                        work_req = WorkerProcessRequestTime(
-                            start_time=request_time,
-                            timeout_time=run_info.end_time,
-                            queued_time=time.time(),
-                        )
-                        times_queue.put(work_req)
-
-                        run_info.created_requests += 1
-                        run_info.queued_requests += 1
-                        added_count += 1
+                    work_req = WorkerProcessRequest(
+                        session=session,
+                        timeout_time=run_info.end_time,
+                        queued_time=time.time(),
+                    )
+                    requests_queue.put(work_req)
+
+                    run_info.created_requests += len(session)
+                    run_info.queued_requests += len(session)
+                    added_count += len(session)
             except StopIteration:
                 # we've reached the limit number, limit time, or exhausted the requests
                 # set to None to stop adding more and tell the loop no more requests

src/guidellm/scheduler/strategy.py

@@ -44,6 +44,10 @@ class SchedulingStrategy(StandardBaseModel):
     type_: Literal["strategy"] = Field(
         description="The type of scheduling strategy schedule requests with.",
     )
+    start_time: float = Field(
+        default_factory=time.time,
+        description="The start time for the scheduling strategy.",
+    )
 
     @property
     def processing_mode(self) -> Literal["sync", "async"]:
@@ -175,8 +179,9 @@ def request_times(self) -> Generator[float, None, None]:
 
         :return: A generator that yields time.time() for immediate request scheduling.
         """
+        init_time = self.start_time
         while True:
-            yield time.time()
+            yield max(init_time, time.time())
 
 
 class ConcurrentStrategy(SchedulingStrategy):
@@ -262,8 +267,9 @@ def request_times(self) -> Generator[float, None, None]:
 
         :return: A generator that yields time.time() for immediate request scheduling.
         """
+        init_time = self.start_time
         while True:
-            yield time.time()
+            yield max(init_time, time.time())
 
 
 class ThroughputStrategy(SchedulingStrategy):
@@ -336,10 +342,9 @@ def request_times(self) -> Generator[float, None, None]:
         :return: A generator that yields the start time.time()
             for immediate request scheduling.
         """
-        start_time = time.time()
-
+        init_time = self.start_time
         while True:
-            yield start_time
+            yield init_time
 
 
 class AsyncConstantStrategy(ThroughputStrategy):
@@ -391,24 +396,24 @@ def request_times(self) -> Generator[float, None, None]:
 
         :return: A generator that yields timestamps for request scheduling.
         """
-        start_time = time.time()
         constant_increment = 1.0 / self.rate
 
+        init_time = self.start_time
         # handle bursts first to get to the desired rate
         if self.initial_burst is not None:
             # send an initial burst equal to the rate
             # to reach the target rate
             burst_count = math.floor(self.rate)
             for _ in range(burst_count):
-                yield start_time
+                yield init_time
 
-            start_time += constant_increment
+            init_time += constant_increment
 
         counter = 0
 
         # continue with constant rate after bursting
         while True:
-            yield start_time + constant_increment * counter
+            yield init_time + constant_increment * counter
             counter += 1
 
 
@@ -461,24 +466,23 @@ def request_times(self) -> Generator[float, None, None]:
 
         :return: A generator that yields timestamps for request scheduling.
         """
-        start_time = time.time()
-
+        init_time = self.start_time
         if self.initial_burst is not None:
             # send an initial burst equal to the rate
             # to reach the target rate
             burst_count = math.floor(self.rate)
             for _ in range(burst_count):
-                yield start_time
+                yield init_time
         else:
-            yield start_time
+            yield init_time
 
         # set the random seed for reproducibility
         rand = random.Random(self.random_seed)  # noqa: S311
 
         while True:
             inter_arrival_time = rand.expovariate(self.rate)
-            start_time += inter_arrival_time
-            yield start_time
+            init_time += inter_arrival_time
+            yield init_time
 
 
 def strategy_display_str(strategy: Union[StrategyType, SchedulingStrategy]) -> str: