Triton Inference Server 动态批处…

import randomfrom collections import deque class DynamicBatcher:    def __init__(self, max_batch=16, timeout_ms=50, max_q=64):        self.max_batch = max_batch        self.timeout = timeout_ms / 1000        self.max_q = max_q        self.q = deque()        self.n_req = 0        self.n_batch = 0     def submit(self, t):        if len(self.q) >= self.max_q:            return False        self.q.append(t)        self.n_req += 1        return True     def process(self, now):        if not self.q:            return 0        b = []        while self.q and len(b) < self.max_batch:            b.append(self.q.popleft())        self.n_batch += 1        return len(b)     def run(self, arrivals):        arrivals.sort()        i, now = 0, 0.0        while i < len(arrivals) or self.q:            while i < len(arrivals) and arrivals[i] <= now:                self.submit(arrivals[i])                i += 1            if self.q:                if now >= self.q[0] + self.timeout or len(self.q) >= self.max_batch:                    self.process(now)                    now += 0.01                    continue            if i < len(arrivals):                now = arrivals[i]            elif self.q:                now += 0.001        return {"reqs": self.n_req, "batches": self.n_batch,                "avg_batch": round(self.n_req / max(self.n_batch, 1), 2)} random.seed(42)arr = [i*0.01 + random.uniform(0, 0.005) for i in range(200)]for mb, tm in [(8,20),(16,50),(32,30)]:    b = DynamicBatcher(max_batch=mb, timeout_ms=tm)    print(f"batch={mb} timeout={tm}ms: {b.run(list(arr))}")