"""Comprehensive Benchmark Suite for AsyncTensorRLHF. Benchmarks: 1. In-VRAM Tensor-Native Reward vs. CPU-Roundtrip Baseline across multiple batch sizes. 2. PPO, M2PO, and GRPO loss & backward throughput (samples/sec & tokens/sec). 3. Replay buffer throughput (push/sample ops/sec under concurrency). 4. Staleness robustness benchmark (PPO vs. M2PO gradient variance under policy drift). """ import os import pathlib import sys import time import torch # Ensure project root in sys.path sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) from src.reward.tensor_native import tensor_native_reward from src.trainer.ppo_loss import compute_ppo_loss, compute_m2po_loss from src.trainer.grpo_loss import compute_grpo_loss from src.buffer.replay_buffer import BoundedReplayBuffer, VersionedReplayBuffer, Experience, VersionedExperience from src.buffer.group_buffer import GroupAwareReplayBuffer def run_reward_benchmark(device): print("\n" + "=" * 65) print("1. TENSOR-NATIVE REWARD vs. CPU-ROUNDTRIP BASELINE BENCHMARK") print("=" * 65) print(f"{'Batch Size':<12} | {'Seq Len':<10} | {'Tensor-Native (ms)':<20} | {'CPU-Decode (ms)':<18} | {'Speedup':<10}") print("-" * 65) batch_sizes = [16, 32, 64, 128, 256] seq_len = 256 pat_len = 4 results = [] for B in batch_sizes: gen_ids = torch.randint(0, 5000, (B, seq_len), device=device) patterns = [torch.randint(0, 5000, (pat_len,), device=device) for _ in range(B)] # Warmup _ = tensor_native_reward(gen_ids, patterns, eos_token_id=99, device=device) if device == "cuda": torch.cuda.synchronize() # Benchmark Tensor-Native (In-VRAM) iters = 20 t0 = time.perf_counter() for _ in range(iters): _ = tensor_native_reward(gen_ids, patterns, eos_token_id=99, device=device) if device == "cuda": torch.cuda.synchronize() t_native = (time.perf_counter() - t0) / iters * 1000 # Benchmark CPU-Decode baseline (simulate copy to CPU and string/list search) t0 = time.perf_counter() for _ in range(iters): cpu_ids = gen_ids.cpu().tolist() cpu_pats = [p.cpu().tolist() for p in patterns] rewards = [] for b in range(B): row = cpu_ids[b] pat = cpu_pats[b] match = any(row[i:i+len(pat)] == pat for i in range(len(row) - len(pat) + 1)) rewards.append(1.0 if match else 0.0) _ = torch.tensor(rewards, device=device) if device == "cuda": torch.cuda.synchronize() t_cpu = (time.perf_counter() - t0) / iters * 1000 speedup = t_cpu / max(t_native, 1e-4) print(f"{B:<12} | {seq_len:<10} | {t_native:<20.2f} | {t_cpu:<18.2f} | {speedup:<10.1f}x") results.append((B, seq_len, t_native, t_cpu, speedup)) return results def run_loss_benchmark(device): print("\n" + "=" * 65) print("2. POLICY LOSS THROUGHPUT (PPO vs. M2PO vs. GRPO)") print("=" * 65) print(f"{'Loss Type':<12} | {'Batch Size':<12} | {'Forward+Backward (ms)':<25} | {'Tokens/sec':<15}") print("-" * 65) B = 64 L = 256 iters = 30 # PPO p_lp = torch.randn(B, L, device=device, requires_grad=True) o_lp = torch.randn(B, L, device=device) adv = torch.randn(B, L, device=device) # Warmup loss = compute_ppo_loss(p_lp, o_lp, adv) loss.backward() t0 = time.perf_counter() for _ in range(iters): p_lp.grad = None loss = compute_ppo_loss(p_lp, o_lp, adv) loss.backward() if device == "cuda": torch.cuda.synchronize() t_ppo = (time.perf_counter() - t0) / iters * 1000 tok_ppo = (B * L) / (t_ppo / 1000) print(f"{'PPO':<12} | {B:<12} | {t_ppo:<25.2f} | {tok_ppo:<15.0f}") # M2PO p_lp.grad = None t0 = time.perf_counter() for _ in range(iters): p_lp.grad = None loss = compute_m2po_loss(p_lp, o_lp, adv, m2_threshold=2.0) loss.backward() if device == "cuda": torch.cuda.synchronize() t_m2po = (time.perf_counter() - t0) / iters * 1000 tok_m2po = (B * L) / (t_m2po / 1000) print(f"{'M2PO':<12} | {B:<12} | {t_m2po:<25.2f} | {tok_m2po:<15.0f}") # GRPO G = 4 B_grpo = B // G p_grp = torch.randn(B_grpo, G, L, device=device, requires_grad=True) o_grp = torch.randn(B_grpo, G, L, device=device) adv_grp = torch.randn(B_grpo, G, L, device=device) t0 = time.perf_counter() for _ in range(iters): p_grp.grad = None loss = compute_grpo_loss(p_grp, o_grp, adv_grp) loss.backward() if device == "cuda": torch.cuda.synchronize() t_grpo = (time.perf_counter() - t0) / iters * 1000 tok_grpo = (B * L) / (t_grpo / 1000) print(f"{'GRPO':<12} | {B:<12} | {t_grpo:<25.2f} | {tok_grpo:<15.0f}") def run_buffer_benchmark(): print("\n" + "=" * 65) print("3. REPLAY BUFFER THROUGHPUT BENCHMARK") print("=" * 65) buf = BoundedReplayBuffer(max_size=50000) N = 20000 exp = Experience( prompt_ids=torch.tensor([1, 2, 3]), generated_ids=torch.tensor([4, 5, 6, 7]), log_probs=torch.tensor([-0.1, -0.2, -0.1, -0.3]), reward=1.0, version=0, ) t0 = time.perf_counter() for _ in range(N): buf.push(exp) t_push = time.perf_counter() - t0 push_rate = N / t_push t0 = time.perf_counter() sampled = 0 while sampled < N: batch = buf.sample(64) if not batch: break sampled += len(batch) t_sample = time.perf_counter() - t0 sample_rate = sampled / t_sample print(f"Push Throughput: {push_rate:,.0f} ops/sec ({N} items pushed in {t_push*1000:.1f} ms)") print(f"Sample Throughput: {sample_rate:,.0f} ops/sec ({sampled} items sampled in {t_sample*1000:.1f} ms)") def run_staleness_benchmark(device): print("\n" + "=" * 65) print("4. ASYNCHRONOUS STALENESS ROBUSTNESS (PPO vs. M2PO)") print("=" * 65) print(f"{'Staleness tau':<15} | {'PPO Grad Norm':<18} | {'M2PO Grad Norm':<18} | {'Variance Reduction':<20}") print("-" * 65) B, L = 32, 64 staleness_levels = [0, 1, 2, 3, 5, 8] for tau in staleness_levels: # Drift std proportional to staleness tau drift = 0.15 * tau old_lp = torch.randn(B, L, device=device) policy_lp_ppo = (old_lp + torch.randn(B, L, device=device) * drift).clone().detach().requires_grad_(True) policy_lp_m2po = policy_lp_ppo.clone().detach().requires_grad_(True) adv = torch.randn(B, L, device=device) loss_p = compute_ppo_loss(policy_lp_ppo, old_lp, adv) loss_p.backward() p_norm = policy_lp_ppo.grad.norm().item() loss_m = compute_m2po_loss(policy_lp_m2po, old_lp, adv, m2_threshold=2.0) loss_m.backward() m_norm = policy_lp_m2po.grad.norm().item() red = ((p_norm - m_norm) / max(p_norm, 1e-6)) * 100 print(f"tau = {tau:<10} | {p_norm:<18.4f} | {m_norm:<18.4f} | {red:<20.1f}%") def main(): device = "cuda" if torch.cuda.is_available() else "cpu" print("=" * 65) print("AsyncTensorRLHF Comprehensive System Benchmark") print(f"Hardware Platform: {device.upper()}") if device == "cuda": print(f"GPU: {torch.cuda.get_device_name(0)} (Capability: {torch.cuda.get_device_capability(0)})") print("=" * 65) run_reward_benchmark(device) run_loss_benchmark(device) run_buffer_benchmark() run_staleness_benchmark(device) print("\n" + "=" * 65) print("ALL COMPREHENSIVE BENCHMARKS COMPLETED SUCCESSFULLY") print("=" * 65) if __name__ == "__main__": main()