""" LLM Inference & Serving Stack - minimal runnable implementation ================================================================ Pure-stdlib (Python 3.9+, no torch/numpy) sanity checks, runs in <1s on CPU. Pairs with: docs/tutorials/llm_inference_serving_tutorial.md (concept reference). [D01] sampling operators: exact support-set math for greedy/temperature/ top-k/top-p/min-p/typical + penalties. Split into two kinds of functions: "compute filtered/renormalized distribution" (pure, deterministic -- almost every assert below is on THIS half) vs. "draw a token" (the only function that touches an RNG). [D02] filter-order non-commutativity: top_k->top_p vs top_p->top_k produce different support sets from the SAME distribution. [D03] streaming stop-sequence detector: a stop string can span multiple token chunks; the buffer must never leak an unsafe partial prefix. [D04] static vs. continuous batching simulator: exact A/B/C completion times + scheduler-capacity utilization, plus an equal-length counterexample showing continuous batching does NOT strictly dominate every workload, plus generic scheduling invariants. [D05] chunked-prefill toy model: reduces max decode-service gap (ITL) at the cost of the long prompt's own TTFT -- not a free lunch. [D06] KV-cache per-token bytes (reuses the KV-cache sibling tutorial's own formula verbatim; no re-derivation of MQA/GQA/MLA here). [D07] block allocator: logical vs. allocated slots/bytes (internal fragmentation from block-size rounding). [D08] prefix-sharing physical block count + the "don't divide KV bytes by TP degree unless heads divide evenly" caveat. [D09] (optional) roofline ridge point as an upper-bound sanity number, explicitly NOT a latency predictor. Run: python3 inference_serving.py """ import math import random from collections import namedtuple # ============================================================ # [D01]/[D02] Sampling: logits -> filtered distribution -> draw # ============================================================ # Convention pinned to the tutorial's CORRECTIONS section: # - greedy() is a SEPARATE branch from temperature sampling; T=0 must # never be passed into z_i / T. # - every filter below takes an already-computed probability vector # `probs` (i.e. temperature has already been applied) and returns a # new, renormalized probability vector with zero mass outside its # support set. None of them touch randomness. # - sample() is the ONLY function below that calls the RNG. # - every support-set threshold below is an EXACT `>=` comparison against # the caller-supplied parameter -- never a fudged/lowered threshold. # Cumulative sums use math.fsum() for precision, but the comparison # itself is never loosened by an epsilon; that would silently redefine # which support set the filter computes. def _validate_logits(logits): if len(logits) == 0: raise ValueError("logits must be non-empty") for z in logits: if isinstance(z, float) and math.isnan(z): raise ValueError("logits must not contain NaN") if all(z == -math.inf for z in logits): raise ValueError("logits must not be all -inf (no finite/legal token)") def _validate_probs(probs): if len(probs) == 0: raise ValueError("probs must be non-empty") for p in probs: if isinstance(p, float) and math.isnan(p): raise ValueError("probs must not contain NaN") if p < 0: raise ValueError("probs must be non-negative") def softmax(logits): _validate_logits(logits) m = max(logits) exps = [math.exp(z - m) for z in logits] s = sum(exps) return [e / s for e in exps] def greedy(logits): """argmax with a DETERMINISTIC tie-break rule: lowest index wins ties. (T=0 must go through this branch, never through temperature_probs.)""" _validate_logits(logits) best_i, best_v = 0, logits[0] for i, v in enumerate(logits): if v > best_v: best_i, best_v = i, v return best_i def temperature_probs(logits, T): """p_i(T) = softmax(z_i / T). Defined ONLY for finite T > 0. Computes (z_i - z_max) / T directly -- NOT z_i / T followed by a SECOND max-subtraction inside softmax(). Dividing by T first means an extreme logit combined with a very small T can overflow to +-inf (or produce inf - inf = nan) before the max-shift ever gets a chance to cancel it; subtracting z_max in logit space FIRST keeps every exponent <= 0. """ _validate_logits(logits) if not math.isfinite(T) or T <= 0: raise ValueError( f"temperature must be finite and > 0; got {T} (call greedy() for T=0)") zmax = max(logits) exps = [math.exp((z - zmax) / T) for z in logits] total = math.fsum(exps) return [e / total for e in exps] def top_k_filter(probs, k): """Support S_k = the k highest-probability tokens (ties broken by lower index, for a deterministic ranking). Positive-temperature scaling does not change the ranking, so S_k by itself does not depend on T>0. k must satisfy 1 <= k <= len(probs): k=0 would legally return the all-zero vector, k<0 would silently trigger Python's negative-slice semantics on `order[:k]`, and k>len(probs) can never produce a support set of size exactly k -- all three are rejected outright.""" _validate_probs(probs) n = len(probs) if k != int(k) or not (1 <= k <= n): raise ValueError(f"k must be an integer with 1 <= k <= {n} (len(probs)), got {k!r}") k = int(k) order = sorted(range(n), key=lambda i: (-probs[i], i)) support = set(order[:k]) mass = sum(probs[i] for i in support) return [probs[i] / mass if i in support else 0.0 for i in range(n)] def top_p_filter(probs, p0): """Sort descending; take the SMALLEST prefix whose cumulative mass first reaches p0 (the boundary token is KEPT). NOT "all p_i >= p0" and NOT "cumulative strictly < p0" -- both of those can wrongly drop the boundary token. p0 must satisfy 0 < p0 <= 1. The cumulative sum used for the `>= p0` test is recomputed via math.fsum() over the tokens included so far on every step (rather than a running `cum += p`) to keep the ONE comparison that matters -- reaching the exact threshold -- as precise as floating point allows; the threshold itself is never lowered by an epsilon, which would silently stop one token too early.""" _validate_probs(probs) if not (0.0 < p0 <= 1.0): raise ValueError(f"p0 must satisfy 0 < p0 <= 1, got {p0}") order = sorted(range(len(probs)), key=lambda i: (-probs[i], i)) support, included = set(), [] for i in order: support.add(i) included.append(probs[i]) if math.fsum(included) >= p0: break mass = sum(probs[i] for i in support) return [probs[i] / mass if i in support else 0.0 for i in range(len(probs))] def min_p_filter(probs, alpha): """S = {i : p_i >= alpha * max_j p_j}, renormalized within S. alpha must satisfy 0 < alpha <= 1: alpha<=0 would keep every token (min-p degenerating to a no-op, and disagreeing with the logit-space formula below where log(alpha) is undefined at alpha<=0), and alpha>1 can produce an EMPTY support set (no p_i can exceed alpha*p_max when alpha>1) -- silently returning an all-zero, non-normalized "distribution" instead of raising, which would only surface downstream as a confusing failure the next time something tries to sample from it.""" _validate_probs(probs) if not (0.0 < alpha <= 1.0): raise ValueError(f"alpha must satisfy 0 < alpha <= 1, got {alpha}") pmax = max(probs) support = {i for i, p in enumerate(probs) if p >= alpha * pmax} mass = sum(probs[i] for i in support) return [probs[i] / mass if i in support else 0.0 for i in range(len(probs))] def min_p_support_from_logits(logits, T, alpha): """Equivalent logit-space condition when p_i = softmax(z_i/T): z_i >= z_max + T*log(alpha). This support set MOVES with T -- it is not a fixed logit-gap threshold. Requires finite T > 0 (the derivation assumes softmax(z/T), which is undefined for T<=0) and 0 < alpha <= 1 (same domain as min_p_filter, so the two independent derivations of the SAME support set stay consistent with each other).""" _validate_logits(logits) if not math.isfinite(T) or T <= 0: raise ValueError(f"temperature must be finite and > 0, got {T}") if not (0.0 < alpha <= 1.0): raise ValueError(f"alpha must satisfy 0 < alpha <= 1, got {alpha}") zmax = max(logits) thresh = zmax + T * math.log(alpha) return {i for i, z in enumerate(logits) if z >= thresh} def typical_filter(probs, typical_p): """H(p) = -sum p_i log p_i; d_i = |-log p_i - H(p)|; sort by d_i ascending (closest to the entropy first); take the smallest prefix in THAT order whose cumulative mass reaches typical_p. The resulting support need not be a contiguous prefix of the probability-sorted order -- see the D01 assertions below. typical_p must satisfy 0 < typical_p <= 1, mirroring top_p_filter's p0 domain; the cumulative-mass comparison is likewise an exact `>=` over an math.fsum()-recomputed running total, never epsilon-loosened.""" _validate_probs(probs) if not (0.0 < typical_p <= 1.0): raise ValueError(f"typical_p must satisfy 0 < typical_p <= 1, got {typical_p}") H = -sum(p * math.log(p) for p in probs if p > 0) d = [abs(-math.log(p) - H) if p > 0 else math.inf for p in probs] order = sorted(range(len(probs)), key=lambda i: (d[i], i)) support, included = set(), [] for i in order: support.add(i) included.append(probs[i]) if math.fsum(included) >= typical_p: break mass = sum(probs[i] for i in support) return [probs[i] / mass if i in support else 0.0 for i in range(len(probs))] def freq_presence_penalty(logits, counts, alpha_f, alpha_p): """z_i' = z_i - alpha_f*c_i - alpha_p*1[c_i>0]. `counts` semantics (whether prompt tokens are counted) is an API-level choice, not a universal constant -- the caller decides what goes into `counts`.""" return [z - alpha_f * counts[i] - (alpha_p if counts[i] > 0 else 0.0) for i, z in enumerate(logits)] def repetition_penalty(logits, seen, r): """Sign-aware penalty, r > 1, applied ONLY to already-seen tokens: z_i/r if z_i>0 else z_i*r. This is NOT "subtract a constant from every seen token" -- the direction depends on the sign of z_i.""" assert r > 1 out = [] for i, z in enumerate(logits): if i in seen: out.append(z / r if z > 0 else z * r) else: out.append(z) return out def apply_hard_mask(logits, allowed): """Constrained decoding primitive: illegal tokens -> -inf logit, then a normal softmax renormalizes over the legal ones.""" return [z if allowed[i] else -math.inf for i, z in enumerate(logits)] class StubRNG: """Deterministic stand-in for random.Random, for tests that need an EXACT `u` value (0.0, an exact cumulative boundary, a value close to 1) rather than a statistical frequency check over many draws.""" def __init__(self, values): self._values = list(values) self._i = 0 def random(self): v = self._values[self._i] self._i += 1 return v def sample(probs, rng): """The ONLY function in this module that touches randomness. u ~ Uniform[0, 1). Uses `u < cum` (NOT `u <= cum`): with `<=`, u==0.0 would select index 0 even when probs[0] == 0.0 (a token filtered out by top-k/top-p/min-p/typical has probability EXACTLY 0, and must never be drawn). The fallback -- needed only when floating-point summation leaves the true final cumulative sum fractionally below 1.0 -- returns the LAST token with POSITIVE probability, never an unconditional len(probs)-1: that trailing index may itself have been filtered to zero mass.""" positive = [i for i, p in enumerate(probs) if p > 0] if not positive: raise ValueError("sample() requires at least one token with positive probability") u = rng.random() cum = 0.0 for i, p in enumerate(probs): cum += p if u < cum: return i return positive[-1] def run_d01(): # --- greedy: no softmax needed, deterministic tie-break --- assert greedy([5, 5, 1]) == 0 # lowest-index tie-break assert greedy([1, 5, 5]) == 1 for bad_T in (0.0, -1.0, float("inf"), float("nan")): try: temperature_probs([1.0, 2.0], bad_T) raise AssertionError(f"temperature_probs must reject T={bad_T}") except ValueError: pass # --- degenerate logits/probs inputs: empty / NaN / all -inf must raise # a clear ValueError, never silently produce nan or garbage output --- for bad_fn, args in ( (softmax, ([],)), (greedy, ([],)), (temperature_probs, ([], 1.0)), (top_k_filter, ([], 1)), (top_p_filter, ([], 0.5)), (min_p_filter, ([], 0.5)), (typical_filter, ([], 0.5)), (min_p_support_from_logits, ([], 1.0, 0.5)), ): try: bad_fn(*args) raise AssertionError(f"{bad_fn.__name__} must reject empty input") except ValueError: pass try: softmax([1.0, float("nan"), 2.0]) raise AssertionError("softmax must reject NaN logits") except ValueError: pass try: softmax([-math.inf, -math.inf]) raise AssertionError("softmax must reject all -inf logits (no legal token)") except ValueError: pass try: temperature_probs([-math.inf, -math.inf], 1.0) raise AssertionError("temperature_probs must reject all -inf logits") except ValueError: pass try: top_p_filter([0.5, float("nan"), 0.5], 0.5) raise AssertionError("top_p_filter must reject NaN probabilities") except ValueError: pass # --- temperature: T->0+ limit concentrates on argmax (unique max) --- unique_logits = [5.0, 3.0, 1.0] p_tiny = temperature_probs(unique_logits, 1e-4) assert p_tiny[0] > 0.999999 # concentrates on argmax # --- tied max logits: math limit SPLITS mass among ties, not one-hot --- tied_logits = [5.0, 5.0, 1.0] p_tied = temperature_probs(tied_logits, 1e-4) assert abs(p_tied[0] - 0.5) < 1e-4 and abs(p_tied[1] - 0.5) < 1e-4 assert p_tied[2] < 1e-6 # --- lower T sharpens (raises max prob) for a fixed logits vector --- p_hot = temperature_probs(unique_logits, 2.0) p_cold = temperature_probs(unique_logits, 0.5) assert max(p_cold) > max(p_hot) # --- top-k: exact support size, excludes anything ranked below k, # stable tie-break for equal logits --- probs5 = softmax([4.0, 3.0, 2.0, 1.0, 0.0]) for k in (1, 2, 3): filt = top_k_filter(probs5, k) support = {i for i, p in enumerate(filt) if p > 0} assert len(support) == k assert max(support) == k - 1 # never keeps a lower-ranked token assert abs(sum(filt) - 1.0) < 1e-12 assert all(p >= 0 for p in filt) if k >= 2: # relative proportions WITHIN the support must match the ORIGINAL # distribution -- catches a broken renormalization that flattens # the support to a uniform distribution instead of rescaling it. assert abs(filt[0] / filt[1] - probs5[0] / probs5[1]) < 1e-9 tie_probs = softmax([3.0, 3.0, 1.0]) # index0==index1 tie assert {i for i, p in enumerate(top_k_filter(tie_probs, 1)) if p > 0} == {0} # --- top-k domain/boundary: k must satisfy 1 <= k <= n (n=5 here) --- n5 = len(probs5) for bad_k in (0, -1, n5 + 1): try: top_k_filter(probs5, bad_k) raise AssertionError(f"top_k_filter must reject k={bad_k}") except ValueError: pass filt_kn = top_k_filter(probs5, n5) # k == n keeps everything assert all(p > 0 for p in filt_kn) assert abs(sum(filt_kn) - 1.0) < 1e-12 # --- top-p: design-review numeric example ----------------------------- # p = [0.4, 0.3, 0.2, 0.1], threshold 0.65 -> boundary token MUST be kept p_ex = [0.4, 0.3, 0.2, 0.1] filt = top_p_filter(p_ex, 0.65) support = {i for i, p in enumerate(filt) if p > 0} assert support == {0, 1} # exactly the first two assert abs(sum(p_ex[i] for i in support) - 0.7) < 1e-12 # cumulative mass 0.7 # naive "cumulative strictly < p0" would stop BEFORE the boundary token # (mass 0.4 < 0.65) and wrongly drop it: assert sum([p_ex[0]]) < 0.65 # naive "keep all p_i >= p0" keeps NOTHING here (no single token >= 0.65): assert all(p < 0.65 for p in p_ex) assert abs(sum(filt) - 1.0) < 1e-12 and all(p >= 0 for p in filt) # relative proportions within the support must match the original ratio # (a bug that renormalizes to a uniform 0.5/0.5 would also pass a bare # "sums to 1" check, so check the actual values): assert abs(filt[0] - p_ex[0] / 0.7) < 1e-12 assert abs(filt[1] - p_ex[1] / 0.7) < 1e-12 # top_p = 1 keeps every token with nonzero probability filt_all = top_p_filter(p_ex, 1.0) assert all(f > 0 for f in filt_all) # --- exact boundary: p0 == 0.7 == the true cumulative mass of {0,1} --- # this is the case that distinguishes a correct `>=` from a wrong # strict `>` (which would keep scanning past the boundary token): filt_boundary = top_p_filter(p_ex, 0.7) support_boundary = {i for i, p in enumerate(filt_boundary) if p > 0} assert support_boundary == {0, 1} # --- epsilon regression: p0 a hair ABOVE the true cumulative mass of # {0,1} (0.7 exactly) must NOT stop at {0,1} -- a threshold that has # been fudged downward by e.g. `- 1e-12` would wrongly treat 0.7 as # "close enough" and drop the token that should push the support to # {0,1,2} --- p0_just_above = 0.7 + 5e-13 filt_eps = top_p_filter(p_ex, p0_just_above) support_eps = {i for i, p in enumerate(filt_eps) if p > 0} assert support_eps == {0, 1, 2} # --- p0 domain: must satisfy 0 < p0 <= 1 --- for bad_p0 in (0.0, -0.1, 1.1): try: top_p_filter(p_ex, bad_p0) raise AssertionError(f"top_p_filter must reject p0={bad_p0}") except ValueError: pass # --- min-p --- z = [3.1, 1.0, 0.4, -2.0] probs_mp = softmax(z) argmax_i = probs_mp.index(max(probs_mp)) prev_support_size = None prev_support_set = None for alpha in (0.9, 0.5, 0.2, 0.05): filt = min_p_filter(probs_mp, alpha) support = {i for i, p in enumerate(filt) if p > 0} assert argmax_i in support # max-prob token always kept size = len(support) if prev_support_size is not None: assert size >= prev_support_size # smaller alpha -> support grows (monotonic) assert prev_support_set <= support # NESTED: shrinking alpha only ADDS # tokens, it never drops one that a # larger alpha had already kept if size >= 2: # relative proportions within the support must match the # original distribution here too (not just "sums to 1") j, k = sorted(support)[0], sorted(support)[1] assert abs(filt[j] / filt[k] - probs_mp[j] / probs_mp[k]) < 1e-9 prev_support_size = size prev_support_set = support only_argmax = min_p_filter(probs_mp, 1.0) assert {i for i, p in enumerate(only_argmax) if p > 0} == {argmax_i} # --- alpha=1 with a TIE for the max: must keep ALL tied-max tokens, # not just one of them (the earlier test only covered a unique max) --- tied_max_probs = softmax([3.0, 3.0, 1.0]) only_tied = min_p_filter(tied_max_probs, 1.0) tied_support = {i for i, p in enumerate(only_tied) if p > 0} assert tied_support == {0, 1} assert abs(only_tied[0] - 0.5) < 1e-12 and abs(only_tied[1] - 0.5) < 1e-12 # --- alpha domain: must satisfy 0 < alpha <= 1 --- for bad_alpha in (0.0, -0.1, 1.1): try: min_p_filter(probs_mp, bad_alpha) raise AssertionError(f"min_p_filter must reject alpha={bad_alpha}") except ValueError: pass # logit shift invariance: adding a constant to all logits changes nothing shifted = [x + 7.0 for x in z] assert min_p_filter(softmax(shifted), 0.3) == min_p_filter(softmax(z), 0.3) # --- logit shift invariance for top-k and top-p too (the tutorial # appendix claims this holds for min-p/top-k/top-p; only min-p was # actually checked above -- close that gap here) --- z_shift_base = [2.0, 1.0, 0.0, -1.0, -3.0] probs_shift_base = softmax(z_shift_base) probs_shift_plus5 = softmax([zz + 5.0 for zz in z_shift_base]) assert top_k_filter(probs_shift_base, 2) == top_k_filter(probs_shift_plus5, 2) assert top_p_filter(probs_shift_base, 0.7) == top_p_filter(probs_shift_plus5, 0.7) # --- min-p support set MOVES with temperature (not a fixed logit gap) --- logits_t = [4.0, 2.0, 1.0, -1.0] alpha = 0.3 sizes = [] for T in (0.2, 0.5, 1.0, 2.0, 5.0): s = min_p_support_from_logits(logits_t, T, alpha) # cross-check: same support via probs-space filter at this T probs_T = temperature_probs(logits_t, T) s_via_probs = {i for i, p in enumerate(min_p_filter(probs_T, alpha)) if p > 0} assert s == s_via_probs sizes.append(len(s)) assert sizes == sorted(sizes) # non-decreasing as T grows (alpha<1) assert sizes[0] < sizes[-1] # and it strictly changes somewhere # --- min_p_support_from_logits domain: T must be finite and > 0 # (negative/zero temperature has no softmax(z/T) interpretation) --- for bad_T in (0.0, -1.0, float("inf"), float("nan")): try: min_p_support_from_logits(logits_t, bad_T, alpha) raise AssertionError(f"min_p_support_from_logits must reject T={bad_T}") except ValueError: pass # --- typical sampling: NOT equivalent to top-p; support need not be a # contiguous prefix of the probability-sorted order (it can skip the # single highest-probability token while keeping two lower ones) --- p_typ = [0.02, 0.35, 0.33, 0.30] # sums to 1.0 assert abs(sum(p_typ) - 1.0) < 1e-12 argmax_typ = p_typ.index(max(p_typ)) # index 1 (prob 0.35) filt_typ = typical_filter(p_typ, 0.63) support_typ = {i for i, p in enumerate(filt_typ) if p > 0} assert support_typ == {2, 3} # skips index1 (the argmax!) assert argmax_typ not in support_typ # this set is not equal to ANY top-m-by-probability prefix (m=1,2,3): order_by_prob = sorted(range(len(p_typ)), key=lambda i: (-p_typ[i], i)) for m in (1, 2, 3): assert support_typ != set(order_by_prob[:m]) assert abs(sum(filt_typ) - 1.0) < 1e-12 # relative proportions within the support must match the original ratio assert abs(filt_typ[2] / filt_typ[3] - p_typ[2] / p_typ[3]) < 1e-9 # --- typical_p domain: must satisfy 0 < typical_p <= 1 --- for bad_tp in (0.0, -0.1, 1.1): try: typical_filter(p_typ, bad_tp) raise AssertionError(f"typical_filter must reject typical_p={bad_tp}") except ValueError: pass # --- penalties --- logits_p = [2.0, -1.0, 0.5] counts = [3, 0, 1] pen = freq_presence_penalty(logits_p, counts, alpha_f=0.2, alpha_p=0.5) assert pen[1] == -1.0 # never appeared -> untouched assert pen[0] == 2.0 - 0.2 * 3 - 0.5 assert pen[2] == 0.5 - 0.2 * 1 - 0.5 rp = repetition_penalty([2.0, -2.0, 1.0], seen={0, 1}, r=2.0) assert rp == [1.0, -4.0, 1.0] # positive/r, negative*r, untouched # --- constrained decoding: illegal tokens get exactly zero mass --- masked = apply_hard_mask([1.0, 2.0, 0.5, 9.0], allowed=[True, True, False, False]) p_masked = softmax(masked) assert p_masked[2] == 0.0 and p_masked[3] == 0.0 assert abs(p_masked[0] + p_masked[1] - 1.0) < 1e-12 # --- sample(): deterministic checks via a stub RNG (the PRIMARY # verification -- exact `u` values pinned to exact bucket boundaries, # rather than relying only on statistical tolerance) --- probs_det = [0.5, 0.2, 0.15, 0.1, 0.05] # cumulative: .5 .7 .85 .95 1.0 assert sample(probs_det, StubRNG([0.0])) == 0 # u=0.0 -> first bucket assert sample(probs_det, StubRNG([0.5])) == 1 # u==cum(bucket0) -> NOT bucket0 # (u < cum is false at the boundary, # so it must fall into the NEXT bucket) assert sample(probs_det, StubRNG([0.999999999])) == 4 # near 1 -> last bucket assert sample(probs_det, StubRNG([1.0])) == 4 # forces the fallback path (true final # cumulative sum can undershoot 1.0 due # to floating point); must still land on # the LAST bucket, not silently misbehave # zero-probability-token regression: u=0.0 must never select a token # whose probability was filtered to exactly zero, whether it's at the # front of the vector or the very last (fallback) index probs_zero_head = [0.0, 0.6, 0.4] assert sample(probs_zero_head, StubRNG([0.0])) == 1 # NOT index 0 (zero mass) # this specific construction (ten 0.1's, naive running sum lands EXACTLY # on the largest double < 1.0, plus a filtered zero-prob tail) forces the # fallback branch for u=1.0: an old buggy `return len(probs) - 1` would # return the ZERO-probability tail index (10); the correct fallback must # return the last POSITIVE-probability index (9) instead. probs_zero_tail = [0.1] * 10 + [0.0] assert sample(probs_zero_tail, StubRNG([1.0])) == 9 try: sample([0.0, 0.0], StubRNG([0.5])) raise AssertionError("sample() must reject an all-zero probability vector") except ValueError: pass # --- draw(): empirical frequency check kept as a SUPPLEMENTARY # statistical sanity test (fixed seed, wide tolerance) -- the # deterministic StubRNG checks above are the primary verification --- rng = random.Random(42) two_way = [0.7, 0.3] n = 4000 count0 = sum(1 for _ in range(n) if sample(two_way, rng) == 0) assert 0.65 <= count0 / n <= 0.75 print("[D01] sampling operators: greedy tie-break, T=0 rejected, T->0 limits " "(unique->one-hot, tied->split), top-k/top-p/min-p/typical support sets " "(exact boundaries, nested/shift-invariant, domain-checked), penalties, " "constrained masking, deterministic + empirical draw checks PASS") def run_d02(): """Filter-order non-commutativity: top_k->top_p vs top_p->top_k on the SAME distribution give DIFFERENT final support sets.""" probs = [0.5, 0.2, 0.15, 0.1, 0.05] order_a = top_p_filter(top_k_filter(probs, 3), 0.8) # top_k THEN top_p order_b = top_k_filter(top_p_filter(probs, 0.8), 3) # top_p THEN top_k support_a = {i for i, p in enumerate(order_a) if p > 0} support_b = {i for i, p in enumerate(order_b) if p > 0} assert support_a == {0, 1} assert support_b == {0, 1, 2} assert support_a != support_b # order matters print(f"[D02] filter order is NOT commutative: top_k->top_p support={sorted(support_a)}, " f"top_p->top_k support={sorted(support_b)} PASS") # ============================================================ # [D03] Streaming stop-sequence detection across token chunks # ============================================================ def _stop_overlap_len(buffer, stop): """Longest suffix of `buffer` that is also a (proper) prefix of `stop` -- that suffix must be held back because it MIGHT still grow into a full match with the next chunk.""" max_k = min(len(buffer), len(stop) - 1) for k in range(max_k, 0, -1): if buffer.endswith(stop[:k]): return k return 0 def stream_with_stop(chunks, stop): """Feed token chunks one at a time. Returns (emitted_text, stopped, leftover_buffer). Never emits text that could still be a partial match of `stop` -- that text stays in the internal buffer until either (a) the match completes (truncate + stop) or (b) more chunks arrive and prove it can't complete after all (then it becomes safe to emit).""" emitted, buffer = "", "" for chunk in chunks: buffer += chunk if stop in buffer: idx = buffer.index(stop) emitted += buffer[:idx] return emitted, True, "" k = _stop_overlap_len(buffer, stop) emitted += buffer[:len(buffer) - k] buffer = buffer[len(buffer) - k:] return emitted, False, buffer def run_d03(): # stop string "STOP" spans three separate token chunks: "cS" | "TO" | "Pxyz" chunks = ["ab", "cS", "TO", "Pxyz"] emitted, stopped, _ = stream_with_stop(chunks, "STOP") assert emitted == "abc" and stopped is True # never leaked "S", "ST", or "STO" as committed output at any point: partial_emits = [] buf, acc = "", "" for c in chunks: buf += c if "STOP" in buf: acc += buf[:buf.index("STOP")] partial_emits.append(acc) break k = _stop_overlap_len(buf, "STOP") acc += buf[:len(buf) - k] buf = buf[len(buf) - k:] partial_emits.append(acc) assert all(not p.endswith(("S", "ST", "STO")) for p in partial_emits[:-1]) # no stop match at all: nothing is lost, everything reconstructible via flush chunks_nostop = ["ab", "cd", "ef"] emitted2, stopped2, remainder = stream_with_stop(chunks_nostop, "STOP") assert stopped2 is False assert emitted2 + remainder == "".join(chunks_nostop) # full text recoverable print(f"[D03] stop-string streaming: 'STOP' split across 3 chunks still detected " f"(emitted={emitted!r}), no-match case fully recoverable via flush PASS") # ============================================================ # [D04] Static vs. continuous batching simulator # ============================================================ Request = namedtuple("Request", ["name", "arrival", "ticks_needed"]) def _validate_requests(requests, capacity): """Reject anything that could make the tick loop below fail to terminate: duplicate names (dict/active/completed all key on `name`, so duplicates silently merge and the loop's `len(completed) < len(requests)` target can never be reached), non-positive/non-integer capacity (nothing can ever be admitted), and non-positive/non-integer ticks_needed (a request with ticks_needed=0 gets decremented straight to -1 and never equals 0 again; a non-integer can step over 0 the same way) or a negative/non-integer arrival (outside the tick domain).""" if not requests: raise ValueError("requests must be non-empty") names = [r.name for r in requests] if len(set(names)) != len(names): raise ValueError(f"request names must be unique, got {names}") if not isinstance(capacity, int) or isinstance(capacity, bool) or capacity <= 0: raise ValueError(f"capacity must be a positive integer, got {capacity!r}") for r in requests: if not isinstance(r.arrival, int) or isinstance(r.arrival, bool) or r.arrival < 0: raise ValueError( f"arrival must be a non-negative integer, got {r.arrival!r} for request {r.name!r}") if not isinstance(r.ticks_needed, int) or isinstance(r.ticks_needed, bool) or r.ticks_needed <= 0: raise ValueError( f"ticks_needed must be a positive integer, got {r.ticks_needed!r} for request {r.name!r}") def simulate_batching(requests, capacity, policy): """Tick-based scheduler-capacity simulator. NOT a model of real GPU utilization -- it is a unit-cost toy: every ACTIVE request advances by exactly one decode token per tick. policy == "static": a batch is formed (up to `capacity` arrived, not-yet- started requests, FCFS) only when the active set is completely empty. A finished member's slot sits idle -- it is NOT backfilled -- until every member of that batch has finished. policy == "continuous": every tick, after removing finished requests, the scheduler tops the active set back up to `capacity` with whichever arrived requests are waiting (FCFS). """ _validate_requests(requests, capacity) by_name = {r.name: r for r in requests} waiting = sorted(requests, key=lambda r: (r.arrival, r.name)) active = {} # name -> remaining ticks started = set() completed = {} serviced_log = [] # (tick, name) once per unit of service t, guard = 0, 0 n_requests = len(requests) while len(completed) < n_requests: guard += 1 if guard >= 100_000: # an explicit exception (not `assert`) so this guard survives # even when the interpreter is run with `python -O`, which # strips bare `assert` statements -- illegal input must never # be able to turn into a silent infinite loop. raise RuntimeError("simulation did not terminate within the tick guard") arrived = [r for r in waiting if r.arrival <= t and r.name not in started] if policy == "continuous": while len(active) < capacity and arrived: r = arrived.pop(0) active[r.name] = r.ticks_needed started.add(r.name) elif policy == "static": if len(active) == 0 and arrived: for r in arrived[:capacity]: active[r.name] = r.ticks_needed started.add(r.name) else: raise ValueError(policy) if not active: # Idle tick: nothing is running. Given validated input this can # only mean some requests simply haven't arrived yet (every # already-arrived, not-yet-started request would have been # admitted above, by either policy, since active was empty) -- # so jump straight to the next arrival instead of burning one # tick at a time waiting for it. not_yet_started = [r for r in waiting if r.name not in started] if not_yet_started: t = max(t, min(r.arrival for r in not_yet_started)) continue raise RuntimeError( "no active or pending requests but the simulation is not complete " "(internal inconsistency)") # -- invariants, checked every tick -- assert len(active) <= capacity # capacity respected for name in active: assert by_name[name].arrival <= t # never runs before arrival assert name not in completed # completed -> no longer occupies a slot (no KV held) for name in list(active.keys()): serviced_log.append((t, name)) # exactly one unit of service this tick active[name] -= 1 t += 1 for name in list(active.keys()): if active[name] == 0: completed[name] = t del active[name] # slot freed, KV reclaimed # generation-token-count conservation per_request_service = {} for tick, name in serviced_log: per_request_service[name] = per_request_service.get(name, 0) + 1 for r in requests: assert per_request_service[r.name] == r.ticks_needed return completed, serviced_log def utilization(completed, requests, capacity): total_work = sum(r.ticks_needed for r in requests) makespan = max(completed.values()) return total_work / (capacity * makespan), makespan def run_d04(): reqs = [Request("A", 0, 1), Request("B", 0, 4), Request("C", 0, 1)] cap = 2 comp_static, _ = simulate_batching(reqs, cap, "static") comp_cont, _ = simulate_batching(reqs, cap, "continuous") assert comp_static == {"A": 1, "B": 4, "C": 5} # C waits until t=4 assert comp_cont == {"A": 1, "C": 2, "B": 4} # C is refilled into A's slot at t=1 u_static, ms_static = utilization(comp_static, reqs, cap) u_cont, ms_cont = utilization(comp_cont, reqs, cap) assert ms_static == 5 and ms_cont == 4 assert abs(u_static - 0.60) < 1e-9 assert abs(u_cont - 0.75) < 1e-9 # --- equal-length counterexample: continuous does NOT strictly dominate --- reqs_eq = [Request("X", 0, 3), Request("Y", 0, 3)] comp_static_eq, _ = simulate_batching(reqs_eq, 2, "static") comp_cont_eq, _ = simulate_batching(reqs_eq, 2, "continuous") assert comp_static_eq == comp_cont_eq == {"X": 3, "Y": 3} # identical makespan u_s_eq, ms_s_eq = utilization(comp_static_eq, reqs_eq, 2) u_c_eq, ms_c_eq = utilization(comp_cont_eq, reqs_eq, 2) assert ms_s_eq == ms_c_eq and abs(u_s_eq - u_c_eq) < 1e-12 # --- delayed-arrival workload: every previous test used arrival=0 for # everything, which can never independently verify "a request must # never be serviced before it arrives" -- P arrives at 0, Q at 3, # R at 5; check every logged service record against arrival/completion --- reqs_delayed = [Request("P", 0, 2), Request("Q", 3, 2), Request("R", 5, 1)] comp_delayed, log_delayed = simulate_batching(reqs_delayed, capacity=2, policy="continuous") by_name_delayed = {r.name: r for r in reqs_delayed} assert comp_delayed == {"P": 2, "Q": 5, "R": 6} for tick, name in log_delayed: assert by_name_delayed[name].arrival <= tick < comp_delayed[name] # --- illegal input must raise immediately, never hang (previously: # duplicate names broke the loop's termination condition; ticks_needed # <=0 or non-integer could step over the "== 0 -> completed" check; # capacity <=0 could never admit anything; the old termination guard # was a bare `assert`, stripped by `python -O`) --- illegal_cases = [ ([Request("A", 0, 1), Request("A", 0, 2)], 2, "continuous"), # duplicate name ([Request("A", 0, 0)], 2, "continuous"), # ticks_needed == 0 ([Request("A", 0, 1.5)], 2, "continuous"), # non-integer ticks_needed ([Request("A", 0, 1)], 0, "continuous"), # capacity == 0 ([Request("A", 0, 1)], -1, "continuous"), # capacity < 0 ([Request("A", -1, 1)], 2, "continuous"), # negative arrival ([Request("A", 0.5, 1)], 2, "continuous"), # non-integer arrival ] for bad_reqs, bad_cap, bad_policy in illegal_cases: try: simulate_batching(bad_reqs, bad_cap, bad_policy) raise AssertionError(f"simulate_batching must reject {bad_reqs, bad_cap}") except ValueError: pass print(f"[D04] static: A=1,B=4,C=5 (util={u_static:.2f}); " f"continuous: A=1,C=2,B=4 (util={u_cont:.2f}); " f"equal-length counterexample: both makespan={ms_s_eq} (no strict dominance); " f"delayed-arrival workload respects arrival<=tick 0, "decode share must leave room for prefill progress" iters_chunked = math.ceil(total_prefill_work / prefill_rate_chunked) max_gap_chunked = 1 return { "ttft_unchunked": iters_unchunked, "max_gap_unchunked": max_gap_unchunked, "ttft_chunked": iters_chunked, "max_gap_chunked": max_gap_chunked, } def run_d05(): r = chunked_vs_unchunked(total_prefill_work=12, budget_per_iter=4, decode_share_per_iter=1) assert r["max_gap_chunked"] < r["max_gap_unchunked"] # decode ITL improves assert r["ttft_chunked"] > r["ttft_unchunked"] # but this prompt's own TTFT worsens assert r == {"ttft_unchunked": 3, "max_gap_unchunked": 3, "ttft_chunked": 4, "max_gap_chunked": 1} print(f"[D05] chunked prefill: decode max-gap {r['max_gap_unchunked']}->{r['max_gap_chunked']} " f"(ITL improves) but long-prompt TTFT {r['ttft_unchunked']}->{r['ttft_chunked']} " f"(worsens) -- not a free lunch PASS") # ============================================================ # [D06]-[D08] KV-cache calculator (reuses the KV-cache sibling tutorial's # own per-token formula; no re-derivation of MQA/GQA/MLA cache-ratio math) # ============================================================ def kv_bytes_per_token(n_layer, n_kv_head, d_head, bytes_per_elem): """m_token = 2 * N_layer * N_kv_head * d_head * bytes_per_elem (the leading 2 is for K and V). Same formula as the sibling KV-cache tutorial's ยง2.1 (per-token slice of its L_ctx-scaled cache formula).""" return 2 * n_layer * n_kv_head * d_head * bytes_per_elem def block_alloc_slots(lengths, block_size): """logical slots = sum of true lengths; allocated slots round EACH request up to a whole number of blocks (internal fragmentation in the last, partially-filled block).""" logical = sum(lengths) allocated = sum(block_size * math.ceil(L / block_size) for L in lengths) return logical, allocated def physical_blocks_no_sharing(lengths, block_size): return sum(math.ceil(L / block_size) for L in lengths) def shared_prefix_blocks(token_seqs, block_size): """Number of COMPLETE blocks shared as an identical prefix across EVERY sequence in `token_seqs` -- an exact, block-by-block token comparison, not an assumed/hand-picked prefix length. Sharing can only happen at whole-block granularity: a common prefix shorter than one block shares nothing, and comparison stops at the first block where any sequence's content diverges (or where the shortest sequence runs out).""" if len(token_seqs) < 2: return 0 min_len = min(len(s) for s in token_seqs) n_blocks = min_len // block_size shared = 0 for b in range(n_blocks): lo, hi = b * block_size, (b + 1) * block_size block0 = token_seqs[0][lo:hi] if all(s[lo:hi] == block0 for s in token_seqs[1:]): shared += 1 else: break return shared def physical_blocks_with_sharing(token_seqs, block_size): """B_physical = B_no_share - (n-1) * B_shared: every request AFTER the first reuses the B_shared blocks of common prefix instead of allocating its own copy, so the saving scales with (n-1), not with a fixed constant. Returns (physical_blocks, shared_blocks, no_share_blocks).""" no_share = physical_blocks_no_sharing([len(s) for s in token_seqs], block_size) b_shared = shared_prefix_blocks(token_seqs, block_size) n = len(token_seqs) physical = no_share - (n - 1) * b_shared return physical, b_shared, no_share def kv_bytes_under_tp(total_bytes, n_kv_head, tp_degree): """Only divide by TP degree when KV heads are evenly sharded across ranks. Otherwise this is an outright modeling error (replicated heads, uneven layouts, etc. need their own accounting).""" if n_kv_head % tp_degree != 0: raise ValueError( f"{n_kv_head} KV heads not evenly divisible by TP degree {tp_degree}: " "cannot assume a uniform per-rank split") return total_bytes / tp_degree def run_d06_d07_d08(): # [D06] per-token bytes, synthetic config m_token = kv_bytes_per_token(n_layer=2, n_kv_head=2, d_head=4, bytes_per_elem=2) # 2 (K,V) * 2 (layers) * 2 (kv heads) * 4 (head dim) * 2 (bytes/elem) = 64 bytes/token assert m_token == 64 # [D07] block allocator: logical vs allocated slots (and bytes) lengths = [1, 5] logical_slots, allocated_slots = block_alloc_slots(lengths, block_size=4) assert (logical_slots, allocated_slots) == (6, 12) logical_bytes = logical_slots * m_token allocated_bytes = allocated_slots * m_token assert (logical_bytes, allocated_bytes) == (384, 768) assert allocated_bytes > logical_bytes # internal fragmentation is real # [D08] prefix-sharing block count -- computed from REAL token sequences # (not a hand-picked constant): two requests share tokens [1,2,3,4] as a # complete first block (block_size=4), then diverge. shared_prefix = [1, 2, 3, 4] seq_a = shared_prefix + [5, 6] # length 6 seq_b = shared_prefix + [7, 8, 9] # length 7 no_share = physical_blocks_no_sharing([len(seq_a), len(seq_b)], block_size=4) assert no_share == 4 # ceil(6/4)+ceil(7/4) = 2+2 with_share, shared_blocks, no_share_check = physical_blocks_with_sharing([seq_a, seq_b], block_size=4) assert no_share_check == no_share assert shared_blocks == 1 # verified token-by-token, not asserted assert with_share == 3 # 4 - (2-1)*1 = 3 # a common prefix SHORTER than one block shares nothing (block-granularity only) seq_c = [1, 2, 3] + [10, 11, 12, 13] # only 3 tokens overlap with seq_a's start seq_d = [1, 2, 3] + [20, 21, 22, 23, 24] physical_partial, shared_partial, no_share_partial = physical_blocks_with_sharing( [seq_c, seq_d], block_size=4) assert shared_partial == 0 assert physical_partial == no_share_partial # three requests sharing the SAME full first block: the saving scales # as (n-1)*B_shared, not a fixed constant seq_e = shared_prefix + [30] seq_f = shared_prefix + [31, 32] physical_3, shared_3, no_share_3 = physical_blocks_with_sharing( [seq_a, seq_e, seq_f], block_size=4) assert shared_3 == 1 assert physical_3 == no_share_3 - 2 * shared_3 # (n-1) = 2 requests reuse the block # TP-division caveat assert kv_bytes_under_tp(1024.0, n_kv_head=8, tp_degree=4) == 256.0 # 8 % 4 == 0, fine try: kv_bytes_under_tp(1024.0, n_kv_head=2, tp_degree=4) raise AssertionError("must reject uneven KV-head / TP-degree split") except ValueError: pass print(f"[D06] KV bytes/token (L=2,kv_head=2,d_head=4,fp16)={m_token}B; " f"[D07] block alloc lengths=[1,5],P=4: logical={logical_slots} slots/{logical_bytes}B, " f"allocated={allocated_slots} slots/{allocated_bytes}B; " f"[D08] prefix-sharing (real token-sequence match) blocks {no_share}->{with_share}, " f"sub-block overlap shares 0, 3-way sharing scales as (n-1)*B_shared, " f"TP-divisibility guard enforced PASS") # ============================================================ # [D09] (optional) roofline ridge point -- an upper-bound number, NOT a # latency predictor # ============================================================ def ridge_point(peak_flops_per_s, hbm_bytes_per_s): """I* = peak FLOP/s / HBM byte/s. Purely a units ratio used to compare against a kernel's own arithmetic intensity; it bounds a BEST CASE, it does not predict actual latency (real kernels sit below the roofline).""" assert peak_flops_per_s > 0 and hbm_bytes_per_s > 0 return peak_flops_per_s / hbm_bytes_per_s def run_d09(): i_star = ridge_point(peak_flops_per_s=312e12, hbm_bytes_per_s=1.55e12) # illustrative A100-class numbers assert i_star > 0 # a decode step's arithmetic intensity in a small-batch regime is typically # far below any realistic ridge point; this demo only checks the ratio is # well-formed, not that it "predicts" any specific latency small_batch_intensity = 8.0 # illustrative FLOPs/byte, well under I* assert small_batch_intensity < i_star print(f"[D09] roofline ridge point I* = {i_star:.1f} FLOPs/byte (upper-bound reference only) PASS") def main(): run_d01() run_d02() run_d03() run_d04() run_d05() run_d06_d07_d08() run_d09() print("\nall inference serving sanity checks passed") if __name__ == "__main__": main()