Testing for parity between cli and server

shortfin/python/shortfin/support/responder.py

@@ -5,6 +5,9 @@
 # SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
 
+from shortfin.support.status_tracker import AbstractStatusTracker
+
+
 class AbstractResponder:
     """Interface for a responder to"""
 
@@ -25,3 +28,9 @@ def stream_start(self, **kwargs):
 
     def stream_part(self, content: bytes | None):
         pass
+
+    def get_status_tracker(self):
+        return AbstractStatusTracker()
+
+    def is_disconnected(self):
+        return False

shortfin/python/shortfin/support/status_tracker.py

@@ -20,4 +20,4 @@ def __init__(self):
 
     def is_disconnected(self) -> bool:
         """Returns True if the connection/request is considered disconnected."""
-        raise NotImplementedError()
+        return False

shortfin/python/shortfin_apps/llm/cli.py

@@ -10,10 +10,11 @@
 import logging
 import sys
 import time
-
+import numpy as np
 
 # Import first as it does dep checking and reporting.
 from pathlib import Path
+from typing import List, Optional
 from shortfin import ProgramIsolation
 from shortfin.support.logging_setup import configure_main_logger
 from shortfin.support.responder import AbstractResponder
@@ -154,7 +155,7 @@ def parse_args(argv):
 
 def process_inputs(args):
     if args.prompt:
-        prompts = [args.prompt]
+        prompts = ["".join(["one " * 2500])]
         if args.benchmark and args.benchmark_tasks is not None:
             prompts = prompts * args.benchmark_tasks
         return prompts
@@ -163,15 +164,18 @@ def process_inputs(args):
 
 
 class Timer:
-    def __init__(self):
+    def __init__(self, name: str):
+        self._name = name
         self._start = None
         self._end = None
 
     def start(self):
         self._start = time.perf_counter()
+        print(f"{self._name} start time: {self._start}")
 
     def end(self):
         self._end = time.perf_counter()
+        print(f"{self._name} end time: {self._end}")
 
     def elapsed(self):
         if self._end is None:
@@ -185,7 +189,16 @@ def __init__(self):
         self._loop = asyncio.get_running_loop()
         self.response = asyncio.Future(loop=self._loop)
         self.responded = False
-        self.timer = Timer()
+        self.idx = self._get_idx()
+        self.name = f"CliResponder-{self.idx}"
+        self.timer = Timer(self.name)
+
+    @classmethod
+    def _get_idx(cls):
+        if not hasattr(cls, "_idx"):
+            cls._idx = 0
+        cls._idx += 1
+        return cls._idx
 
     def start_response(self):
         self.timer.start()
@@ -194,6 +207,7 @@ def ensure_response(self):
         self.timer.end()
 
     def send_response(self, response):
+        print(f"{self.name} Sending response")
         assert not self.responded, "Response already sent"
         if self._loop.is_closed():
             raise IOError("Web server is shut down")
@@ -232,20 +246,20 @@ async def main(argv):
     class Task:
         def __init__(self, prompt):
             self.prompt = prompt
-            self.responder = None
+            self.responder: Optional[CliResponder] = None
 
         def runtime(self):
             return self.responder.timer.elapsed()
 
     logger.info(msg=f"Setting up a tasklist of {len(prompts)} items")
-    tasks = []
+    tasks: List[Task] = []
     for p in prompts:
         task = Task(p)
         tasks.append(task)
 
     async def worker(name, queue, fiber):
         while True:
-            task = await queue.get()
+            task: Task = await queue.get()
             responder = CliResponder()
             gen_req = GenerateReqInput(
                 text=task.prompt, sampling_params=sampling_params
@@ -270,7 +284,7 @@ async def worker(name, queue, fiber):
 
     logger.info(msg=f"Processing tasks")
 
-    global_timer = Timer()
+    global_timer = Timer("global")
     global_timer.start()
     for t in tasks:
         queue.put_nowait(t)
@@ -282,13 +296,14 @@ async def worker(name, queue, fiber):
         w.cancel()
 
     if args.benchmark:
-        latency_sum = sum([s.runtime() for s in tasks])
-        latency_avg = latency_sum / len(tasks)
         total_time = global_timer.elapsed()
         reqs = len(prompts) / total_time
 
         print(f"Requests per second: {reqs:2f}")
-        print(f"AverageLatency:      {latency_avg:2f}")
+        latencies = [s.runtime() for s in tasks]
+        print(
+            f"Latencies: av: {np.mean(latencies)}, min: {np.min(latencies)}, max: {np.max(latencies)}, median: {np.median(latencies)}, sd: {np.std(latencies)}"
+        )
 
     logger.info(msg=f"Shutting down service")
     service.shutdown()

shortfin/python/shortfin_apps/llm/routes/generate.py

@@ -14,13 +14,16 @@
 
 generation_router = APIRouter()
 
+test_text = "".join(["one " * 2500])
+
 
 @generation_router.post("/generate")
 @generation_router.put("/generate")
 async def generate_request(gen_req: GenerateReqInput, request: Request):
     # app.state.services is populated by the ShortfinLlmLifecycleManager
     # see shortfin/python/shortfin_apps/llm/components/lifecycle.py
     service: GenerateService = request.app.state.services["default"]
+    gen_req.text = test_text
     gen_req.post_init()
     responder = FastAPIResponder(request)
     ClientGenerateBatchProcess(