import functools import os import re import torch _ENABLED = os.getenv("MAGI_TIRX_DOWN", "0") == "1" _VALIDATE = os.getenv("MAGI_TIRX_DOWN_VALIDATE", "0") == "1" _checked = set() _reported = False KP = 13696 N = 5120 @functools.lru_cache(maxsize=2) def executable(mp): import tirx_kernels.gemm.fp16_bf16_gemm as gemm from tirx_kernels.runner import compile_kernel cfg = dict(gemm.GEMM_CONFIGS[1024]) cfg.update( cta_n=64, cta_k=128, pipe_depth=5, wb_pipe_depth=2, l2_group_size=2, overlap_epilogue=True, ) original_cfg = gemm._cfg_for try: gemm._cfg_for = lambda n: dict(cfg) kernel = gemm.make_kernel("bf16", mp, N, KP) result = compile_kernel(kernel) finally: gemm._cfg_for = original_cfg print(f"TIRX_DOWN_COMPILED M={mp} N={N} K={KP}", flush=True) return result def supported(module, z, weight): return ( _ENABLED and not torch.is_grad_enabled() and not module.training and z.is_cuda and weight.device == z.device and z.dtype == torch.bfloat16 and weight.dtype == torch.bfloat16 and z.ndim == 3 and weight.ndim == 3 and z.shape[0] == 3 and 0 < z.shape[1] <= 4096 and z.shape[2] == 13652 and tuple(weight.shape) == (3, 5120, 13652) and z.is_contiguous() and weight.is_contiguous() and not z.requires_grad and not weight.requires_grad ) def down(module, z, weight): global _reported m = z.shape[1] mp = 2048 if m <= 2048 else 4096 # 保存原权重对象,避免地址复用导致误用缓存。 # 权重被原地修改后,通过 version 重新建立缓存。 # inference tensor没有版本计数;固定权重推理按对象身份缓存。 version = None if torch.is_inference(weight) else weight._version cached = getattr(module, "_tirx_down_weight_cache", None) if ( cached is None or cached[0] is not weight or cached[1] != version ): padded_weight = torch.zeros( 3, N, KP, dtype=weight.dtype, device=weight.device ) padded_weight[:, :, :13652].copy_(weight) module._tirx_down_weight_cache = ( weight, version, padded_weight ) else: padded_weight = cached[2] # 全部清零,确保 K 尾部和 M 尾部没有未初始化值。 padded_input = torch.zeros( 3, mp, KP, dtype=z.dtype, device=z.device ) padded_input[:, :m, :13652].copy_(z) padded_output = torch.empty( 3, mp, N, dtype=z.dtype, device=z.device ) ex = executable(mp) for e in range(3): ex(padded_input[e], padded_weight[e], padded_output[e]) output = padded_output[:, :m, :].contiguous() if not _reported: print( f"TIRX_DOWN_ACTIVE input={tuple(z.shape)} " f"padded_M={mp} padded_K={KP}", flush=True, ) _reported = True # 检查前三种真实形状;这不是全模型质量验证。 key = tuple(z.shape) if _VALIDATE and key not in _checked and len(_checked) < 3: expected = torch.bmm(z, weight.transpose(-2, -1)) finite = bool(torch.isfinite(output).all().item()) # 分块统计,控制验证时的临时显存。 a = output.reshape(-1) b = expected.reshape(-1) max_abs = 0.0 error_sq = 0.0 reference_sq = 0.0 for start in range(0, a.numel(), 1 << 20): af = a[start:start + (1 << 20)].float() bf = b[start:start + (1 << 20)].float() delta = af - bf max_abs = max(max_abs, delta.abs().max().item()) error_sq += delta.square().sum(dtype=torch.float64).item() reference_sq += bf.square().sum(dtype=torch.float64).item() relative_l2 = (error_sq / max(reference_sq, 1e-30)) ** 0.5 print( f"TIRX_DOWN_CHECK input={tuple(z.shape)} " f"finite={finite} max_abs={max_abs:.8g} " f"relative_l2={relative_l2:.8g}", flush=True, ) # 仅用于拦截明显异常,不能代替质量评价。 if not finite or relative_l2 > 1e-3: raise RuntimeError("TIRx down真实输入检查未通过") _checked.add(key) return output def transform_source(source): pattern = re.compile( r"^(?P[ \t]*)" r"y\s*=\s*torch\.bmm\(" r"z,\s*w_dn\.transpose\(-2,\s*-1\)" r"\)\.to\(dtype\)[ \t]*$", re.MULTILINE, ) matches = list(pattern.finditer(source)) if len(matches) != 1: raise RuntimeError( f"down BMM入口匹配到{len(matches)}处,未安装补丁" ) match = matches[0] indent = match.group("indent") replacement = ( indent + "if _tirx_down_supported(self, z, w_dn):\n" + indent + " y = _tirx_down(self, z, w_dn)\n" + indent + "else:\n" + indent + " y = torch.bmm(z, w_dn.transpose(-2, -1)).to(dtype)" ) return source[:match.start()] + replacement + source[match.end():] @torch.inference_mode() def prewarm_model(model): from inference.model.dit.dit_module import SparseMoEMLP modules = [ m for m in model.modules() if isinstance(m, SparseMoEMLP) ] if not modules: raise RuntimeError("down预热未找到SparseMoEMLP") print( f"TIRX_PREWARM_DOWN: 开始 modules={len(modules)} " f"enabled={_ENABLED}", flush=True, ) first = True for index, module in enumerate(modules): if module.training: raise RuntimeError("down预热要求eval模式") experts = module.experts device = experts[0].down_proj.weight.device if device.type != "cuda": raise RuntimeError("down预热要求模型已在GPU,当前不支持CPU offload") if ( len(experts) != 3 or any( tuple(e.down_proj.weight.shape) != (5120, 13652) for e in experts ) ): raise RuntimeError("down预热只适用于当前三专家、全宽13652模型") # 与原始 _cache_key("bmm", None) 完全一致。 key = ("bmm", torch.bfloat16, device.type, device.index, -1) cache = module._grouped_weight_cache if key not in cache: up = torch.stack( [e.up_gate_proj.weight.to(torch.bfloat16) for e in experts], dim=0, ) dn = torch.stack( [e.down_proj.weight.to(torch.bfloat16) for e in experts], dim=0, ) cache[key] = (up, dn) if _ENABLED: weight = cache[key][1] version = ( None if torch.is_inference(weight) else weight._version ) padded = torch.zeros( 3, N, KP, dtype=torch.bfloat16, device=device ) padded[:, :, :13652].copy_(weight) module._tirx_down_weight_cache = (weight, version, padded) # 两档内核各启动一次;不调用down(),避免消耗真实输入检查名额。 if first: for mp in (2048, 4096): x = torch.zeros( mp, KP, dtype=torch.bfloat16, device=device ) y = torch.empty( mp, N, dtype=torch.bfloat16, device=device ) ex = executable(mp) for _ in range(3): ex(x, padded[0], y) torch.cuda.synchronize(device) del x, y first = False print( f"TIRX_PREWARM_DOWN: 权重准备 {index + 1}/{len(modules)}", flush=True, ) torch.cuda.synchronize() print("TIRX_PREWARM_DOWN: 完成", flush=True)