fabric/control/registry.py

"""In-memory registry for nodes and modules."""
from __future__ import annotations
import time
import logging

from fabric.shared.schemas import (
    Heartbeat, NodeInfo, ModuleStatus, ModuleState,
    PlaceRequest, PlaceResponse, PriorityClass,
)

log = logging.getLogger("fabric.registry")

STALE_NODE_TIMEOUT_S = 60


class Registry:
    def __init__(self):
        self.nodes: dict[str, NodeInfo] = {}
        self._version = 0

    @property
    def version(self) -> int:
        return self._version

    def handle_heartbeat(self, hb: Heartbeat) -> None:
        node = self.nodes.get(hb.node_id)
        if node is None:
            log.info("New node registered: %s (%s)", hb.node_id, hb.node_ip)
        self.nodes[hb.node_id] = NodeInfo(
            node_id=hb.node_id,
            node_ip=hb.node_ip,
            agent_port=hb.agent_port,
            hypervisor=hb.hypervisor,
            gpu_status=hb.gpu_status,
            modules=hb.modules,
            last_heartbeat=time.time(),
            alive=True,
            gpus=hb.gpus,
            vram_total_mb=hb.vram_total_mb,
            vram_free_mb=hb.vram_free_mb,
            driver_version=hb.driver_version,
        )
        self._version += 1

    def prune_stale_nodes(self) -> list[str]:
        now = time.time()
        stale = []
        for nid, node in list(self.nodes.items()):
            if now - node.last_heartbeat > STALE_NODE_TIMEOUT_S:
                node.alive = False
                stale.append(nid)
        return stale

    def get_all_modules(self) -> list[ModuleStatus]:
        modules = []
        for node in self.nodes.values():
            if not node.alive:
                continue
            for m in node.modules:
                modules.append(m)
        return modules

    def get_alive_nodes(self) -> list[NodeInfo]:
        return [n for n in self.nodes.values() if n.alive]

    def find_module(self, model_name: str) -> list[tuple[NodeInfo, ModuleStatus]]:
        results = []
        for node in self.nodes.values():
            if not node.alive:
                continue
            for m in node.modules:
                if m.name == model_name or model_name in m.aliases:
                    results.append((node, m))
        return results

    def place(self, req: PlaceRequest) -> PlaceResponse | None:
        candidates = self.find_module(req.model)
        if not candidates:
            return None

        state_priority = {
            ModuleState.running: 0,
            ModuleState.loaded: 1,
            ModuleState.idle: 2,
            ModuleState.sleeping: 3,
            ModuleState.warming: 4,
            ModuleState.unloaded: 5,
            ModuleState.error: 6,
        }
        # Capacity-aware tie-break: among equally-good states, prefer the node
        # with the most free physical VRAM (more headroom => safer to (re)load).
        candidates.sort(key=lambda c: (state_priority.get(c[1].state, 99), -(c[0].vram_free_mb or 0)))

        node, module = candidates[0]
        return PlaceResponse(
            node_id=node.node_id,
            node_ip=node.node_ip,
            infer_url=module.infer_url,
            module_name=module.name,
            state=module.state,
        )

    def total_gpu_vram_mb(self) -> int:
        total = 0
        for node in self.nodes.values():
            if not node.alive:
                continue
            for g in node.gpu_status:
                total += g.vram_mb
        return total

    def total_gpu_capacity_mb(self) -> int:
        """Sum of physical-GPU total VRAM across alive nodes (hardware capacity)."""
        return sum(n.vram_total_mb for n in self.nodes.values() if n.alive)

    def total_gpu_free_mb(self) -> int:
        """Sum of physical-GPU free VRAM across alive nodes (placement headroom)."""
        return sum(n.vram_free_mb for n in self.nodes.values() if n.alive)