sgl-project--sglang/benchmark/scheduler/bench_token_storage.py

"""Benchmark `list[int]` vs `array.array('q')` storage for
`Req.origin_input_ids` / `Req.output_ids` over one request lifecycle.

Simulated steps (per batch):
    1. ingest        -- tokenizer list[int] -> storage container.
    2. prefix_match  -- scheduler radix-tree lookup; RadixKey.match()
                        zip+!= walk. Exposes the per-element PyLong-boxing
                        cost array.array introduces (list[int] iterates
                        existing PyLongs and pays nothing).
    3. prefill       -- (a) fill_ids = origin + output,
                        (b) per-req slice fill_ids[prefix_len:],
                        (c) cross-req flatten + pinned cuda tensor build.
    4. decode        -- per-step output.append(next_token) for n_decode steps.
    5. finish        -- cache_finished_req:
                        (a) concat (origin + output)[:kv_committed_len]
                            for the radix-tree insert.
                        (b) RadixKey.match() zip+!= walk during insert's
                            tree traversal — second PyLong-boxing hotspot
                            on the array.array path.

Usage:
    python benchmark/scheduler/bench_token_storage.py
"""

from __future__ import annotations

import time
from array import array
from collections import defaultdict
from contextlib import contextmanager
from itertools import chain
from typing import Any, Callable, Iterator

import numpy as np
import torch

# Per-req stages accumulate across reqs in a batch; batch_torch_tensor
# is the single cross-req prepare_for_extend tensor build.
STAGES = (
    "ingest",
    "prefix_match",
    "prefill_concat",
    "prefill_perreq_slice",
    "batch_torch_tensor",
    "decode_append",
    "finish_concat",
    "cache_finished_req",
)


def _ingest_list(seed: list[int]) -> list[int]:
    return seed


def _ingest_pyarray(seed: list[int]) -> array:
    return array("q", seed)


def _empty_list() -> list[int]:
    return []


def _empty_pyarray() -> array:
    return array("q")


def _zip_iterate(t0: Any, t1: Any) -> int:
    """Simulate zip iteration which surface PyLong boxing cost in array scenario"""
    i = 0
    for a, b in zip(t0, t1):
        if a != b:
            break
        i += 1
    return i


def _batch_tensor_from_lists(parts: list[list[int]]) -> torch.Tensor:
    flat = list(chain.from_iterable(parts))
    return torch.tensor(flat, dtype=torch.int64, pin_memory=True).to(
        "cuda", non_blocking=True
    )


def _batch_tensor_from_pyarrays(parts: list[array]) -> torch.Tensor:
    # np.frombuffer gives a zero-copy view; np.concatenate is one C-level
    # memcpy. This bypasses the per-element PyLong->int64 walk that
    # torch.tensor(array('q')) would otherwise do.
    views = [np.frombuffer(p, dtype=np.int64) for p in parts]
    combined = np.concatenate(views) if len(views) > 1 else views[0]
    return torch.from_numpy(combined).pin_memory().to("cuda", non_blocking=True)


LIST_KIT = {
    "ingest_fn": _ingest_list,
    "empty_fn": _empty_list,
    "batch_torch_fn": _batch_tensor_from_lists,
}

PYARRAY_KIT = {
    "ingest_fn": _ingest_pyarray,
    "empty_fn": _empty_pyarray,
    "batch_torch_fn": _batch_tensor_from_pyarrays,
}


@contextmanager
def timed(timings: dict[str, float], stage: str) -> Iterator[None]:
    t0 = time.monotonic_ns()
    try:
        yield
    finally:
        timings[stage] += time.monotonic_ns() - t0


def simulate(
    seeds: list[list[int]],
    n_decode: int,
    *,
    ingest_fn: Callable[[list[int]], Any],
    empty_fn: Callable[[], Any],
    batch_torch_fn: Callable[[list[Any]], torch.Tensor],
) -> dict[str, float]:
    """One scheduling-round lifecycle. Returns per-stage cumulative ns."""
    timings: dict[str, float] = defaultdict(float)
    n_reqs = len(seeds)
    n_origins = [len(s) for s in seeds]
    origins: list[Any] = [None] * n_reqs
    outputs: list[Any] = [None] * n_reqs

    # 1. ingest
    for i, seed in enumerate(seeds):
        with timed(timings, "ingest"):
            origins[i] = ingest_fn(seed)
        outputs[i] = empty_fn()

    # 2. prefix_match: simulating the worse scenario of PyLong-boxing overhead during prefix_match
    for i in range(n_reqs):
        with timed(timings, "prefix_match"):
            _ = _zip_iterate(origins[i], origins[i])

    # 3. prefill
    per_req_slices: list[Any] = [None] * n_reqs
    for i in range(n_reqs):
        # 3a. fill_ids = origin_input_ids + output_ids
        with timed(timings, "prefill_concat"):
            fill_ids = origins[i] + outputs[i]
        # 3b. input_ids = fill_ids[len(prefix_indices):]; prefix_len=0 here.
        with timed(timings, "prefill_perreq_slice"):
            per_req_slices[i] = fill_ids[0:]
    # 3c. prepare_for_extend tensor build: flatten per-req slices, then
    #     build the pinned GPU tensor (kit-specific path).
    with timed(timings, "batch_torch_tensor"):
        _ = batch_torch_fn(per_req_slices)

    # 4. decode
    for i in range(n_reqs):
        with timed(timings, "decode_append"):
            for j in range(n_decode):
                outputs[i].append(j)

    # 5. finish: cache_finished_req -> insert -> _insert_helper tree walk.
    for i in range(n_reqs):
        # 5a. (origin + output)[:kv_committed_len] for the radix-tree insert.
        with timed(timings, "finish_concat"):
            committed = (origins[i] + outputs[i])[: n_origins[i] + n_decode]
        # 5b. simulating the worse scenario of PyLong-boxing overhead during cache_finished_req
        with timed(timings, "cache_finished_req"):
            _ = _zip_iterate(committed, committed)

    return timings


def bench_lifecycle(
    seeds: list[list[int]],
    n_decode: int,
    iterations: int,
    *,
    ingest_fn: Callable[[list[int]], Any],
    empty_fn: Callable[[], Any],
    batch_torch_fn: Callable[[list[Any]], torch.Tensor],
    warmup: int = 5,
) -> dict[str, float]:
    """Run simulate() N times, return mean per-stage us per batch.

    GPU sync is excluded from per-iteration timing: production issues
    `to(device, non_blocking=True)` and continues, so we measure issue
    cost rather than H2D completion.
    """
    kit = {
        "ingest_fn": ingest_fn,
        "empty_fn": empty_fn,
        "batch_torch_fn": batch_torch_fn,
    }
    torch.cuda.synchronize()
    for _ in range(warmup):
        simulate(seeds, n_decode, **kit)
    torch.cuda.synchronize()
    accum: dict[str, float] = defaultdict(float)
    for _ in range(iterations):
        t = simulate(seeds, n_decode, **kit)
        for k, v in t.items():
            accum[k] += v
    torch.cuda.synchronize()
    return {k: accum[k] / iterations / 1000.0 for k in STAGES}  # ns -> us


def print_breakdown(title: str, results: dict[str, dict[str, float]]) -> None:
    """Print per-stage timings with delta us vs the first (baseline) column."""
    labels = list(results.keys())
    baseline_label = labels[0]
    baseline = results[baseline_label]

    width = max(len(s) for s in STAGES)

    header_cells = [f"{baseline_label + ' us':>11s}"]
    for lbl in labels[1:]:
        header_cells.append(f"{lbl + ' us':>11s}")
        header_cells.append(f"{'delta':>10s}")

    print(f"=== {title} ===")
    print(f"{'Stage':<{width}s}  " + "  ".join(header_cells))
    print("-" * (width + 2 + sum(len(c) + 2 for c in header_cells)))

    for s in STAGES:
        cells = [f"{baseline[s]:>11.3f}"]
        for lbl in labels[1:]:
            v = results[lbl][s]
            cells.append(f"{v:>11.3f}")
            d = v - baseline[s]
            cells.append(f"{d:>+10.3f}")
        print(f"{s:<{width}s}  " + "  ".join(cells))

    print("-" * (width + 2 + sum(len(c) + 2 for c in header_cells)))

    base_total = sum(baseline.values())
    total_cells = [f"{base_total:>11.3f}"]
    for lbl in labels[1:]:
        v = sum(results[lbl].values())
        total_cells.append(f"{v:>11.3f}")
        d = v - base_total
        total_cells.append(f"{d:>+10.3f}")
    print(f"{'TOTAL':<{width}s}  " + "  ".join(total_cells))

    print()
    for lbl in labels[1:]:
        v = sum(results[lbl].values())
        d = v - base_total
        speedup = base_total / v if v > 0 else 0.0
        verdict = "LOSES" if d > 0 else "WINS"
        print(
            f"  {lbl:<14s} vs {baseline_label}: {verdict} by {abs(d):>8.2f} us  ({speedup:.2f}x)"
        )
    print()


def microbench_torch_tensor_paths(
    sizes: tuple[int, ...] = (1_000, 10_000, 100_000)
) -> None:
    """Compare three CPU-buffer -> pinned cuda tensor paths.

    A. torch.tensor(list,        pin) -> cuda
    B. torch.tensor(array('q'),  pin) -> cuda
    C. torch.from_numpy(np.frombuffer(array('q'))).pin() -> cuda
    """

    def t(fn, iterations: int) -> float:
        for _ in range(20):
            fn()
        torch.cuda.synchronize()
        t0 = time.monotonic_ns()
        for _ in range(iterations):
            fn()
        torch.cuda.synchronize()
        return (time.monotonic_ns() - t0) / iterations / 1000.0

    print("=== microbench: CPU-buffer -> pinned cuda tensor (us/op) ===\n")
    width = 56
    print(f"{'Path':<{width}s} " + "  ".join(f"{f'N={n}':>10s}" for n in sizes))
    print("-" * (width + 2 + 12 * len(sizes)))

    for label, build in [
        (
            "(A) torch.tensor(list,        pin) -> cuda",
            lambda x: torch.tensor(x, dtype=torch.int64, pin_memory=True).to(
                "cuda", non_blocking=True
            ),
        ),
        (
            "(B) torch.tensor(array('q'),  pin) -> cuda  (naive)",
            lambda x: torch.tensor(x, dtype=torch.int64, pin_memory=True).to(
                "cuda", non_blocking=True
            ),
        ),
        (
            "(C) from_numpy(frombuf(array('q'))).pin() -> cuda",
            lambda x: torch.from_numpy(np.frombuffer(x, dtype=np.int64))
            .pin_memory()
            .to("cuda", non_blocking=True),
        ),
    ]:
        cells = []
        for n in sizes:
            iters = max(50, 200_000 // max(n, 1))
            if "(A)" in label:
                src = list(range(n))
            else:
                src = array("q", range(n))
            us = t(lambda src=src, build=build: build(src), iters)
            cells.append(f"{us:>10.2f}")
        print(f"{label:<{width}s} " + "  ".join(cells))
    print()


def main() -> None:
    microbench_torch_tensor_paths()

    n_reqs = 2
    cases = [
        ("short  prompt  N_origin=1K    N_decode=1K", 1_000, 1_000, 1_000),
        ("medium prompt  N_origin=10K   N_decode=1K", 10_000, 1_000, 200),
        ("long   prompt  N_origin=100K  N_decode=1K", 100_000, 1_000, 30),
    ]
    print(f"Batch size = {n_reqs} reqs/batch (per-req stages accumulate)\n")
    for label, n_origin, n_decode, iters in cases:
        seeds = [list(range(n_origin)) for _ in range(n_reqs)]
        results = {
            "list": bench_lifecycle(seeds, n_decode, iters, **LIST_KIT),
            "pyarray": bench_lifecycle(seeds, n_decode, iters, **PYARRAY_KIT),
        }
        print_breakdown(label, results)


if __name__ == "__main__":
    main()