Performance Engineering and Cost Optimization Questions

import asyncio
import time
from typing import Any, List, NamedTuple

class Req(NamedTuple):
    payload: Any
    future: asyncio.Future
    enqueue_ts: float
    deadline_ts: float

class DynamicBatcher:
    def __init__(self, max_batch_size: int, max_wait_ms: int, max_hold_ms: int=50):
        self.max_batch_size = max_batch_size
        self.max_wait_s = max_wait_ms / 1000.0
        self.max_hold_s = max_hold_ms / 1000.0
        self.queue: asyncio.Queue[Req] = asyncio.Queue()
        self.flush_event = asyncio.Event()
        self._flusher_task = asyncio.create_task(self._flusher())

    async def submit(self, payload, per_request_max_wait_ms=None):
        fut = asyncio.get_event_loop().create_future()
        now = time.monotonic()
        per_wait = (per_request_max_wait_ms/1000.0) if per_request_max_wait_ms else self.max_wait_s
        req = Req(payload, fut, now, now + per_wait)
        await self.queue.put(req)
        # notify flusher
        self.flush_event.set()
        return await fut

    async def _flusher(self):
        while True:
            await self.flush_event.wait()
            self.flush_event.clear()
            batch: List[Req] = []
            start = time.monotonic()
            # collect up to max_batch_size while honoring deadlines
            while len(batch) < self.max_batch_size:
                try:
                    timeout = self._next_timeout(batch, start)
                    req = await asyncio.wait_for(self.queue.get(), timeout=timeout)
                    batch.append(req)
                    # if full, break to process immediately
                    if len(batch) >= self.max_batch_size:
                        break
                except asyncio.TimeoutError:
                    break

            if not batch:
                continue

            # ensure FIFO and that no req misses deadline
            now = time.monotonic()
            expired = [r for r in batch if r.deadline_ts <= now]
            # If any expired, we flush only those plus earliest others up to max_batch_size
            # Simple policy: keep FIFO order, but always process when oldest hits deadline
            to_process = batch[:self.max_batch_size]
            # perform inference asynchronously (simulate)
            results = await self._run_inference([r.payload for r in to_process])
            for r, res in zip(to_process, results):
                if not r.future.done():
                    r.future.set_result(res)
            # requeue leftovers if any (shouldn't happen here because queue.get removed them)
            # If implementation removed items beyond to_process, they remain taken; in production use peek or separate buffer

    def _next_timeout(self, current_batch, start_ts):
        # compute time remaining until oldest deadline in queue or max_hold
        now = time.monotonic()
        try:
            oldest: Req = self.queue._queue[0]  # direct access for estimation; OK in pseudocode
            time_until_deadline = max(0.0, oldest.deadline_ts - now)
        except Exception:
            time_until_deadline = self.max_hold_s
        time_until_hold = max(0.0, self.max_hold_s - (now - start_ts))
        return min(time_until_deadline, time_until_hold)

    async def _run_inference(self, batch_payloads: List[Any]):
        await asyncio.sleep(0.01)  # placeholder for GPU call
        return [{"out": p} for p in batch_payloads]

import math
from typing import List

class LatencyTracker:
    def __init__(self):
        # store raw latencies
        self._latencies: List[float] = []
    
    def add(self, latency_ms: float) -> None:
        """Record a new latency (constant time)."""
        # Optionally validate input
        if not isinstance(latency_ms, (int, float)):
            raise TypeError("latency_ms must be a number")
        self._latencies.append(float(latency_ms))
    
    def p95(self) -> float:
        """Return the 95th percentile using the nearest-rank method.
        Raises ValueError if no latencies recorded.
        """
        n = len(self._latencies)
        if n == 0:
            raise ValueError("No latencies recorded")
        # sort a copy so we don't disturb insertion order
        sorted_vals = sorted(self._latencies)
        # nearest-rank: rank = ceil(0.95 * n)
        rank = math.ceil(0.95 * n)
        # convert to 0-based index, clamp to last element
        idx = max(0, min(n - 1, rank - 1))
        return sorted_vals[idx]

import functools
import random
import time
import requests

def retry_backoff(max_retries=5, initial_delay_ms=100, max_delay_ms=10000, jitter=True):
    """
    Decorator to retry a function (typically network call) with exponential backoff.
    Stops retrying on 4xx HTTP client errors (except 429 which is retried).
    """
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            attempt = 0
            delay = initial_delay_ms / 1000.0
            while True:
                try:
                    return fn(*args, **kwargs)
                except requests.HTTPError as e:
                    status = getattr(e.response, "status_code", None)
                    # Do not retry on 4xx (client errors) except 429 (Too Many Requests)
                    if status and 400 <= status < 500 and status != 429:
                        raise
                    # fallthrough to retry for 429 or 5xx
                    attempt += 1
                except (requests.ConnectionError, requests.Timeout) as e:
                    attempt += 1
                # If exceeded retries, re-raise last exception
                if attempt > max_retries:
                    raise
                # compute next delay with exponential backoff and optional jitter
                next_delay = min(delay * (2 ** (attempt - 1)), max_delay_ms / 1000.0)
                if jitter:
                    # full jitter: random between 0 and next_delay
                    next_delay = random.uniform(0, next_delay)
                time.sleep(next_delay)
        return wrapper
    return decorator