"""Masked / cached attention for the MiniMax-H3 blocks. musubi's H3 Attention runs dense FlashAttention over the whole packed sequence. install() patches its forward with two extra modes, selected per block call through `active(...)`: mask : torch flex_attention with a BlockMask over the whole packed sequence (training, reference) dense: a masks.DenseDF instead of a BlockMask -- the same training mask as FlashAttention-3 prefix calls kv : the call's rows attend to cached keys/values of earlier rows plus their own (KV-cache sampling, FlashAttention-3; see KV_ATTENTION); with store=list the call's own keys/values are appended per layer. A mask may be combined with kv to fill the cache (chunk 0 together with the condition rows). The mode is set inside each block call (not around the loop) so that activation-checkpoint recomputation in backward sees it too. """ from __future__ import annotations import contextlib import torch import torch.nn.functional as F from torch.nn.attention.flex_attention import flex_attention from musubi_tuner.minimax_h3 import model as h3model from .layout import CHUNK_SIZES from .masks import CHUNK_ROWS, DenseDF _compiled_flex = torch.compile(flex_attention, dynamic=True) # KV-cache attention over the concatenated past: "fa3" FlashAttention-3 (23% faster per sampled chunk); # "sdpa" torch SDPA (the earlier behaviour, kept for tests) KV_ATTENTION = "fa3" _state = {"mask": None, "kv": None} _orig_forward = h3model.Attention.forward _orig_rope = h3model._apply_rope_split_half # kept for tests/test_samplers.py def _forward(self, hidden_states, rotation_table=None): mask, kv = _state["mask"], _state["kv"] if mask is None and kv is None: return _orig_forward(self, hidden_states, rotation_table) b, s, _ = hidden_states.shape q, k, v = self.qkv_proj(hidden_states).split(self.inner_dim, dim=-1) q = self.q_norm(q.reshape(b, s, self.num_heads, self.head_dim)) k = self.k_norm(k.reshape(b, s, self.num_heads, self.head_dim)) v = v.reshape(b, s, self.num_heads, self.head_dim) if rotation_table is not None: q = rope(q, rotation_table) k = rope(k, rotation_table) dt = hidden_states.dtype if hidden_states.dtype in (torch.float16, torch.bfloat16) else torch.bfloat16 if isinstance(mask, DenseDF): out = dense_df(q.to(dt), k.to(dt), v.to(dt), mask) return self.out_proj(out.reshape(b, s, self.inner_dim)) q, k, v = (t.transpose(1, 2).contiguous().to(dt) for t in (q, k, v)) # [B,H,S,D] if kv is not None and kv.get("store") is not None: kv["store"].append((k, v)) if mask is not None: out = _compiled_flex(q, k, v, block_mask=mask) else: pieces = kv["past"][self._rtcwm_layer] if kv.get("past") is not None else [] if kv.get("causal_blocks"): # the call's rows are blocks of n rows, block p seeing the n = kv["causal_blocks"] # past and blocks <= p (the proxy rows of a span) pk = torch.cat([_key(p) for p in pieces], 2).transpose(1, 2) pv = torch.cat([p[1] for p in pieces], 2).transpose(1, 2) qs, ks, vs = (t.transpose(1, 2) for t in (q, k, v)) out = torch.cat([_fa3(qs[:, i:i + n], torch.cat([pk, ks[:, :i + n]], 1), torch.cat([pv, vs[:, :i + n]], 1)) for i in range(0, s, n)], 1) return self.out_proj(out.reshape(b, s, self.inner_dim)) if pieces: # concatenated per layer: one layer's copy at a time; k = torch.cat([_key(p) for p in pieces] + [k], 2) # a piece may carry a pending time shift v = torch.cat([p[1] for p in pieces] + [v], 2) if KV_ATTENTION == "sdpa": out = F.scaled_dot_product_attention(q, k, v).transpose(1, 2) else: out = _fa3(q.transpose(1, 2), k.transpose(1, 2), v.transpose(1, 2)) return self.out_proj(out.reshape(b, s, self.inner_dim)) return self.out_proj(out.transpose(1, 2).reshape(b, s, self.inner_dim)) def _key(piece): """Cached keys [B,H,S,D] of a piece (k, v) or (k, v, rotation): the rotation (rtcwm.model.sampler._moved) moves them on the time axis -- RoPE rotations compose -- without storing a moved copy.""" if len(piece) == 2: return piece[0] return rope(piece[0].transpose(1, 2), piece[2]).transpose(1, 2) def _fa3(q, k, v): from flash_attn_interface import flash_attn_func out = flash_attn_func(q, k, v) return out[0] if isinstance(out, tuple) else out def dense_df(q, k, v, lay: DenseDF): """Attention of a packed training span [head | DUV 0..5 | audio | video 0..n-1] under df_mask, as unmasked FlashAttention-3 calls over key prefixes (see masks.DenseDF). q, k, v [B,S,H,D].""" h, a = lay.head_rows, lay.audio_rows dr, cr = lay.duv_rows // len(CHUNK_SIZES), CHUNK_ROWS d1 = h + lay.duv_rows v0 = d1 + a assert q.shape[1] == v0 + lay.n_chunks * cr, (q.shape, lay) duv = lambda t, p: t[:, h + p * dr:h + (p + 1) * dr] vid = lambda t, c: t[:, v0 + c * cr:v0 + (c + 1) * cr] def main(t): parts = [t[:, :h], t[:, d1:v0]] for c in range(lay.n_chunks): parts += [duv(t, c), vid(t, c)] return torch.cat(parts, 1) km, vm = main(k), main(v) kp, vp = (torch.cat([t[:, :h], t[:, d1:v0], t[:, h:d1]], 1) for t in (k, v)) st = h + a # static keys lead both orders n = st + (dr + cr if lay.static_sees_first else 0) o_st = _fa3(torch.cat([q[:, :h], q[:, d1:v0]], 1), km[:, :n], vm[:, :n]) o_duv = [_fa3(duv(q, p), kp[:, :st + (p + 1) * dr], vp[:, :st + (p + 1) * dr]) for p in range(len(CHUNK_SIZES))] o_vid = [_fa3(vid(q, c), km[:, :st + (c + 1) * (dr + cr)], vm[:, :st + (c + 1) * (dr + cr)]) for c in range(lay.n_chunks)] return torch.cat([o_st[:, :h], *o_duv, o_st[:, h:], *o_vid], 1) def rope(hidden_states, rotation_table): """musubi's _apply_rope_split_half written element-wise. The original multiplies every (token, head, pair) by its 2x2 rotation with a batched matmul -- one tiny cuBLAS gemv per few thousand elements, ~40% of a sampling step. Same rotation table, accumulated in fp32 like the matmul.""" pairs = rotation_table.shape[-3] r = rotation_table.float() # [B,S,1,pairs,2,2] = [[cos, -sin], [sin, cos]] a, b = hidden_states[..., :pairs].float(), hidden_states[..., pairs:2 * pairs].float() rotated = torch.cat((r[..., 0, 0] * a + r[..., 0, 1] * b, r[..., 1, 0] * a + r[..., 1, 1] * b), dim=-1) return torch.cat((rotated.to(hidden_states.dtype), hidden_states[..., 2 * pairs:]), dim=-1) _orig_mod_scale_shift, _orig_mod_gate = h3model._mod_scale_shift, h3model._mod_gate # kept for tests _row_index = {} def _rows(batch_segments, device): """Per-row modulation index of the (single-sample) segment plan, built once per forward: every block of a forward call shares the same plan object.""" key = (id(batch_segments), device) hit = _row_index.get(key) if hit is not None and hit[0] is batch_segments: return hit[1] (segments,) = batch_segments seg = torch.tensor(segments, dtype=torch.long) idx = torch.repeat_interleave(seg[:, 2], seg[:, 1] - seg[:, 0]).to(device, non_blocking=True) _row_index.clear() _row_index[key] = (batch_segments, idx) return idx def mod_scale_shift(hidden_states, shift, scale, batch_segments): """musubi's _mod_scale_shift without the Python loop over segments (hundreds of small kernels per block with the text's token tags): the same x * (1 + scale) + shift, gathered per row.""" idx = _rows(batch_segments, hidden_states.device) return hidden_states * (1.0 + scale[idx]).unsqueeze(0) + shift[idx].unsqueeze(0) def mod_gate(residual, update, gate, batch_segments): """musubi's _mod_gate, vectorised the same way: residual + update * gate.""" idx = _rows(batch_segments, residual.device) return residual + update * gate[idx].unsqueeze(0) def install(model) -> None: """Patch H3 attention, its RoPE and its AdaLN modulation (idempotent) and number the layers for the KV cache.""" h3model.Attention.forward = _forward h3model._apply_rope_split_half = rope # the released (unmasked) forward too h3model._mod_scale_shift, h3model._mod_gate = mod_scale_shift, mod_gate for i, blk in enumerate(model.blocks): (blk.attn if hasattr(blk, "attn") else blk.module.attn)._rtcwm_layer = i # FSDP-wrapped or not @contextlib.contextmanager def active(mask=None, kv=None): _state["mask"], _state["kv"] = mask, kv try: yield finally: _state["mask"], _state["kv"] = None, None