Performance Optimization Under Resource Constraints Questions

import threading, time, statistics, queue, grpc, random
import psutil
from collections import defaultdict
from datetime import datetime
# optional: pip install nvidia-ml-py3
try:
    import pynvml
    pynvml.nvmlInit()
    has_gpu = True
except:
    has_gpu = False

# --- user config
TARGET_QPS = 100
DURATION = 30  # seconds (measurement after warmup)
WARMUP = 5
WORKERS = 10
ADDRESS = "localhost:50051"

# --- simple gRPC stub call placeholder (replace with real generated stub)
def call_inference(stub, payload):
    # example: return stub.Predict(payload, timeout=5)
    time.sleep(0.01)  # simulate network+inference
    return {"ok": True}

# --- util to sample CPU/GPU
cpu_samples, gpu_samples = [], []
def sampler(stop_evt):
    while not stop_evt.is_set():
        cpu_samples.append(psutil.cpu_percent(interval=0.5))
        if has_gpu:
            h = pynvml.nvmlDeviceGetHandleByIndex(0)
            util = pynvml.nvmlDeviceGetUtilizationRates(h).gpu
            gpu_samples.append(util)

# --- worker
latencies = []
lat_lock = threading.Lock()
def worker(stop_evt, start_time):
    # optional: create stub per thread
    stub = None
    inter_arrival = 1.0 / (TARGET_QPS / WORKERS)
    while time.time() < start_time + WARMUP + DURATION:
        now = time.time()
        # only measure during measurement window
        if now >= start_time + WARMUP:
            t0 = time.perf_counter()
            call_inference(stub, {})
            t1 = time.perf_counter()
            with lat_lock:
                latencies.append((t1 - t0)*1000)  # ms
        else:
            call_inference(stub, {})
        time.sleep(inter_arrival * (0.8 + 0.4*random.random()))  # jitter

# --- run
def run():
    stop_evt = threading.Event()
    start_time = time.time()
    sampler_thread = threading.Thread(target=sampler, args=(stop_evt,))
    sampler_thread.start()
    threads = [threading.Thread(target=worker, args=(stop_evt, start_time)) for _ in range(WORKERS)]
    for t in threads: t.start()
    for t in threads: t.join()
    stop_evt.set()
    sampler_thread.join()

    # compute percentiles
    if not latencies:
        print("No samples")
        return
    lat_ms = sorted(latencies)
    def pct(p): return lat_ms[int(len(lat_ms)*p/100)]
    print("p50", pct(50), "p95", pct(95), "p99", pct(99))
    print("CPU median", statistics.median(cpu_samples))
    if has_gpu: print("GPU median", statistics.median(gpu_samples))

if __name__ == "__main__":
    random.seed(42)
    run()

torch.onnx.export(model.eval(), dummy_input,
                  "model.onnx", opset_version=14,
                  input_names=["input"], output_names=["output"],
                  dynamic_axes={"input":[0], "output":[0]})

python -m onnx.checker model.onnx
python -m onnxsim model.onnx model_simpl.onnx

trtexec --onnx=model_simpl.onnx --saveEngine=model.engine --fp16 --shapes=input:1x3x224x224

sess = onnxruntime.InferenceSession("model_simpl.onnx", providers=["CUDAExecutionProvider"])

np.max(np.abs(out_orig - out_acc))

import time
import requests
from collections import OrderedDict
from typing import Any, Dict

class PredictionCache:
    def __init__(self, max_items=1000):
        self.cache = OrderedDict()
        self.max_items = max_items

    def get(self, key):
        if key in self.cache:
            self.cache.move_to_end(key)
            return self.cache[key]
        return None

    def set(self, key, value):
        self.cache[key] = value
        self.cache.move_to_end(key)
        if len(self.cache) > self.max_items:
            self.cache.popitem(last=False)

class ResilientModelClient:
    def __init__(self, base_url: str, timeout=1.0, max_retries=4, backoff_factor=0.5, cache=None):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.cache = cache or PredictionCache()

    def _predict_request(self, payload: Dict[str, Any]):
        return requests.post(f"{self.base_url}/predict", json=payload, timeout=self.timeout)

    def predict(self, input_features: Dict[str, Any], allow_stale=True):
        key = str(sorted(input_features.items()))
        # Try network with retries + exponential backoff
        attempt = 0
        while attempt <= self.max_retries:
            try:
                resp = self._predict_request(input_features)
                if resp.status_code == 200:
                    result = resp.json()
                    self.cache.set(key, result)
                    return {"result": result, "source": "server", "attempts": attempt+1}
                # treat 5xx as transient
                if 500 <= resp.status_code < 600:
                    raise requests.HTTPError(f"Server error {resp.status_code}")
                # 4xx -> don't retry
                return {"error": resp.text, "status": resp.status_code, "source": "server"}
            except (requests.Timeout, requests.ConnectionError, requests.HTTPError) as e:
                attempt += 1
                if attempt > self.max_retries:
                    break
                backoff = self.backoff_factor * (2 ** (attempt - 1))
                time.sleep(backoff)
        # Fallback to cache if allowed
        cached = self.cache.get(key)
        if cached is not None and allow_stale:
            return {"result": cached, "source": "cache", "stale": True}
        return {"error": "unavailable", "source": "client"}

# Example usage:
# client = ResilientModelClient("https://models.example.com", timeout=0.5)
# out = client.predict({"text": "hello"}, allow_stale=True)