graphing errors over time

lib/backend.py

@@ -5,7 +5,7 @@ import base64
 import subprocess
 import dataclasses
 import logging
-from asyncio import wait, sleep, gather, Semaphore, FIRST_COMPLETED, create_task, get_running_loop
+from asyncio import wait, sleep, gather, Semaphore, FIRST_COMPLETED, create_task
 from typing import Tuple, Awaitable, NoReturn, List, Union, Callable, Optional
 from functools import cached_property
 from distutils.util import strtobool
@@ -26,8 +26,7 @@ from lib.data_types import (
     LogAction,
     ApiPayload_T,
     JsonDataException,
-    RequestMetrics,
-    BenchmarkResult
+    RequestMetrics
 )
 
 VERSION = "0.1.0"
@@ -74,11 +73,6 @@ class Backend:
         self._pubkey = self._fetch_pubkey()
         self.__start_healthcheck: bool = False
 
-        # NEW: FIFO queue + worker count
-        self.request_queue: "asyncio.Queue[tuple[EndpointHandler[ApiPayload_T], web.Request, asyncio.Future]]" = asyncio.Queue()
-        # If parallel allowed, let multiple workers drain the queue (order preserved by FIFO per worker; overall start order is FIFO).
-        self._num_workers: int = 1 if not self.allow_parallel_requests else int(os.environ.get("WORKERS", "4"))
-
     @property
     def pubkey(self) -> Optional[RSA.RsaKey]:
         if self._pubkey is None:
@@ -96,22 +90,6 @@ class Backend:
         timeout = ClientTimeout(total=None)
         return ClientSession(self.model_server_url, timeout=timeout, connector=connector)
 
-    async def _worker(self):
-        while True:
-            handler, request, fut = await self.request_queue.get()
-            try:
-                # Skip if already cancelled while waiting in the queue
-                if fut.cancelled():
-                    continue
-                res = await self.__process_enqueued_request(handler, request)
-                if not fut.cancelled():
-                    fut.set_result(res)
-            except Exception as e:
-                if not fut.cancelled():
-                    fut.set_exception(e)
-            finally:
-                self.request_queue.task_done()
-
     def create_handler(
         self,
         handler: EndpointHandler[ApiPayload_T],
@@ -148,36 +126,7 @@ class Backend:
         handler: EndpointHandler[ApiPayload_T],
         request: web.Request,
     ) -> Union[web.Response, web.StreamResponse]:
-        """use this function to enqueue requests for FIFO processing"""
-        loop = get_running_loop()
-        fut: asyncio.Future = loop.create_future()
-
-        # If the client disconnects while waiting in the FIFO, cancel the future so the worker skips it
-        cancel_watch = create_task(request.wait_for_disconnection())
-        def _cancel_if_disconnected(_):
-            if not fut.done():
-                fut.cancel()
-        cancel_watch.add_done_callback(_cancel_if_disconnected)
-
-        try:
-            await self.request_queue.put((handler, request, fut))
-            return await fut
-        except asyncio.CancelledError:
-            # Propagate cancellation to ensure aiohttp doesn't expect a response body
-            raise
-        finally:
-            # Best-effort cleanup of the watcher
-            cancel_watch.cancel()
-
-    async def __process_enqueued_request(
-        self,
-        handler: EndpointHandler[ApiPayload_T],
-        request: web.Request,
-    ) -> Union[web.Response, web.StreamResponse]:
-        """
-        This contains the original __handle_request logic and is invoked by workers,
-        ensuring FIFO execution via asyncio.Queue.
-        """
+        """use this function to forward requests to the model endpoint"""
         try:
             data = await request.json()
             auth_data, payload = handler.get_data_from_request(data)
@@ -185,11 +134,8 @@ class Backend:
             return web.json_response(data=e.message, status=422)
         except json.JSONDecodeError:
             return web.json_response(dict(error="invalid JSON"), status=422)
-
         workload = payload.count_workload()
-        request_metrics: RequestMetrics = RequestMetrics(
-            request_idx=auth_data.request_idx, reqnum=auth_data.reqnum, workload=workload, status="Created"
-        )
+        request_metrics: RequestMetrics = RequestMetrics(request_idx=auth_data.request_idx, reqnum=auth_data.reqnum, workload=workload, status="Created")
 
         async def cancel_api_call_if_disconnected() -> web.Response:
             await request.wait_for_disconnection()
@@ -230,8 +176,6 @@ class Backend:
         acquired = False
         try:
             self.metrics._request_start(request_metrics)
-
-            # Preserve existing semaphore behavior for serializing requests when requested
             if self.allow_parallel_requests is False:
                 log.debug(f"Waiting to aquire Sem for reqnum:{request_metrics.reqnum}")
                 await self.sem.acquire()
@@ -241,7 +185,6 @@ class Backend:
                 )
             else:
                 log.debug(f"Starting request for reqnum:{request_metrics.reqnum}")
-
             done, pending = await wait(
                 [
                     create_task(make_request()),
@@ -309,14 +252,8 @@ class Backend:
                 self.backend_errored(str(e))
 
     async def _start_tracking(self) -> None:
-        # Start the FIFO workers alongside existing loops
-        worker_tasks = tuple(self._worker() for _ in range(self._num_workers))
         await gather(
-            self.__read_logs(),
-            self.metrics._send_metrics_loop(),
-            self.__healthcheck(),
-            self.metrics._send_delete_requests_loop(),
-            *worker_tasks,
+            self.__read_logs(), self.metrics._send_metrics_loop(), self.__healthcheck(), self.metrics._send_delete_requests_loop()
         )
 
     def backend_errored(self, msg: str) -> None:
@@ -395,26 +332,18 @@ class Backend:
 
             for run in range(1, self.benchmark_handler.benchmark_runs + 1):
                 start = time.time()
-                benchmark_requests = []
+                tasks = []
+                total_workload = 0
 
-                for i in range(concurrent_requests):
+                for _ in range(concurrent_requests):
                     payload = self.benchmark_handler.make_benchmark_payload()
-                    workload = payload.count_workload()
-                    task = self.__call_api(handler=self.benchmark_handler, payload=payload)
-                    benchmark_requests.append(
-                        BenchmarkResult(request_idx=i, workload=workload, task=task)
+                    total_workload += payload.count_workload()
+                    tasks.append(
+                        self.__call_api(handler=self.benchmark_handler, payload=payload)
                     )
 
-                responses = await gather(*[br.task for br in benchmark_requests])
-                for br, response in zip(benchmark_requests, responses):
-                    br.response = response
-
-                total_workload = sum(br.workload for br in benchmark_requests if br.is_successful)
+                responses = await gather(*tasks)
                 time_elapsed = time.time() - start
-                successful_responses = sum([1 for br in benchmark_requests if br.is_successful])
-                if successful_responses == 0:
-                    self.backend_errored("No successful responses from benchmark")
-                    log.debug(f"benchmark failed: {successful_responses}/{concurrent_requests} successful responses")
 
                 throughput = total_workload / time_elapsed
                 sum_throughput += throughput
@@ -428,7 +357,7 @@ class Backend:
                             f"Run: {run}, concurrent_requests: {concurrent_requests}",
                             f"Total workload: {total_workload}, time_elapsed: {time_elapsed}s",
                             f"Throughput: {throughput} workload/s",
-                            f"Successful responses: {successful_responses}/{concurrent_requests}",
+                            f"Successful responses: {len([r for r in responses if r.status == 200])}",
                             "#" * 60,
                         ]
                     )

lib/data_types.py

@@ -3,7 +3,7 @@ import logging
 from dataclasses import dataclass, field
 from enum import Enum
 from abc import ABC, abstractmethod
-from typing import Dict, Any, Union, Tuple, Optional, Set, TypeVar, Generic, Type, Awaitable
+from typing import Dict, Any, Union, Tuple, Optional, Set, TypeVar, Generic, Type
 from aiohttp import web, ClientResponse
 import inspect
 
@@ -206,17 +206,6 @@ class RequestMetrics:
     status: str
     success: bool = False
 
-@dataclass
-class BenchmarkResult:
-    request_idx: int
-    workload: float
-    task: Awaitable[ClientResponse]
-    response: Optional[ClientResponse] = None
-
-    @property
-    def is_successful(self) -> bool:
-        return self.response is not None and self.response.status == 200
-
 @dataclass
 class ModelMetrics:
     """Model specific metrics"""
@@ -257,7 +246,7 @@ class ModelMetrics:
     def wait_time(self) -> float:
         if (len(self.requests_working) == 0):
             return 0.0
-        return sum([request.workload for request in self.requests_working.values()]) / max(self.max_throughput, 0.00001)
+        return sum([request.workload for request in self.requests_working.values()]) / self.max_throughput
     
     @property
     def cur_load(self) -> float:

lib/metrics.py

@@ -145,56 +145,31 @@ class Metrics:
     #######################################Private#######################################
 
     async def __send_delete_requests_and_reset(self):
-        async def post(report_addr: str, idxs: list[int], success_flag: bool) -> bool:
+
+        async def send_data(report_addr: str, success: bool) -> bool:
             data = {
                 "worker_id": self.id,
-                "request_idxs": idxs,
-                "success": success_flag,
+                "request_idxs": [r.request_idx for r in self.model_metrics.requests_deleting if r.success == success],
+                "success": success
             }
-            log.debug(
-                f"Deleting requests that {'succeeded' if success_flag else 'failed'}: {data['request_idxs']}"
-            )
             full_path = report_addr.rstrip("/") + "/delete_requests/"
             for attempt in range(1, 4):
                 try:
                     session = await self.http()
                     async with session.post(full_path, json=data) as res:
-                        log.debug(f"delete_requests response: {res.status}")
                         res.raise_for_status()
                     return True
                 except asyncio.TimeoutError:
-                    log.debug("delete_requests timed out")
+                    log.debug(f"delete_requests timed out")
                 except (ClientResponseError, Exception) as e:
                     log.debug(f"delete_requests failed with error: {e}")
                 await asyncio.sleep(2)
                 log.debug(f"retrying delete_request, attempt: {attempt}")
-            return False
-
-        # Take a snapshot of what we plan to send this tick.
-        # New arrivals after this snapshot will remain in the queue for the next tick.
-        snapshot = list(self.model_metrics.requests_deleting)
-        success_idxs = [r.request_idx for r in snapshot if r.success is True]
-        failed_idxs  = [r.request_idx for r in snapshot if r.success is False]
-
-        if not success_idxs and not failed_idxs:
-            return  # nothing to do
 
         for report_addr in self.report_addr:
-            sent_success = True
-            sent_failed  = True
-
-            if success_idxs:
-                sent_success = await post(report_addr, success_idxs, True)
-            if failed_idxs:
-                sent_failed = await post(report_addr, failed_idxs, False)
-
-            if sent_success and sent_failed:
-                # Remove only the items we actually sent from the live queue.
-                sent_set = set(success_idxs) | set(failed_idxs)
-                self.model_metrics.requests_deleting[:] = [
-                    r for r in self.model_metrics.requests_deleting
-                    if r.request_idx not in sent_set
-                ]
+            success = await send_data(report_addr, success=True) and await send_data(report_addr, success=False)
+            if success is True:
+                self.model_metrics.requests_deleting.clear()
                 break
 
 

workers/comfyui-json/README.md

@@ -12,21 +12,9 @@ A docker image is provided but you may use any if the above requirements are met
 
 ## Benchmarking
 
-### Custom Benchmark Workflows
+A simple image generation benchmark runs when each worker initializes to validate GPU performance and identify underperforming machines.
 
-You can provide a custom ComfyUI workflow for benchmarking by creating `workers/comfyui-json/misc/benchmark.json`. This allows you to test performance using your preferred models and workflow complexity.
-
-**Ways to provide the benchmark file:**
-- Fork this repository and add your `benchmark.json` file
-- Write the file during worker provisioning (onstart script or setup phase)
-
-An example file is provided in the repository. To ensure varied generations, use the placeholder `__RANDOM_INT__` in place of static seed values - it will be replaced with a random integer for each benchmark run.
-
-### Default Benchmark (Fallback)
-
-If `benchmark.json` is not available, a simple image generation benchmark runs when each worker initializes. This validates GPU performance and helps identify underperforming machines.
-
-The default benchmark uses Stable Diffusion v1.5 with ComfyUI's standard text-to-image workflow. Configure it using these environment variables:
+The benchmark uses Stable Diffusion v1.5 with ComfyUI's default text-to-image workflow. Configure the benchmark complexity and duration using these variables:
 
 | Environment Variable | Default Value | Description |
 | -------------------- | ------------- | ----------- |
@@ -36,7 +24,7 @@ The default benchmark uses Stable Diffusion v1.5 with ComfyUI's standard text-to
 
 Each benchmark run uses a random prompt from `misc/test_prompts.txt` and a random seed to ensure consistent GPU load patterns.
 
-#### Calibrating Fallback Benchmark Duration
+### Calibrating Benchmark Duration
 
 To screen for underperforming hardware, set `BENCHMARK_TEST_STEPS` to match your expected production workflow duration. This allows you to identify machines that won't meet performance requirements.
 

workers/comfyui-json/data_types.py

@@ -5,13 +5,12 @@ import dataclasses
 from typing import Dict, Any
 from functools import cache
 from math import ceil
-from pathlib import Path
-import json
-import logging
 
 from lib.data_types import ApiPayload, JsonDataException
 
-log = logging.getLogger(__file__)
+
+with open("workers/comfyui/misc/test_prompts.txt", "r") as f:
+    test_prompts = f.readlines()
 
 def count_workload() -> float:
     # Always 100.0 where there is a single instance of ComfyUI handling requests
@@ -25,32 +24,9 @@ class ComfyWorkflowData(ApiPayload):
     @classmethod
     def for_test(cls):
         """
-        If the user has provided a benchmark workflow we can use it here to properly gauge performance.
-        Otherwise, use the variables available to simulate workflows of the required running time
+        Use the variables available to simulate workflows of the required running time
         Example: SD1.5, simple image gen 10000 steps, 512px x 512px will run for approximately 9 minutes @ ~18 it/s (RTX 4090)
         """
-        # Try to load benchmark.json
-        benchmark_file = Path("workers/comfyui-json/misc/benchmark.json")
-        
-        if benchmark_file.exists():
-            try:
-                with open(benchmark_file, "r") as f:
-                    benchmark_workflow = json.load(f)
-                return cls(
-                    input={
-                        "request_id": f"test-{random.randint(1000, 99999)}",
-                        "workflow_json": benchmark_workflow
-                    }
-                )
-            except (json.JSONDecodeError, IOError):
-                # JSON is malformed or file can't be read, fall through to default
-                log.error(f"Failed to benchmark using {benchmark_file}")
-        
-        # Fallback: read prompts and construct payload
-        log.info("Using fallback method for benchmarking")
-        with open("workers/comfyui-json/misc/test_prompts.txt", "r") as f:
-            test_prompts = f.readlines()
-        
         test_prompt = random.choice(test_prompts).rstrip()
         return cls(
             input={

workers/comfyui-json/misc/benchmark.json.example

@@ -1,107 +0,0 @@
-{
-    "3": {
-        "inputs": {
-            "seed": "__RANDOM_INT__",
-            "steps": 20,
-            "cfg": 8,
-            "sampler_name": "euler",
-            "scheduler": "normal",
-            "denoise": 1,
-            "model": [
-            "4",
-            0
-            ],
-            "positive": [
-            "6",
-            0
-            ],
-            "negative": [
-            "7",
-            0
-            ],
-            "latent_image": [
-            "5",
-            0
-            ]
-        },
-        "class_type": "KSampler",
-        "_meta": {
-            "title": "KSampler"
-        }
-    },
-    "4": {
-        "inputs": {
-            "ckpt_name": "v1-5-pruned-emaonly-fp16.safetensors"
-        },
-        "class_type": "CheckpointLoaderSimple",
-        "_meta": {
-            "title": "Load Checkpoint"
-        }
-    },
-    "5": {
-        "inputs": {
-            "width": 512,
-            "height": 512,
-            "batch_size": 1
-        },
-        "class_type": "EmptyLatentImage",
-        "_meta": {
-            "title": "Empty Latent Image"
-        }
-    },
-    "6": {
-        "inputs": {
-            "text": "beautiful scenery nature glass bottle landscape, , purple galaxy bottle,",
-            "clip": [
-            "4",
-            1
-            ]
-        },
-        "class_type": "CLIPTextEncode",
-        "_meta": {
-            "title": "CLIP Text Encode (Prompt)"
-        }
-    },
-    "7": {
-        "inputs": {
-            "text": "text, watermark",
-            "clip": [
-            "4",
-            1
-            ]
-        },
-        "class_type": "CLIPTextEncode",
-        "_meta": {
-            "title": "CLIP Text Encode (Prompt)"
-        }
-    },
-    "8": {
-        "inputs": {
-            "samples": [
-            "3",
-            0
-            ],
-            "vae": [
-            "4",
-            2
-            ]
-        },
-        "class_type": "VAEDecode",
-        "_meta": {
-            "title": "VAE Decode"
-        }
-    },
-    "9": {
-        "inputs": {
-            "filename_prefix": "ComfyUI",
-            "images": [
-            "8",
-            0
-            ]
-        },
-        "class_type": "SaveImage",
-        "_meta": {
-            "title": "Save Image"
-        }
-    }
-}

workers/comfyui-json/server.py

@@ -19,7 +19,6 @@ MODEL_SERVER_START_LOG_MSG = "To see the GUI go to: "
 MODEL_SERVER_ERROR_LOG_MSGS = [
     "MetadataIncompleteBuffer",  # This error is emitted when the downloaded model is corrupted
     "Value not in list: ",  # This error is emitted when the model file is not there at all
-    "[ERROR] Provisioning Script failed", # Error inserted by provisioning script if models/nodes fail to download
 ]
 
 

workers/openai/data_types/server.py

@@ -119,25 +119,14 @@ class GenericHandler(EndpointHandler[GenericData], ABC):
 class CompletionsData(GenericData):
     @classmethod
     def for_test(cls) -> "CompletionsData":
-        system_prompt = """You are a helpful AI assistant. You have access to the following knowledge base:
-    
-        Zebras (US: /ˈziːbrəz/, UK: /ˈzɛbrəz, ˈziː-/)[2] (subgenus Hippotigris) are African equines 
-        with distinctive black-and-white striped coats. There are three living species: Grévy's zebra 
-        (Equus grevyi), the plains zebra (E. quagga), and the mountain zebra (E. zebra). Zebras share the 
-        genus Equus with horses and asses, the three groups being the only living members of the family 
-        Equidae. Zebra stripes come in different patterns, unique to each individual. Zebras inhabit eastern 
-        and southern Africa and can be found in a variety of habitats such as savannahs, grasslands, 
-        woodlands, shrublands, and mountainous areas.
-        
-        Please answer the following question based on the above context."""
-        unique_question = " ".join(random.choices(WORD_LIST, k=int(100)))
+        prompt = " ".join(random.choices(WORD_LIST, k=int(250)))
         model = os.environ.get("MODEL_NAME")
         if not model:
             raise ValueError("MODEL_NAME environment variable not set")
 
         test_input = {
             "model": model,
-            "prompt": f"{system_prompt}\n\n{unique_question}",
+            "prompt": prompt,
             "temperature": 0.7,
             "max_tokens": 500,
         }
@@ -164,18 +153,7 @@ class ChatCompletionsData(GenericData):
 
     @classmethod
     def for_test(cls) -> "ChatCompletionsData":
-        system_prompt = """You are a helpful AI assistant. You have access to the following knowledge base:
-    
-        Zebras (US: /ˈziːbrəz/, UK: /ˈzɛbrəz, ˈziː-/)[2] (subgenus Hippotigris) are African equines 
-        with distinctive black-and-white striped coats. There are three living species: Grévy's zebra 
-        (Equus grevyi), the plains zebra (E. quagga), and the mountain zebra (E. zebra). Zebras share the 
-        genus Equus with horses and asses, the three groups being the only living members of the family 
-        Equidae. Zebra stripes come in different patterns, unique to each individual. Zebras inhabit eastern 
-        and southern Africa and can be found in a variety of habitats such as savannahs, grasslands, 
-        woodlands, shrublands, and mountainous areas.
-        
-        Please answer the following question based on the above context."""
-        unique_question = " ".join(random.choices(WORD_LIST, k=int(100)))
+        prompt = " ".join(random.choices(WORD_LIST, k=int(250)))
         model = os.environ.get("MODEL_NAME")
         if not model:
             raise ValueError("MODEL_NAME environment variable not set")
@@ -183,10 +161,7 @@ class ChatCompletionsData(GenericData):
         # Chat completions use messages format instead of prompt
         test_input = {
             "model": model,
-            "messages": [
-                {"role": "system", "content": system_prompt},  # Shared prefix
-                {"role": "user", "content": unique_question}   # Unique per request
-            ],
+            "messages": [{"role": "user", "content": prompt}],
             "temperature": 0.7,
             "max_tokens": 500,
         }

workers/openai/test_load.py

@@ -82,7 +82,6 @@ def do_one(endpoint_name: str,
         # 1) Check if we got a worker back from route
         worker_url = msg.get("url", "")
         if not worker_url:
-            status = msg.get("status", "")
             m = re.search(r"total workers:\s*(\d+).*loading workers:\s*(\d+).*standby workers:\s*(\d+).*error workers:\s*(\d+)", status, re.I | re.S)
             if m:
                 tot, loading, standby, err = map(int, m.groups())