h3-blackwell-runtime/tools/benchmark_ulysses.py

"""Benchmark ragged H3 Ulysses transport and attention under torchrun."""

import argparse
import json
import os
import statistics
import time
from pathlib import Path

import torch
import torch.distributed as dist

from h3_blackwell_runtime.attention import run_attention, run_flash4_attention_bshd, run_sol_attention_bshd
from h3_blackwell_runtime.distributed import SequenceParallelContext


def synchronize(device: torch.device) -> None:
    if device.type == "cuda":
        torch.cuda.synchronize(device)


def run_backend(backend: str, q: torch.Tensor, k: torch.Tensor, v: torch.Tensor) -> torch.Tensor:
    if backend == "flash4":
        return run_flash4_attention_bshd(q, k, v, is_causal=False)
    if backend == "sol_attn":
        return run_sol_attention_bshd(q, k, v, is_causal=False)
    return run_attention(
        q.transpose(1, 2).contiguous(),
        k.transpose(1, 2).contiguous(),
        v.transpose(1, 2).contiguous(),
        backend=backend,
        is_causal=False,
    ).transpose(1, 2)


parser = argparse.ArgumentParser()
parser.add_argument("--sequence", type=int, default=20000)
parser.add_argument("--heads", type=int, default=56)
parser.add_argument("--head-dim", type=int, default=128)
parser.add_argument("--backend", default="sdpa")
parser.add_argument("--warmup", type=int, default=3)
parser.add_argument("--iterations", type=int, default=10)
parser.add_argument("--output", type=Path)
args = parser.parse_args()

local_rank = int(os.environ.get("LOCAL_RANK", "0"))
use_cuda = torch.cuda.is_available()
device = torch.device(f"cuda:{local_rank}" if use_cuda else "cpu")
if use_cuda:
    torch.cuda.set_device(device)
dist.init_process_group(backend="nccl" if use_cuda else "gloo", device_id=device if use_cuda else None)
rank = dist.get_rank()
world_size = dist.get_world_size()
context = SequenceParallelContext.create(args.sequence, args.heads, args.head_dim)

generator = torch.Generator(device=device).manual_seed(440420 + rank)
shape = (1, context.local_token_length, args.heads, args.head_dim)
dtype = torch.bfloat16 if use_cuda else torch.float32
q = torch.randn(shape, generator=generator, dtype=dtype, device=device)
k = torch.randn(shape, generator=generator, dtype=dtype, device=device)
v = torch.randn(shape, generator=generator, dtype=dtype, device=device)


def iteration() -> tuple[float, float, float, torch.Tensor]:
    dist.barrier()
    synchronize(device)
    started = time.perf_counter()
    full_q, full_k, full_v = context.seq_to_heads(q, k, v)
    synchronize(device)
    after_forward = time.perf_counter()
    head_output = run_backend(args.backend, full_q, full_k, full_v)
    synchronize(device)
    after_attention = time.perf_counter()
    local_output = context.heads_to_seq(head_output)
    synchronize(device)
    finished = time.perf_counter()
    return (
        after_forward - started,
        after_attention - after_forward,
        finished - after_attention,
        local_output,
    )


for _ in range(args.warmup):
    *_timings, output = iteration()
    del output

rank_timings = []
for _ in range(args.iterations):
    forward, attention, inverse, output = iteration()
    rank_timings.append((forward, attention, inverse, forward + attention + inverse))
    del output

timings = torch.tensor(rank_timings, dtype=torch.float64, device=device)
gathered = [torch.empty_like(timings) for _ in range(world_size)]
dist.all_gather(gathered, timings)

if use_cuda:
    peak_memory = torch.tensor([torch.cuda.max_memory_allocated(device)], dtype=torch.int64, device=device)
else:
    peak_memory = torch.tensor([0], dtype=torch.int64, device=device)
memory_by_rank = [torch.empty_like(peak_memory) for _ in range(world_size)]
dist.all_gather(memory_by_rank, peak_memory)

if rank == 0:
    stacked = torch.stack(gathered).cpu()
    stage_names = ("forward_all_to_all", "attention", "inverse_all_to_all", "total")
    stages = {}
    for index, name in enumerate(stage_names):
        maximum_rank = stacked[:, :, index].amax(dim=0).tolist()
        stages[name] = {
            "median_seconds": statistics.median(maximum_rank),
            "minimum_seconds": min(maximum_rank),
            "maximum_seconds": max(maximum_rank),
        }
    element_size = q.element_size()
    report = {
        "world_size": world_size,
        "backend": args.backend,
        "device": torch.cuda.get_device_name(device) if use_cuda else "cpu",
        "torch": torch.__version__,
        "sequence": args.sequence,
        "heads": args.heads,
        "head_dim": args.head_dim,
        "token_lengths": list(context.token_lengths),
        "head_lengths": list(context.head_lengths),
        "dtype": str(dtype),
        "iterations": args.iterations,
        "aggregate_transport_bytes_per_iteration": {
            "forward_qkv": 3 * args.sequence * args.heads * args.head_dim * element_size,
            "inverse_output": args.sequence * args.heads * args.head_dim * element_size,
        },
        "stages": stages,
        "peak_allocated_bytes_by_rank": [int(value.item()) for value in memory_by_rank],
    }
    serialized = json.dumps(report, indent=2)
    if args.output is not None:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(serialized + "\n", encoding="utf-8")
    print(serialized)

dist.destroy_process_group()