Strip out multiturn features

Author: sjmonson

src/guidellm/request/__init__.py

@@ -5,17 +5,14 @@
     RequestLoaderDescription,
 )
 from .request import GenerationRequest
-from .session import GenerativeRequestSession, RequestSession
 from .types import RequestT, ResponseT
 
 __all__ = [
     "GenerationRequest",
     "GenerativeRequestLoader",
     "GenerativeRequestLoaderDescription",
-    "GenerativeRequestSession",
     "RequestLoader",
     "RequestLoaderDescription",
-    "RequestSession",
     "RequestT",
     "ResponseT",
 ]

src/guidellm/request/loader.py

@@ -15,7 +15,6 @@
 from guidellm.dataset import ColumnInputTypes, load_dataset
 from guidellm.objects import StandardBaseModel
 from guidellm.request.request import GenerationRequest
-from guidellm.request.session import GenerativeRequestSession
 
 __all__ = [
     "GenerativeRequestLoader",
@@ -106,14 +105,14 @@ def __init__(
         self.preserve_iter_state = iter_type == "infinite"  # ensure no caching requests
         self._preserved_iter = None
 
-    def __iter__(self) -> Iterator[GenerativeRequestSession]:
+    def __iter__(self) -> Iterator[GenerationRequest]:
         scope_create_count = 0
 
         while (dataset_iter := self._get_dataset_iter(scope_create_count)) is not None:
             scope_create_count += 1
 
             for item in dataset_iter:
-                yield GenerativeRequestSession(self._create_request(item))
+                yield self._create_request(item)
 
             self._preserved_iter = None
 

src/guidellm/request/session.py

@@ -6,7 +6,6 @@
 )
 
 from guidellm.objects import StandardBaseModel
-from guidellm.request.session import RequestSession
 from guidellm.request.types import RequestT, ResponseT
 from guidellm.scheduler.strategy import SchedulingStrategy
 
@@ -143,7 +142,7 @@ class SchedulerRequestResult(
 
 @dataclass
 class WorkerProcessRequest(Generic[RequestT, ResponseT]):
-    session: RequestSession[RequestT, ResponseT]
+    request: RequestT
     timeout_time: float
     queued_time: float
 

src/guidellm/scheduler/result.py

@@ -234,7 +234,6 @@ async def _start_processes(
                     queues,
                     scheduling_strategy,
                     stop_event,
-                    False,  # TODO: Make configurable
                     requests_limit,
                     id_,
                     num_processes,
@@ -300,17 +299,16 @@ def _add_requests(
                     if run_info.created_requests >= run_info.end_number:
                         raise StopIteration
 
-                    session = next(requests_iter)
-                    work_req = WorkerProcessRequest(
-                        session=session,
+                    work_req = WorkerProcessRequest[RequestT, ResponseT](
+                        request=next(requests_iter),
                         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)
+                    run_info.created_requests += 1
+                    run_info.queued_requests += 1
+                    added_count += 1
             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/scheduler.py

@@ -123,7 +123,7 @@ async def resolve_scheduler_request(
         results_queue: Queue[WorkerProcessResult[RequestT, ResponseT]],
         process_id: int,
     ) -> WorkerProcessRequest[RequestT, ResponseT]:
-        request = process_request.session.get_next_request()
+        request = process_request.request
         timeout_time = process_request.timeout_time
         queued_time = process_request.queued_time
 
@@ -170,22 +170,19 @@ async def resolve_scheduler_request(
         )
         asyncio.create_task(self.send_result(results_queue, result))
 
-        process_request.session.push_response(response)
         return process_request
 
     def process_loop_asynchronous(
         self,
         queues: MPQueues[RequestT, ResponseT],
         strategy: SchedulingStrategy,
         stop_event: Event,
-        prioritize_sessions: bool,
         max_concurrency: int,
         process_id: int,
         num_processes: int,
     ):
         async def _process_runner():
             lock = asyncio.Semaphore(max_concurrency)
-            pending_requests: list[WorkerProcessRequest[RequestT, ResponseT]] = []
             times_iter = islice(
                 strategy.request_times(),
                 process_id,
@@ -202,50 +199,18 @@ async def _process_runner():
                 await asyncio.sleep(start_time - time.time() - 1)
                 await lock.acquire()
 
-                process_request = None
                 try:
-                    process_request = (
-                        pending_requests.pop()
-                        if pending_requests
-                        else queues.requests.get_nowait()
-                    )
+                    process_request = queues.requests.get_nowait()
                     dequeued_time = time.time()
                 except QueueEmpty:
                     lock.release()
                     continue
 
-                async def wait_then_requeue(
-                    process_request: WorkerProcessRequest[RequestT, ResponseT],
-                ):
-                    # Wait to requeue the request session if it specifies a delay
-                    if delay := process_request.session.get_next_delay():
-                        await asyncio.sleep(delay)
-
-                    # Push session to the stack
-                    process_request.queued_time = time.time()
-                    pending_requests.append(process_request)
-                    if prioritize_sessions:
-                        # Release the lock with the session on top of the stack
-                        lock.release()
-
                 def _request_callback(
-                    future: asyncio.Future[WorkerProcessRequest[RequestT, ResponseT]],
+                    _: asyncio.Future[WorkerProcessRequest[RequestT, ResponseT]],
                 ):
-                    # If we are prioritizing sessions, hold
-                    # the lock until the session is done
                     nonlocal lock
-                    if not prioritize_sessions:
-                        lock.release()
-
-                    try:
-                        process_request = future.result()
-                    except asyncio.CancelledError:
-                        return
-                    if not process_request.session.complete:
-                        asyncio.create_task(wait_then_requeue(process_request))
-                    elif prioritize_sessions:
-                        # no more requests in this session, release the lock
-                        lock.release()
+                    lock.release()
 
                 task = asyncio.create_task(
                     self.resolve_scheduler_request(
@@ -319,7 +284,6 @@ def process_loop_asynchronous(
         queues: MPQueues[GenerationRequest, ResponseSummary],
         strategy: SchedulingStrategy,
         stop_event: Event,
-        prioritize_sessions: bool,
         max_concurrency: int,
         process_id: int,
         num_processes: int,
@@ -329,7 +293,6 @@ def process_loop_asynchronous(
             queues=queues,
             strategy=strategy,
             stop_event=stop_event,
-            prioritize_sessions=prioritize_sessions,
             max_concurrency=max_concurrency,
             process_id=process_id,
             num_processes=num_processes,