Revert loop logic changes

Author: sjmonson

src/guidellm/scheduler/scheduler.py

@@ -130,15 +130,17 @@ async def run(
             futures, queues, stop_event = await self._start_processes(
                 manager, executor, scheduling_strategy
             )
-            run_info, requests_iter = self._run_setup(
+            run_info, requests_iter, times_iter = self._run_setup(
                 futures, scheduling_strategy, max_number, max_duration
             )
 
             # Add some initial requests to the queue
             requests_iter = self._add_requests(
                 requests_iter,
                 queues.requests,
+                times_iter,
                 run_info,
+                loop_limit=run_info.strategy.queued_requests_limit,
             )
             # Wait for the test to start
             await asyncio.sleep(time.time() - scheduling_strategy.start_time)
@@ -154,14 +156,18 @@ async def run(
                         if future.done() and (err := future.exception()) is not None:
                             raise err
 
-                    if requests_iter is None and run_info.processing_requests <= 0:
+                    if (
+                        requests_iter is None
+                        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,
                         queues.requests,
+                        times_iter,
                         run_info,
                     )
                     await asyncio.sleep(0)  # enable requests to start
@@ -250,8 +256,9 @@ def _run_setup(
         scheduling_strategy: SchedulingStrategy,
         max_number: Optional[int],
         max_duration: Optional[float],
-    ) -> tuple[SchedulerRunInfo, Iterator[Any]]:
+    ) -> tuple[SchedulerRunInfo, Iterator[Any], Iterator[float]]:
         requests_iter = iter(self.request_loader)
+        times_iter = iter(scheduling_strategy.request_times())
         end_time = scheduling_strategy.start_time + (max_duration or math.inf)
         end_number = max_number or math.inf
 
@@ -277,28 +284,33 @@ def _run_setup(
             strategy=scheduling_strategy,
         )
 
-        return info, requests_iter
+        return info, requests_iter, times_iter
 
     def _add_requests(
         self,
         requests_iter: Optional[Iterator[Any]],
         requests_queue: Queue[WorkerProcessRequest[RequestT, ResponseT]],
+        times_iter: Iterator[float],
         run_info: SchedulerRunInfo,
+        loop_limit: Optional[int] = None,
     ) -> Optional[Iterator[Any]]:
         if requests_iter is not None:
+            if loop_limit is None:
+                loop_limit = settings.max_add_requests_per_loop
+
             try:
                 added_count = 0
 
-                if time.time() >= run_info.end_time:
-                    raise StopIteration
-
-                while not requests_queue.full() and added_count < (
-                    run_info.strategy.queued_requests_limit
-                    or settings.max_add_requests_per_loop
-                ):
+                while not requests_queue.full() and added_count < loop_limit:
                     if run_info.created_requests >= run_info.end_number:
                         raise StopIteration
 
+                    if (
+                        next(times_iter) >= run_info.end_time
+                        or time.time() >= run_info.end_time
+                    ):
+                        raise StopIteration
+
                     work_req = WorkerProcessRequest[RequestT, ResponseT](
                         request=next(requests_iter),
                         timeout_time=run_info.end_time,