diff --git a/tests/unit/test_d2_scale.py b/tests/unit/test_d2_scale.py new file mode 100644 index 00000000..c9733b3a --- /dev/null +++ b/tests/unit/test_d2_scale.py @@ -0,0 +1,106 @@ +""" +D2: Scale test — more heads, larger hd. +""" +import torch, math +import cutlass.cute as cute +import cutlass.torch as ct +import cuda.bindings.driver as cuda +from dsv4.kernels.attention.fmha import FmhaKernel + + +def test_multihead(hd=64, n_h=1, batch=1, T=128, s_k=128): + torch.manual_seed(42) + + q = torch.randn(batch, n_h, T, hd, dtype=torch.bfloat16, device='cuda') + k = torch.randn(batch, s_k, hd, dtype=torch.bfloat16, device='cuda') + v = torch.randn(batch, s_k, hd, dtype=torch.bfloat16, device='cuda') + + # FP32 reference (un-normalized) + qf = q.float() + kf = k.float() + vf = v.float() + scale = 1.0 / math.sqrt(hd) + + ref_unnorm = torch.zeros(batch, n_h, T, hd, dtype=torch.float32, device='cuda') + for b in range(batch): + for h in range(n_h): + attn = qf[b, h] @ kf[b].T * scale + attn_max = attn.max(dim=-1, keepdim=True)[0] + attn_exp = torch.exp(attn - attn_max) + ref_unnorm[b, h] = attn_exp @ vf[b] + + kernel = FmhaKernel(head_dim=hd, s_k=s_k, use_smem_p=False, normalize=False) + pv_n_tile = kernel.pv_n_tile + n_pv_tiles = hd // pv_n_tile + stream = cuda.CUstream(torch.cuda.current_stream().cuda_stream) + + # Compile once + q0 = q[0, 0].unsqueeze(-1) + k0 = k[0].unsqueeze(-1) + v0_tile = v[0, :, 0:pv_n_tile].contiguous() + v0_k = v0_tile.unsqueeze(-1) + c0 = torch.zeros(T, pv_n_tile, 1, dtype=torch.bfloat16, device='cuda') + lse0 = torch.zeros(T, 1, 1, dtype=torch.float32, device='cuda') + + mQ = ct.from_dlpack(q0).mark_layout_dynamic(leading_dim=ct.get_leading_dim(q0)) + mK = ct.from_dlpack(k0).mark_layout_dynamic(leading_dim=ct.get_leading_dim(k0)) + mV = ct.from_dlpack(v0_k).mark_layout_dynamic(leading_dim=ct.get_leading_dim(v0_k)) + mC = ct.from_dlpack(c0).mark_layout_dynamic(leading_dim=ct.get_leading_dim(c0)) + mLSE = ct.from_dlpack(lse0).mark_layout_dynamic(leading_dim=ct.get_leading_dim(lse0)) + + print(f' Compiling (hd={hd}, n_h={n_h})...', flush=True) + compiled = cute.compile(kernel, mQ, mK, mV, mC, stream, mLSE) + + o = torch.zeros(batch, n_h, T, hd, dtype=torch.bfloat16, device='cuda') + + for b in range(batch): + for h in range(n_h): + q_h = q[b, h].unsqueeze(-1) + k_b = k[b].unsqueeze(-1) + v_b = v[b] + c_h = torch.zeros(T, hd, dtype=torch.bfloat16, device='cuda') + + for nt in range(n_pv_tiles): + v_start = nt * pv_n_tile + v_end = v_start + pv_n_tile + v_tile = v_b[:, v_start:v_end].contiguous() + v_k = v_tile.unsqueeze(-1) + c_tile = torch.zeros(T, pv_n_tile, 1, dtype=torch.bfloat16, device='cuda') + lse0.zero_() + + mQ = ct.from_dlpack(q_h).mark_layout_dynamic(leading_dim=ct.get_leading_dim(q_h)) + mK = ct.from_dlpack(k_b).mark_layout_dynamic(leading_dim=ct.get_leading_dim(k_b)) + mV = ct.from_dlpack(v_k).mark_layout_dynamic(leading_dim=ct.get_leading_dim(v_k)) + mC = ct.from_dlpack(c_tile).mark_layout_dynamic(leading_dim=ct.get_leading_dim(c_tile)) + mLSE = ct.from_dlpack(lse0).mark_layout_dynamic(leading_dim=ct.get_leading_dim(lse0)) + + compiled(mQ, mK, mV, mC, stream, mLSE) + c_h[:, v_start:v_end] = c_tile[:, :, 0] + + o[b, h] = c_h + + torch.cuda.synchronize() + + cos = torch.nn.functional.cosine_similarity( + o.flatten().float().unsqueeze(0), ref_unnorm.flatten().unsqueeze(0) + ).item() + print(f' hd={hd}, n_h={n_h}, batch={batch}: cos {cos:.6f} {"PASS" if cos >= 0.99 else "FAIL"}') + + +def test(): + print("=== D2: Scale Test ===\n") + + # hd=64 + test_multihead(64, 16, 1, 128, 128) + test_multihead(64, 64, 1, 128, 128) # Flash decode config (hd=64 test) + + # hd=128 + test_multihead(128, 2, 1, 128, 128) + test_multihead(128, 8, 1, 128, 128) + + # hd=256 + test_multihead(256, 2, 1, 128, 128) + + +if __name__ == '__main__': + test()