""" FLUXNATION — Fused CUDA kernel for FLUX.1 SingleStreamBlock. Architecture: PyTorch C++ Extension via torch.ops (no ctypes) - torch.ops.flux_ext.ssb_prepare: modulation + L1 GEMM + split QKV + RoPE - torch.ops.flux_ext.ssb_finish: L2 GEMM + fused gate+residual+bias Build: run build_ext.bat (or: py -3.10 setup_ext.py install) """ class FLUXNATION: """Main FLUXNATION kernel node. Pass-through MODEL — kernel is already active.""" CATEGORY = "FLUXNATION" RETURN_TYPES = ("MODEL",) RETURN_NAMES = ("model",) FUNCTION = "apply" @classmethod def INPUT_TYPES(cls): return {"required": {"model": ("MODEL",)}} def apply(self, model): from .FLUXPATCH import _spike_controller spike = "SPIKE ON" if _spike_controller.enabled else "SPIKE OFF" print(f"[FLUXNATION] FP8 fused CUDA kernel ACTIVE | {spike}") return (model,) class FLUXATTENHUT_Node: """Configure spike attention parameters.""" CATEGORY = "FLUXNATION" RETURN_TYPES = ("MODEL",) RETURN_NAMES = ("model",) FUNCTION = "configure" @classmethod def INPUT_TYPES(cls): return { "required": { "model": ("MODEL",), "switch_pct": ("FLOAT", {"default": 0.50, "min": 0.1, "max": 0.9, "step": 0.05, "tooltip": "Fraction of steps before spike activates (0.50 = step 10/20)"}), "cap_pct": ("FLOAT", {"default": 0.45, "min": 0.1, "max": 0.9, "step": 0.05, "tooltip": "Fraction of attention blocks to keep (0.45 = 45%)"}), "tau": ("FLOAT", {"default": 0.05, "min": 0.01, "max": 0.5, "step": 0.01, "tooltip": "Block scoring threshold — lower = more blocks kept"}), } } def configure(self, model, switch_pct, cap_pct, tau): from .FLUXPATCH import _spike_controller _spike_controller.switch_pct = switch_pct _spike_controller.event_cap_pct = cap_pct _spike_controller.tau = tau print(f"[FLUXATTENHUT] switch={switch_pct*100:.0f}% cap={cap_pct*100:.0f}% tau={tau}") return (model,) class FLUXSPIKEY_Node: """Enable or disable spike attention + step caching.""" CATEGORY = "FLUXNATION" RETURN_TYPES = ("MODEL",) RETURN_NAMES = ("model",) FUNCTION = "configure" @classmethod def INPUT_TYPES(cls): return { "required": { "model": ("MODEL",), "spike": ("BOOLEAN", {"default": True, "tooltip": "Enable block-sparse spike attention"}), "step_cache": ("BOOLEAN", {"default": True, "tooltip": "Cache attention output on alternating spike steps"}), } } def configure(self, model, spike, step_cache): from .FLUXPATCH import _spike_controller _spike_controller.enabled = spike _spike_controller.cache_enabled = step_cache print(f"[FLUXSPIKEY] spike={'ON' if spike else 'OFF'} cache={'ON' if step_cache else 'OFF'}") return (model,) class ULT7RAFLUX: """Master status node — shows kernel state, spike config, and timing info.""" CATEGORY = "FLUXNATION" RETURN_TYPES = ("MODEL",) RETURN_NAMES = ("model",) FUNCTION = "status" @classmethod def INPUT_TYPES(cls): return { "required": { "model": ("MODEL",), "spike": ("BOOLEAN", {"default": True}), "step_cache": ("BOOLEAN", {"default": True}), "switch_pct": ("FLOAT", {"default": 0.50, "min": 0.1, "max": 0.9, "step": 0.05}), "cap_pct": ("FLOAT", {"default": 0.45, "min": 0.1, "max": 0.9, "step": 0.05}), "tau": ("FLOAT", {"default": 0.05, "min": 0.01, "max": 0.5, "step": 0.01}), } } def status(self, model, spike, step_cache, switch_pct, cap_pct, tau): from .FLUXPATCH import _spike_controller _spike_controller.enabled = spike _spike_controller.cache_enabled = step_cache _spike_controller.switch_pct = switch_pct _spike_controller.event_cap_pct = cap_pct _spike_controller.tau = tau print( f"[ULT7RAFLUX] FP8 fused kernel | " f"spike={'ON' if spike else 'OFF'} | " f"cache={'ON' if step_cache else 'OFF'} | " f"switch={switch_pct*100:.0f}% cap={cap_pct*100:.0f}% tau={tau}" ) return (model,) NODE_CLASS_MAPPINGS = { "FLUXNATION": FLUXNATION, "FLUXATTENHUT": FLUXATTENHUT_Node, "FLUXSPIKEY": FLUXSPIKEY_Node, "ULT7RAFLUX": ULT7RAFLUX, } NODE_DISPLAY_NAME_MAPPINGS = { "FLUXNATION": "FLUXNATION", "FLUXATTENHUT": "FLUXATTENHUT", "FLUXSPIKEY": "FLUXSPIKEY", "ULT7RAFLUX": "ULT7RAFLUX", } _EXT_LOADED = False def _patch_safetensor_loading(): """Fix Windows mmap access violation in safetensors loading. PyTorch 2.5.x + safetensors safe_open() on Windows triggers access violations when memory-mapping large files (>10GB). This patches ComfyUI's load_torch_file to use safetensors.torch.load_file() which loads the full file without mmap. """ import sys if sys.platform != "win32": return try: import comfy.utils as comfy_utils import safetensors.torch import torch _original_load = comfy_utils.load_torch_file def _read_safetensor_streaming(path, target_device='cpu'): """Stream-load a safetensor file without mmap or full-file buffering. Reads the header to get tensor metadata, then reads each tensor individually from disk into a pre-allocated torch tensor. """ import struct, json, numpy as np dtype_map = { 'F16': (torch.float16, 2), 'BF16': (torch.bfloat16, 2), 'F32': (torch.float32, 4), 'F64': (torch.float64, 8), 'I8': (torch.int8, 1), 'I16': (torch.int16, 2), 'I32': (torch.int32, 4), 'I64': (torch.int64, 8), 'U8': (torch.uint8, 1), 'BOOL': (torch.bool, 1), 'F8_E4M3': (getattr(torch, 'float8_e4m3fn', torch.uint8), 1), 'F8_E5M2': (getattr(torch, 'float8_e5m2', torch.uint8), 1), } np_dtype_map = { 'F16': np.float16, 'BF16': np.void, 'F32': np.float32, 'F64': np.float64, 'I8': np.int8, 'I16': np.int16, 'I32': np.int32, 'I64': np.int64, 'U8': np.uint8, 'BOOL': np.bool_, } with open(path, 'rb') as f: header_size = struct.unpack('1GB) to avoid mmap crash if size_gb > 1.0: print(f"[FLUXNATION] Loading {os.path.basename(ckpt)} ({size_gb:.1f} GB) via streaming loader") try: import time t0 = time.time() sd, metadata = _read_safetensor_streaming(ckpt, device.type) elapsed = time.time() - t0 print(f"[FLUXNATION] Loaded {len(sd)} tensors in {elapsed:.1f}s (streaming, no-mmap)") if return_metadata: return sd, metadata return sd except Exception as e: print(f"[FLUXNATION] Streaming load failed ({e}), falling back to original") return _original_load(ckpt, safe_load=safe_load, device=device, return_metadata=return_metadata) else: return _original_load(ckpt, safe_load=safe_load, device=device, return_metadata=return_metadata) comfy_utils.load_torch_file = _safe_load_torch_file print("[FLUXNATION] Patched safetensor loading (non-mmap, Windows crash fix)") except Exception as e: print(f"[FLUXNATION] Could not patch safetensor loading: {e}") def _try_load_extension(): """Try to import the compiled flux_ext extension.""" global _EXT_LOADED try: import flux_ext # noqa: F401 — triggers TORCH_LIBRARY registration _EXT_LOADED = True print("[FLUXNATION] Loaded PyTorch C++ extension successfully") except ImportError: import torch import os root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) for candidate in [ os.path.join(root, "flux_ext.pyd"), os.path.join(root, "flux_ext.so"), ]: if os.path.exists(candidate): try: torch.ops.load_library(candidate) _EXT_LOADED = True print(f"[FLUXNATION] Loaded from {candidate}") break except Exception as e: print(f"[FLUXNATION] Failed to load {candidate}: {e}") if not _EXT_LOADED: print("[FLUXNATION] Extension not found. Run build_ext.bat to compile.") return False _register_fake_kernels() return True def _register_fake_kernels(): """Register Meta/FakeTensor implementations for torch.compile.""" import torch if not hasattr(torch, 'library') or not hasattr(torch.library, 'register_fake'): print("[FLUXNATION] torch.library.register_fake not available, skipping") return try: @torch.library.register_fake("flux_ext::ssb_prepare") def _fake_prepare(x, vec, pe, mod_w, mod_w_scale, mod_b, l1w, l1w_scale, l1b, qns, kns): B, S = x.size(0), x.size(1) q = torch.empty([B, 24, S, 128], dtype=torch.float16, device=x.device) k = torch.empty([B, 24, S, 128], dtype=torch.float16, device=x.device) v = torch.empty([B, 24, S, 128], dtype=torch.float16, device=x.device) mlp = torch.empty([B, S, 12288], dtype=torch.float16, device=x.device) gate = torch.empty([B, 3072], dtype=torch.float16, device=x.device) return (q, k, v, mlp, gate) @torch.library.register_fake("flux_ext::ssb_finish") def _fake_finish(x, attn, mlp, gate, l2w, l2w_scale, l2b): return torch.empty_like(x, dtype=torch.bfloat16) @torch.library.register_fake("flux_ext::ssb_cleanup") def _fake_cleanup(): return except Exception as e: print(f"[FLUXNATION] Warning: could not register fake kernels: {e}") # ── Initialization: patch safetensor loading + load extension ── import os as _os _patch_safetensor_loading() if _os.environ.get("FLUX_EXT_DISABLE", "0") == "1": print("[FLUXNATION] DISABLED by FLUX_EXT_DISABLE env var (stock mode)") elif _try_load_extension(): from .FLUXPATCH import apply_patch apply_patch() else: print("[FLUXNATION] DISABLED — extension not found. Run build_ext.bat.")