"""Pre-inference memory budgeting + OOM-avoidance planning for Apple Silicon (MPS). Implements the qwen-brain ruling for the never-OOM mandate: * **Dynamic budget** keyed off LIVE available unified memory (recomputed every preflight), NOT a fixed fraction of the 128 GB total — the box is shared, so we must react to what other sessions are actually using right now. * **Two gates**: a LOAD gate (can the ~58 GB resident weights fit *before* we build the pipeline?) and an INFER gate (does resident + this request's activation peak fit?), evaluated per request. * **Soft auto-degrade with a hard floor**: walk a degrade ladder (resolution → refs → drop CFG double-pass → Quality→Fast); if even the bottom rung won't fit, refuse with an actionable message rather than OOM/​swap-thrash. On unified memory, ``enable_model_cpu_offload`` saves no physical RAM (CPU+GPU share one pool), so it is deliberately NOT a degrade rung here. All heavy imports (torch, psutil) are deferred so this module imports GPU-free in CI. """ from __future__ import annotations import os GB = 1024**3 # Resident bf16 weights — measured from the HF repo metadata: # transformer 40.86 + text_encoder (Qwen2.5-VL-7B) 16.58 + vae 0.25 = 57.7 GB, # plus ~0.25 GB when the VAE is upcast to fp32 on MPS (anti black-image). RESIDENT_GB = 58.0 LORA_GB = 1.0 # Lightning adapter is always loaded (toggled, not unloaded) LOAD_TRANSIENT_GB = 5.0 # shard materialization transient during from_pretrained DEFAULT_RESERVE_GB = 20.0 # OS + other sessions + macOS 'available' is an estimate (MPS, shared RAM) DEFAULT_CUDA_RESERVE_GB = 3.0 # CUDA context + allocator fragmentation (dedicated VRAM, not shared) # --- Activation model (recalibrated from MEASURED MPS peaks, fresh-process) ------------- # Measured driver_allocated peaks @ 1024^2 output: edit/Fast n_img=1 -> 28.1, edit/Quality # n_img=1 -> 28.1, compose/Fast n_img=2 -> 40.8, compose/Quality n_img=2 -> 42.7 GB. # Findings: ~14 GB per INPUT image dominates (the pipeline auto-resizes inputs to fixed # areas — CONDITION 384^2 + VAE 1024^2 — so this per-image encode cost is ~CONSTANT in the # OUTPUT resolution); CFG double-pass adds only ~0-2 GB; no super-linear (n=1->2 is linear). # Coefficients are an UPPER ENVELOPE over the measured points (conservative -> never-OOM). # _ACT_BASE carries a fragmentation margin: live long-lived peaks ran ~2 GB above the # fresh-process fit (compose/Fast measured 40.8 fresh but 45.6 in the live app), so the # base is raised from the bare fresh-fit (~15) to cover the fragmented regime on the first # run; calibration then ratchets up further for any config that exceeds it. _ACT_BASE = 18.0 # fixed overhead (Metal/kernel caches, fragmentation, base buffers) _ACT_PER_IMG = 13.5 # GB per input image (VL 384^2 + VAE 1MP encode + input latent tokens) _ACT_DENOISE = 0.5 # GB per (n_img+1)*MP_out*cfg_fac (output denoise + VAE decode; small) # CUDA coefficients are ~8x LOWER than MPS: on CUDA the model uses FlashAttention/efficient # SDPA, which does NOT materialize the large attention tensors that MPS does. MEASURED on # ZeroGPU (RTX Pro 6000): edit/Fast n_img=1 @1024^2 peak ABOVE the resident baseline = 4.2 GB # (vs the MPS formula's ~32 GB). These are conservative (~3x the measurement) so the first # run of every Edit/Compose/Quality config fits the ~37 GB free without a FALSE degrade; # the device-keyed calibration (trust-calib on CUDA) then refines each config to its real # ~4-10 GB, and the catchable-CUDA-OOM retry remains the hard never-OOM backstop. _ACT_BASE_CUDA = 8.0 _ACT_PER_IMG_CUDA = 6.0 _ACT_DENOISE_CUDA = 0.6 # Resolution rungs (max_pixels) for the degrade ladder — the dominant lever. RES_RUNGS = (1024 * 1024, 896 * 896, 768 * 768, 640 * 640) # In-process calibration cache: (mode, res_bucket, n_ref, quality) -> measured peak GB. # Populated by record_peak() after each successful run; used in place of the # conservative formula thereafter (x safety margin already baked into the record). _CALIB: dict[tuple, float] = {} _CALIB_MARGIN = 1.1 # measured peak is already the high-water mark; small slack for proxy miss def _reserve_gb() -> float: try: return float(os.environ.get("QIE_MPS_RESERVE_GB", DEFAULT_RESERVE_GB)) except (TypeError, ValueError): return DEFAULT_RESERVE_GB def _cuda_reserve_gb() -> float: try: return float(os.environ.get("QIE_CUDA_RESERVE_GB", DEFAULT_CUDA_RESERVE_GB)) except (TypeError, ValueError): return DEFAULT_CUDA_RESERVE_GB def available_gb() -> float: """Live available system memory (GB). macOS reports a reclaimable-inclusive estimate.""" import psutil return psutil.virtual_memory().available / GB def budget_gb(device: str) -> float: """Dynamic memory budget (GB). Recompute every preflight — cheap. BUDGET = min(0.90 * mps.recommended_max_memory, available - reserve) The physical (available - reserve) term binds on a loaded box; the recommended_max cap is a backstop for when `available` is transiently high. """ phys = available_gb() - _reserve_gb() if device == "mps": import torch cap = 0.90 * torch.mps.recommended_max_memory() / GB return min(cap, phys) return phys # CPU best-effort; CUDA has its own VRAM logic in models.should_cpu_offload def _res_bucket(height: int, width: int) -> int: px = height * width for mp in RES_RUNGS: if px >= mp - 1: return mp return RES_RUNGS[-1] def fit_dims(base_w: int, base_h: int, max_pixels: int, multiple: int = 16) -> tuple[int, int]: """Same contract as models.fit_dimensions, parameterised by max_pixels (no PIL dep).""" w, h = base_w, base_h if w * h > max_pixels: scale = (max_pixels / (w * h)) ** 0.5 w = int(w * scale) h = int(h * scale) w = max(256, (w // multiple) * multiple) h = max(256, (h // multiple) * multiple) return w, h def activation_gb(height: int, width: int, n_ref: int, cfg_on: bool, device: str = "mps") -> float: """Conservative (over-estimating) peak activation memory (GB) for one inference. Device-specific coefficients: MPS uses the big upper-envelope (it materializes attention tensors); CUDA uses ~8x lower coefficients (FlashAttention; measured ~4 GB for edit/Fast 1-img). ``n_img`` = 1 + n_ref input images. The per-image encode term is ~fixed in output resolution (the pipeline auto-resizes inputs to fixed areas); only the denoise/decode term scales with output MP. ``cfg_on`` (true_cfg>1) doubles ONLY the denoise term — the VL/VAE conditioning is computed once and reused for the cond+uncond passes. """ base, per_img, denoise = ( (_ACT_BASE_CUDA, _ACT_PER_IMG_CUDA, _ACT_DENOISE_CUDA) if device == "cuda" else (_ACT_BASE, _ACT_PER_IMG, _ACT_DENOISE) ) n_img = 1 + n_ref mp_out = (height * width) / (1024 * 1024) cfg_fac = 2 if cfg_on else 1 return base + per_img * n_img + denoise * (n_img + 1) * mp_out * cfg_fac def _act_estimate( device: str, height: int, width: int, n_ref: int, cfg_on: bool, has_user_lora: bool = False ) -> float: """Activation-only estimate (GB), device-aware. The formula is already re-fit to measured reality (+margin), so a calibrated value BELOW it carries no information, only risk: a low cross-request driver-delta (e.g. the 15.5 GB artifact when a prior run left the MPS heap warm and empty_cache didn't fully release it) must NEVER be allowed to drop the estimate below the real peak, or the preflight under-budgets and OOMs on a loaded box. So calibration (and penalize) can only RAISE the estimate above the formula, never lower it. record_peak still ratchets up for genuinely-higher fragmented peaks. """ key = (device, "act", _res_bucket(height, width), n_ref, cfg_on, has_user_lora) formula = activation_gb(height, width, n_ref, cfg_on, device) calib = _CALIB.get(key) if calib is None: return formula if device == "cuda": # CUDA has an ACCURATE peak counter (reset_peak_memory_stats + max_memory_allocated), # so a recorded peak REPLACES the (MPS-derived, inflated) formula — letting Compose # relax to CUDA reality (~25-30 GB) instead of being pinned at ~46 GB and forever # over-degrading on a 96 GB GPU. record_peak floors it at _ACT_BASE. return calib # MPS: UPWARD-ONLY ratchet — the driver-delta is an unreliable sticky-heap proxy, so # calibration can only RAISE the conservative formula, never lower it below the real peak. return max(formula, calib) def activation_budget_gb(device: str) -> float: """Free memory available for THIS inference's activations (GB). KEY CORRECTNESS POINT: at inference time the ~58 GB weights are ALREADY resident, so psutil.available has already dropped to exclude them. The activation therefore only needs to fit in the *remaining* free RAM (minus reserve), and within the MPS working-set headroom above what is already allocated. Comparing resident+activation against a post-load `available` would double-count the now-resident weights and spuriously refuse every request (observed in the first smoke test). """ if device == "cuda": import torch # Activation budget on CUDA = TOTAL VRAM - currently-ALLOCATED tensors - reserve. # NOT mem_get_info()'s driver-"free": after loading the ~58 GB model the PyTorch # caching allocator RESERVES most of VRAM, and the driver counts reserved-but-unused # as "used" -> free reads ~0 -> we'd spuriously refuse every request (observed: # "free 0 GB"). The activation reuses that reserved pool, so (total - allocated) is # the real headroom. e.g. xlarge 96 GB - 58 allocated - 3 reserve = ~35 GB for acts. _free_b, total_b = torch.cuda.mem_get_info() allocated_b = torch.cuda.memory_allocated() return max(0.0, (total_b - allocated_b) / GB - _cuda_reserve_gb()) free = available_gb() - _reserve_gb() if device == "mps": import torch # Budget cap = 0.95 * recommended_max, intentionally BELOW the 1.0 allocator HIGH # watermark (app.py): that ~5 GB gap absorbs allocator fragmentation so a # legitimately-approved run doesn't spuriously trip the watermark. The heaviest # real mode (compose/Quality, ~42.7 GB act -> ~97.7 GB total) fits under 0.95. # Do NOT raise this to 1.0 — it would erase the fragmentation gap. cap = 0.95 * torch.mps.recommended_max_memory() / GB used = torch.mps.driver_allocated_memory() / GB return max(0.0, min(free, cap - used)) return max(0.0, free) def record_peak( device: str, mode: str, height: int, width: int, n_ref: int, cfg_on: bool, measured_peak_gb: float, has_user_lora: bool = False, ) -> None: """Record a measured activation peak for calibration (FIX #2: floor-guarded). Rejects an implausibly-small delta (driver heap released before the read, or a proxy miss) so the cache can never be poisoned toward ~0 — which would silently disable activation budgeting and cause an OOM. Keeps the max (x margin) across runs, never below the physical floor. Cache is device-keyed (MPS driver-delta vs CUDA accurate peak-counter measurements must never cross-pollute). """ floor = _ACT_BASE # any real inference allocates at least the base overhead if measured_peak_gb < floor: return key = (device, "act", _res_bucket(height, width), n_ref, cfg_on, has_user_lora) _CALIB[key] = max(_CALIB.get(key, 0.0), measured_peak_gb * _CALIB_MARGIN) def load_gate(device: str) -> tuple[bool, float, float]: """Can resident weights fit BEFORE building? Returns (ok, need_gb, budget_gb). Resolution can't fix a resident-weights overflow, so a failure here means refuse. """ need = RESIDENT_GB + LORA_GB + LOAD_TRANSIENT_GB budget = budget_gb(device) return need <= budget, need, budget def top_consumers(n: int = 5) -> str: """Human-readable list of the biggest memory-using processes (for refusal messages).""" try: import psutil procs = [] for p in psutil.process_iter(["name", "memory_info"]): try: procs.append((p.info["memory_info"].rss, p.info["name"])) except Exception: continue procs.sort(reverse=True) return ", ".join(f"{name}={rss / GB:.1f}GB" for rss, name in procs[:n]) except Exception: return "(unavailable)" def plan_request( device: str, mode: str, base_w: int, base_h: int, n_ref: int, speed: str, steps: int, true_cfg: float, has_user_lora: bool = False, ) -> dict: """Walk the degrade ladder; return the least-degraded plan that fits BUDGET. Returns a dict: {width,height,steps,true_cfg,n_ref,speed,refused,note,degrades, need_gb,budget_gb}. ``refused=True`` means even the bottom rung overflows. Degrade order (revised from the measured data): the per-input-image cost (~13.5 GB) is FIXED in output resolution (the pipeline auto-resizes inputs), so REF-DROP is the only strong lever and comes FIRST; CFG-drop and RESOLUTION are weak (~0-2 GB) and come later. Steps are NOT a memory lever (flat in memory). """ act_budget = activation_budget_gb(device) quality = speed == "Quality" is_compose = mode == "compose" full_w, full_h = fit_dims(base_w, base_h, RES_RUNGS[0]) min_ref = 0 if is_compose else n_ref # edit can't shed inputs def candidates(): # (mp, n_ref, true_cfg, speed) — ordered least-degraded -> most. yield RES_RUNGS[0], n_ref, true_cfg, speed # full, no degrade if is_compose: # drop reference images — the dominant lever, at FULL resolution for r in range(n_ref - 1, -1, -1): yield RES_RUNGS[0], r, true_cfg, speed if quality: # drop CFG double-pass (weak) yield RES_RUNGS[0], min_ref, 1.0, speed for mp in RES_RUNGS[1:]: # resolution down (weak lever) — last yield mp, min_ref, (1.0 if quality else true_cfg), speed if quality: # force Quality->Fast (Lightning 4-step) yield RES_RUNGS[-1], min_ref, 1.0, "Fast" for mp, nref_c, cfg_c, speed_c in candidates(): w, h = fit_dims(base_w, base_h, mp) act = _act_estimate(device, w, h, nref_c, cfg_c > 1.0, has_user_lora) if act <= act_budget: degrades = [] if nref_c < n_ref: degrades.append(f"inputs {n_ref + 1}->{nref_c + 1}") if cfg_c <= 1.0 < true_cfg: degrades.append("cfg double-pass off") if (w, h) != (full_w, full_h): degrades.append(f"resolution {full_w}x{full_h}->{w}x{h}") if speed_c != speed: degrades.append(f"{speed}->{speed_c}") steps_c = 4 if speed_c == "Fast" and speed == "Quality" else steps footprint = RESIDENT_GB + LORA_GB + act note = _note(act, act_budget, footprint, w, h, nref_c, cfg_c > 1.0, degrades) return dict( width=w, height=h, steps=steps_c, true_cfg=cfg_c, n_ref=nref_c, speed=speed_c, refused=False, note=note, degrades=degrades, act_gb=round(act, 1), act_budget_gb=round(act_budget, 1), need_gb=round(footprint, 1), budget_gb=round(RESIDENT_GB + LORA_GB + act_budget, 1), ) # Hard floor: even the bottom rung's activation overflows free memory. w, h = fit_dims(base_w, base_h, RES_RUNGS[-1]) act = _act_estimate(device, w, h, min_ref, False, has_user_lora) footprint = RESIDENT_GB + LORA_GB + act note = ( f"OOM-REFUSED: even {w}x{h} Fast needs ~{act:.0f} GB activation > free {act_budget:.0f} GB " f"(available {available_gb():.0f} - reserve {_reserve_gb():.0f}; weights {RESIDENT_GB:.0f} GB resident). " f"Free ~{act - act_budget:.0f} GB. Top: {top_consumers()}" ) return dict( width=w, height=h, steps=4, true_cfg=1.0, n_ref=min_ref, speed="Fast", refused=True, note=note, degrades=["REFUSED"], act_gb=round(act, 1), act_budget_gb=round(act_budget, 1), need_gb=round(footprint, 1), budget_gb=round(RESIDENT_GB + LORA_GB + act_budget, 1), ) def _note(act, act_budget, footprint, w, h, n_ref, quality, degrades) -> str: base = ( f"OOM preflight: activation ~{act:.0f} GB <= free {act_budget:.0f} GB " f"(weights {RESIDENT_GB + LORA_GB:.0f} GB resident; peak footprint ~{footprint:.0f} GB) " f"@ {w}x{h}{' Q' if quality else ' F'}, {n_ref + 1} image(s)." ) if degrades: return base + " Degraded: " + "; ".join(degrades) + "." return base + " No degrade." def penalize( device: str, height: int, width: int, n_ref: int, cfg_on: bool, failed_budget_gb: float, has_user_lora: bool = False, ) -> None: """Self-correcting calibration after an ACTUAL OOM (qwen-brain Q2.2). Bumps this config's cached estimate above the budget it just overflowed, so the next identical request degrades preemptively in the preflight instead of repeating the OOM + retry loop. Device-keyed (per record_peak). """ key = (device, "act", _res_bucket(height, width), n_ref, cfg_on, has_user_lora) _CALIB[key] = max(_CALIB.get(key, 0.0), failed_budget_gb * 1.2) def step_down(width, height, n_ref, speed, true_cfg, steps, base_w, base_h, mode): """Next-more-aggressive config after an ACTUAL MPS OOM (reactive degrade). Mirrors the preflight ladder (REF-DROP first — the only strong lever; CFG then resolution are weak) so a mis-estimate converges to a fitting config instead of crashing. Each step strictly reduces memory and PRESERVES the user's step count (steps aren't a memory lever); only the Quality->Fast switch changes steps to the Lightning 4-step count. Returns a config dict, or None at the floor (-> clean refuse). This is the hard never-OOM guarantee. """ # 1. (compose) drop a reference image — dominant lever, keep current resolution. if mode == "compose" and n_ref > 0: return dict(width=width, height=height, n_ref=n_ref - 1, speed=speed, true_cfg=true_cfg, steps=steps) # 2. (quality) drop CFG double-pass (weak). if true_cfg > 1.0: return dict(width=width, height=height, n_ref=n_ref, speed=speed, true_cfg=1.0, steps=steps) # 3. resolution down (weak lever) — last. cur = width * height idx = next((i for i, mp in enumerate(RES_RUNGS) if cur >= mp - 1), len(RES_RUNGS) - 1) if idx < len(RES_RUNGS) - 1: w, h = fit_dims(base_w, base_h, RES_RUNGS[idx + 1]) if (w, h) != (width, height): return dict(width=w, height=h, n_ref=n_ref, speed=speed, true_cfg=true_cfg, steps=steps) # 4. force Quality->Fast (Lightning 4-step). if speed != "Fast": return dict(width=width, height=height, n_ref=n_ref, speed="Fast", true_cfg=1.0, steps=4) return None