fix(rocm/win): stub class metaclass for ProcessGroup.BackendType; amd-smi circuit breaker

torchao.float8.inference accesses ProcessGroup.BackendType as a class-level
attribute.  Plain type() stubs have no __getattr__ on the metaclass so this
raises AttributeError.  Introduce _StubClassMeta whose __getattr__ returns
child stub classes, fixing the torchao import chain.

Add an amd-smi circuit breaker in amd.py: after 3 consecutive failures the
module stops spawning the process, eliminating the repeated Windows UAC /
DiskPart elevation prompts caused by polling a non-functional amd-smi.

Also guard BNB_ROCM_VERSION=72 behind a DLL existence check so bitsandbytes
fails with its own detection message rather than a harder "DLL not found" when
the Windows ROCm bnb wheel is not yet installed.
This commit is contained in:
LeoBorcherding 2026-05-11 05:22:03 -05:00
commit 4d09cbbda4
2 changed files with 44 additions and 4 deletions

View file

@ -1111,6 +1111,20 @@ def run_training_process(
m.__getattr__ = _ga
return m
# Metaclass for stub *classes* so class-level attribute access works too.
# e.g. torchao does ProcessGroup.BackendType — plain type() has no __getattr__
# on the metaclass, so we need this to avoid AttributeError on class attrs.
class _StubClassMeta(type):
def __getattr__(cls, attr):
if attr.startswith("__"):
raise AttributeError(attr)
child = _StubClassMeta(attr, (), {"__init__": lambda self, *a, **kw: None})
setattr(cls, attr, child)
return child
def _make_stub_class(name):
return _StubClassMeta(name, (), {"__init__": lambda self, *a, **kw: None})
if sys.platform == "win32":
_c10d_key = "torch._C._distributed_c10d"
if _c10d_key not in sys.modules: # guard: never overwrite real NVIDIA impl
@ -1121,7 +1135,7 @@ def run_training_process(
def _c10d_stub_getattr(_attr):
if _attr.startswith("__"):
raise AttributeError(_attr)
_cls = type(_attr, (), {"__init__": lambda self, *a, **kw: None})
_cls = _make_stub_class(_attr)
setattr(_c10d_stub, _attr, _cls)
return _cls
@ -1165,7 +1179,7 @@ def run_training_process(
_mod = sys.modules[_full]
setattr(_td, _attr, _mod)
return _mod
_cls = type(_attr, (), {"__init__": lambda self, *a, **kw: None})
_cls = _make_stub_class(_attr)
setattr(_td, _attr, _cls)
return _cls
_td.__getattr__ = _td_getattr
@ -1184,9 +1198,16 @@ def run_training_process(
# ── 1e. Point bitsandbytes at the ROCm 7.2 DLL on Windows ──
# The AMD continuous-release wheel ships libbitsandbytes_rocm72.dll.
# BNB_ROCM_VERSION overrides the version string bnb uses to locate the DLL.
# Only set BNB_ROCM_VERSION when that DLL is actually present — setting it
# when the DLL is absent makes bnb fail harder than the default detection.
if sys.platform == "win32" and os.environ.get("UNSLOTH_ROCM_TORCH_INSTALLED") == "1":
os.environ.setdefault("BNB_ROCM_VERSION", "72")
import importlib.util as _ilu
_bnb_spec = _ilu.find_spec("bitsandbytes")
if _bnb_spec and _bnb_spec.origin:
import pathlib as _pl
_bnb_dll = _pl.Path(_bnb_spec.origin).parent / "libbitsandbytes_rocm72.dll"
if _bnb_dll.exists():
os.environ.setdefault("BNB_ROCM_VERSION", "72")
# ── 2. Now import ML libraries (fresh in this clean process) ──
try:

View file

@ -27,9 +27,19 @@ logger = get_logger(__name__)
# can take 15-25 s on cold hardware. Linux is consistently < 2 s.
_AMD_SMI_DEFAULT_TIMEOUT = 30 if platform.system() == "Windows" else 10
# Circuit breaker: stop calling amd-smi after this many consecutive failures.
# On Windows, each failed call spawns a process that may show a UAC/DiskPart
# elevation prompt. Once we know amd-smi doesn't work we stop polling it.
_AMD_SMI_FAILURE_LIMIT = 3
_amd_smi_consecutive_failures = 0
_amd_smi_disabled = False
def _run_amd_smi(*args: str, timeout: int = _AMD_SMI_DEFAULT_TIMEOUT) -> Optional[Any]:
"""Run amd-smi with the given arguments and return parsed JSON, or None."""
global _amd_smi_consecutive_failures, _amd_smi_disabled
if _amd_smi_disabled:
return None
try:
result = subprocess.run(
["amd-smi", *args, "--json"],
@ -41,10 +51,19 @@ def _run_amd_smi(*args: str, timeout: int = _AMD_SMI_DEFAULT_TIMEOUT) -> Optiona
)
except (OSError, subprocess.TimeoutExpired) as e:
logger.warning("amd-smi query failed: %s", e)
_amd_smi_consecutive_failures += 1
if _amd_smi_consecutive_failures >= _AMD_SMI_FAILURE_LIMIT:
logger.warning("amd-smi unavailable -- disabling GPU polling to avoid repeated prompts")
_amd_smi_disabled = True
return None
if result.returncode != 0 or not result.stdout.strip():
logger.warning("amd-smi returned code %d", result.returncode)
_amd_smi_consecutive_failures += 1
if _amd_smi_consecutive_failures >= _AMD_SMI_FAILURE_LIMIT:
logger.warning("amd-smi unavailable -- disabling GPU polling to avoid repeated prompts")
_amd_smi_disabled = True
return None
_amd_smi_consecutive_failures = 0 # reset on success
try:
return json.loads(result.stdout)
except json.JSONDecodeError: