import torch # import flashinfer import matplotlib.pyplot as plt # from sparse_sageattn import sparse_sageattn from einops import rearrange, repeat from sageattention import sageattn from spas_sage_attn import block_sparse_sage2_attn_cuda def get_cuda_arch_versions(): cuda_archs = [] for i in range(torch.cuda.device_count()): major, minor = torch.cuda.get_device_capability(i) cuda_archs.append(f"sm{major}{minor}") return cuda_archs from spas_sage_attn import block_sparse_sage2_attn_cuda def sparge_mask_convert(mask: torch.Tensor, block_size: int = 128, arch="sm") -> torch.Tensor: assert block_size in [128, 64], "Radial Attention only supports block size of 128 or 64" assert mask.shape[0] == mask.shape[1], "Input mask must be square." if block_size == 128: if arch == "sm90": new_mask = torch.repeat_interleave(mask, 2, dim=0) else: new_mask = torch.repeat_interleave(mask, 2, dim=1) elif block_size == 64: if arch == "sm90": num_row, num_col = mask.shape reshaped_mask = mask.view(num_row, num_col // 2, 2) new_mask = torch.max(reshaped_mask, dim=2).values else: num_row, num_col = mask.shape reshaped_mask = mask.view(num_row // 2, 2, num_col) new_mask = torch.max(reshaped_mask, dim=1).values return new_mask def get_indptr_from_mask(mask, query): # query shows the device of the indptr # indptr (torch.Tensor) - the block index pointer of the block-sparse matrix on row dimension, # shape `(MB + 1,)`, where `MB` is the number of blocks in the row dimension. # The first element is always 0, and the last element is the number of blocks in the row dimension. # The rest of the elements are the number of blocks in each row. # the mask is already a block sparse mask indptr = torch.zeros(mask.shape[0] + 1, device=query.device, dtype=torch.int32) indptr[0] = 0 row_counts = mask.sum(dim=1).flatten() # Ensure 1D output [num_blocks_row] indptr[1:] = torch.cumsum(row_counts, dim=0) return indptr def get_indices_from_mask(mask, query): # indices (torch.Tensor) - the block indices of the block-sparse matrix on column dimension, # shape `(nnz,),` where `nnz` is the number of non-zero blocks. # The elements in `indices` array should be less than `NB`: the number of blocks in the column dimension. nonzero_indices = torch.nonzero(mask) indices = nonzero_indices[:, 1].to(dtype=torch.int32, device=query.device) return indices def shrinkMaskStrict(mask, block_size=128): seqlen = mask.shape[0] block_num = seqlen // block_size mask = mask[:block_num * block_size, :block_num * block_size].view(block_num, block_size, block_num, block_size) col_densities = mask.sum(dim = 1) / block_size # we want the minimum non-zero column density in the block non_zero_densities = col_densities > 0 high_density_cols = col_densities > 1/3 frac_high_density_cols = high_density_cols.sum(dim=-1) / (non_zero_densities.sum(dim=-1) + 1e-9) block_mask = frac_high_density_cols > 0.6 block_mask[0:0] = True block_mask[-1:-1] = True return block_mask def pad_qkv(input_tensor, block_size=128): """ Pad the input tensor to be a multiple of the block size. input shape: (seqlen, num_heads, hidden_dim) """ seqlen, num_heads, hidden_dim = input_tensor.shape # Calculate the necessary padding padding_length = (block_size - (seqlen % block_size)) % block_size # Create a padded tensor with zeros padded_tensor = torch.zeros((seqlen + padding_length, num_heads, hidden_dim), device=input_tensor.device, dtype=input_tensor.dtype) # Copy the original tensor into the padded tensor padded_tensor[:seqlen, :, :] = input_tensor return padded_tensor def get_diagonal_split_mask(i, j, token_per_frame, sparse_type, query): assert(sparse_type in ["radial"]) dist = abs(i - j) group = dist.bit_length() threshold = 128 # hardcoded threshold for now, which is equal to block-size decay_length = 2 ** token_per_frame.bit_length() / 2 ** group if decay_length >= threshold: return torch.ones((token_per_frame, token_per_frame), device=query.device, dtype=torch.bool) split_factor = int(threshold / decay_length) modular = dist % split_factor if modular == 0: return torch.ones((token_per_frame, token_per_frame), device=query.device, dtype=torch.bool) else: return torch.zeros((token_per_frame, token_per_frame), device=query.device, dtype=torch.bool) def get_window_width(i, j, token_per_frame, sparse_type, num_frame, decay_factor=1, block_size=128, model_type=None): assert(sparse_type in ["radial"]) dist = abs(i - j) if model_type == "wan": if dist < 1: return token_per_frame if dist == 1: return token_per_frame // 2 elif model_type == "hunyuan": if dist <= 1: return token_per_frame else: raise ValueError(f"Unknown model type: {model_type}") group = dist.bit_length() decay_length = 2 ** token_per_frame.bit_length() / 2 ** group * decay_factor threshold = block_size if decay_length >= threshold: return decay_length else: return threshold def gen_log_mask_shrinked(query, s, video_token_num, num_frame, block_size=128, sparse_type="log", decay_factor=0.5, model_type=None): """ A more memory friendly version, we generate the attention mask of each frame pair at a time, shrinks it, and stores it into the final result """ final_log_mask = torch.zeros((s // block_size, s // block_size), device=query.device, dtype=torch.bool) token_per_frame = video_token_num // num_frame video_text_border = video_token_num // block_size col_indices = torch.arange(0, token_per_frame, device=query.device).view(1, -1) row_indices = torch.arange(0, token_per_frame, device=query.device).view(-1, 1) final_log_mask[video_text_border:] = True final_log_mask[:, video_text_border:] = True for i in range(num_frame): for j in range(num_frame): local_mask = torch.zeros((token_per_frame, token_per_frame), device=query.device, dtype=torch.bool) if j == 0 and model_type == "wan": # this is attention sink local_mask = torch.ones((token_per_frame, token_per_frame), device=query.device, dtype=torch.bool) else: window_width = get_window_width(i, j, token_per_frame, sparse_type, num_frame, decay_factor=decay_factor, block_size=block_size, model_type=model_type) local_mask = torch.abs(col_indices - row_indices) <= window_width split_mask = get_diagonal_split_mask(i, j, token_per_frame, sparse_type, query) local_mask = torch.logical_and(local_mask, split_mask) remainder_row = (i * token_per_frame) % block_size remainder_col = (j * token_per_frame) % block_size # get the padded size all_length_row = remainder_row + ((token_per_frame - 1) // block_size + 1) * block_size all_length_col = remainder_col + ((token_per_frame - 1) // block_size + 1) * block_size padded_local_mask = torch.zeros((all_length_row, all_length_col), device=query.device, dtype=torch.bool) padded_local_mask[remainder_row:remainder_row + token_per_frame, remainder_col:remainder_col + token_per_frame] = local_mask # shrink the mask block_mask = shrinkMaskStrict(padded_local_mask, block_size=block_size) # set the block mask to the final log mask block_row_start = (i * token_per_frame) // block_size block_col_start = (j * token_per_frame) // block_size block_row_end = block_row_start + block_mask.shape[0] block_col_end = block_col_start + block_mask.shape[1] final_log_mask[block_row_start:block_row_end, block_col_start:block_col_end] = torch.logical_or( final_log_mask[block_row_start:block_row_end, block_col_start:block_col_end], block_mask) print(f"mask sparsity: {1 - final_log_mask.sum() / final_log_mask.numel()}") return final_log_mask class MaskMap: _log_mask = None def __init__(self, video_token_num=25440, num_frame=16): self.video_token_num = video_token_num self.num_frame = num_frame def queryLogMask(self, query, sparse_type, block_size=128, decay_factor=0.5, model_type=None): if MaskMap._log_mask is None: MaskMap._log_mask = torch.ones((query.shape[0] // block_size, query.shape[0] // block_size), device=query.device, dtype=torch.bool) MaskMap._log_mask = gen_log_mask_shrinked(query, query.shape[0], self.video_token_num, self.num_frame, sparse_type=sparse_type, decay_factor=decay_factor, model_type=model_type, block_size=block_size) return MaskMap._log_mask def SpargeSageAttnBackend(query, key, value, mask_map=None, video_mask=None, pre_defined_mask=None, block_size=128): if video_mask.all(): # dense case kv_border = pre_defined_mask[0].sum() if pre_defined_mask is not None else key.shape[0] output_video = sageattn( query[:mask_map.video_token_num, :, :].unsqueeze(0), key[:kv_border, :, :].unsqueeze(0), value[:kv_border, :, :].unsqueeze(0), tensor_layout="NHD", )[0] if pre_defined_mask is not None: output_text = flashinfer.single_prefill_with_kv_cache( q=query[mask_map.video_token_num:, :, :], k=key[:pre_defined_mask[0].sum(), :, :], v=value[:pre_defined_mask[0].sum(), :, :], causal=False, return_lse=False, ) return torch.cat([output_video, output_text], dim=0) else: return output_video # sparse-sageattention only supports (b, h, s, d) layout, need rearrange first query_hnd = rearrange(query.unsqueeze(0), "b s h d -> b h s d") key_hnd = rearrange(key.unsqueeze(0), "b s h d -> b h s d") value_hnd = rearrange(value.unsqueeze(0), "b s h d -> b h s d") arch = get_cuda_arch_versions()[query.device.index] converted_mask = repeat(sparge_mask_convert(mask=video_mask, block_size=block_size, arch=arch), "s t -> b h s t", b=query_hnd.shape[0], h=query_hnd.shape[1]) converted_mask = converted_mask.to(torch.int8) if pre_defined_mask is None: # wan case output = block_sparse_sage2_attn_cuda( query_hnd[:, :, :mask_map.video_token_num, :], key_hnd[:, :, :mask_map.video_token_num, :], value_hnd[:, :, :mask_map.video_token_num, :], mask_id=converted_mask, tensor_layout="HND", ) # rearrange back to (s, h, d), we know that b = 1 output = rearrange(output, "b h s d -> s (b h) d", b=1) return output query_video = query_hnd[:, :, :mask_map.video_token_num, :] key_video = key_hnd value_video = value_hnd kv_border = (pre_defined_mask[0].sum() + 63) // 64 converted_mask[:, :, :, kv_border:] = False output_video = block_sparse_sage2_attn_cuda( query_video, key_video, value_video, mask_id=converted_mask[:, :, :mask_map.video_token_num // block_size, :].contiguous(), tensor_layout="HND", ) # rearrange back to (s, h, d), we know that b = 1 output_video = rearrange(output_video, "b h s d -> s (b h) d", b=1) # gt = sparse_sageattn( # query_video, # key_video, # value_video, # mask_id=None, # is_causal=False, # tensor_layout="HND", # )[0] # import pdb; pdb.set_trace() output_text = flashinfer.single_prefill_with_kv_cache( q=query[mask_map.video_token_num:, :, :], k=key[:pre_defined_mask[0].sum(), :, :], v=value[:pre_defined_mask[0].sum(), :, :], causal=False, return_lse=False, ) return torch.cat([output_video, output_text], dim=0) def FlashInferBackend(query, key, value, mask_map=None, pre_defined_mask=None, bsr_wrapper=None): if pre_defined_mask is not None: video_video_o, video_video_o_lse = bsr_wrapper.run( query[:mask_map.video_token_num, :, :], key[:mask_map.video_token_num, :, :], value[:mask_map.video_token_num, :, :], return_lse=True ) # perform non-causal flashinfer on the text tokens video_text_o, video_text_o_lse = flashinfer.single_prefill_with_kv_cache( q=query[:mask_map.video_token_num, :, :], k=key[mask_map.video_token_num:, :, :], v=value[mask_map.video_token_num:, :, :], causal=False, return_lse=True, custom_mask=pre_defined_mask[:mask_map.video_token_num, mask_map.video_token_num:] ) # merge the two results o_video, _ = flashinfer.merge_state(v_a=video_video_o, s_a=video_video_o_lse, v_b=video_text_o, s_b=video_text_o_lse) o_text = flashinfer.single_prefill_with_kv_cache( q=query[mask_map.video_token_num:, :, :], k=key, v=value, causal=False, return_lse=False, custom_mask=pre_defined_mask[mask_map.video_token_num:, :] ) return torch.cat([o_video, o_text], dim=0) else: o = bsr_wrapper.run( query[:mask_map.video_token_num, :, :], key[:mask_map.video_token_num, :, :], value[:mask_map.video_token_num, :, :] ) return o def RadialAttention(query, key, value, mask_map=None, sparsity_type="radial", block_size=128, decay_factor=1, model_type=None, pre_defined_mask=None, use_sage_attention=False): orig_seqlen, num_head, hidden_dim = query.shape if sparsity_type == "dense": video_mask = torch.ones((mask_map.video_token_num // block_size, mask_map.video_token_num // block_size), device=query.device, dtype=torch.bool) else: video_mask = mask_map.queryLogMask(query, sparsity_type, block_size=block_size, decay_factor=decay_factor, model_type=model_type) if mask_map else None backend = "sparse_sageattn" if use_sage_attention else "flashinfer" if backend == "flashinfer": video_mask = video_mask[:mask_map.video_token_num // block_size, :mask_map.video_token_num // block_size] # perform block-sparse attention on the video tokens workspace_buffer = torch.empty(128 * 1024 * 1024, device=query.device, dtype=torch.uint8) bsr_wrapper = flashinfer.BlockSparseAttentionWrapper( workspace_buffer, backend="fa2", ) indptr = get_indptr_from_mask(video_mask, query) indices = get_indices_from_mask(video_mask, query) bsr_wrapper.plan( indptr=indptr, indices=indices, M=mask_map.video_token_num, N=mask_map.video_token_num, R=block_size, C=block_size, num_qo_heads=num_head, num_kv_heads=num_head, head_dim=hidden_dim, q_data_type=query.dtype, kv_data_type=key.dtype, o_data_type=query.dtype, ) return FlashInferBackend(query, key, value, mask_map, pre_defined_mask, bsr_wrapper) elif backend == "sparse_sageattn": return SpargeSageAttnBackend(query, key, value, mask_map, video_mask, pre_defined_mask, block_size=block_size) if __name__ == "__main__": query = torch.randn(1, 2, 4, 64).cuda() # mask = torch.tensor([ # [True, False, True, False], # [False, True, False, True], # [True, False, False, True], # [False, True, True, False] # ], dtype=torch.bool) # indices = get_indices_from_mask(mask, query) # indptr = get_indptr_from_mask(mask, query) # print("Indices: ", indices) # print("Indptr: ", indptr) video_token_num = 3840 * 30 num_frame = 30 token_per_frame = video_token_num / num_frame padded_video_token_num = ((video_token_num + 1) // 128 + 1) * 128 print("padded: ", padded_video_token_num) temporal_mask = gen_log_mask_shrinked(query, padded_video_token_num, video_token_num, num_frame, sparse_type="radial", decay_factor=1, model_type="hunyuan") plt.figure(figsize=(10, 8), dpi=500) plt.imshow(temporal_mask.cpu().numpy()[:, :], cmap='hot') plt.colorbar() plt.title("Temporal Mask") plt.savefig("temporal_mask.png", dpi=300, bbox_inches='tight', pad_inches=0.1) plt.close() # save the mask tensor torch.save(temporal_mask, "temporal_mask.pt")