music3-mac16 / runtime /memory_manager.py

Upload project files

2fb10fd verified 2 months ago

5.43 kB

"""runtime/memory_manager.py — Milestone 4: Memory Manager (PM v0.2 §5)。
职责: 基于 mac16.yaml 的 memory_budget, 动态决定 load/unload/evict。
当前阶段驻留管线的决策模型是确定性的 (AR → unload → Flow), 本模块把
"决策" 从管线代码中抽出来, 并提供真实内存遥测的校验闭环:
  1. 读取 profile 的 memory_budget_breakdown
  2. 决策器: 根据 active/cache/峰值与预算判断某阶段是否可以驻留
  3. 阶段会话: enter → materialize, exit → 释放并校验回落
  4. 测量: 每阶段峰值/耗时/缓存归零检查, 输出 memory report
后续 (M4 扩展) 可加:
  - 动态 eviction: 当 Flow 阶段 chunk 峰值超预算时, 降低 chunk 帧数
  - 双预算档位: preview (更激进) vs create (更保守)
"""
from __future__ import annotations

import gc
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Optional

import mlx.core as mx
import yaml

from runtime.profiler import gb


@dataclass
class StageDecision:
    """一次 load/unload 决策的记录。"""

    stage: str
    action: str  # load | unload | skip
    weight_gb: float
    active_before_gb: float
    active_after_gb: float
    peak_gb: float
    budget_ok: bool
    cache_cleared: bool
    note: str = ""


@dataclass
class MemoryReport:
    stages: list[StageDecision] = field(default_factory=list)
    max_peak_gb: float = 0.0
    profile_name: str = "?"

    def add(self, d: StageDecision) -> None:
        self.stages.append(d)
        self.max_peak_gb = max(self.max_peak_gb, d.peak_gb)

    def to_dict(self) -> dict:
        max_peak_gb = max(
            (stage.peak_gb for stage in self.stages), default=self.max_peak_gb
        )
        return {
            "profile": self.profile_name,
            "max_peak_gb": round(max_peak_gb, 3),
            "stages": [
                {
                    "stage": s.stage,
                    "action": s.action,
                    "weight_gb": round(s.weight_gb, 3),
                    "active_before_gb": round(s.active_before_gb, 3),
                    "active_after_gb": round(s.active_after_gb, 3),
                    "peak_gb": round(s.peak_gb, 3),
                    "budget_ok": s.budget_ok,
                    "cache_cleared": s.cache_cleared,
                }
                for s in self.stages
            ],
        }


class MemoryManager:
    """基于 profile 预算的确定性阶段决策器 + 遥测。"""

    def __init__(
        self,
        profile_path: Optional[Path] = None,
        budget_gb: Optional[float] = None,
    ) -> None:
        self.profile = self._load_profile(profile_path)
        self.budget_gb = budget_gb or float(self.profile.get("memory_budget_gb", 11.0))

    # -------------------------------------------------- 公共 API
    def can_resident(self, weight_gb: float, current_active_gb: float) -> bool:
        """决策: 当前 active + 待加载权重是否在预算内 (留 10% 余量)。"""
        return current_active_gb + weight_gb <= self.budget_gb * 0.9

    def stage_enter(self, stage: str, weight_gb: float, loader: Callable):
        """进入阶段: 检查预算 → 执行加载 → 记录峰值。返回加载的模型。"""
        before = gb(mx.get_active_memory())
        ok = self.can_resident(weight_gb, before)
        if not ok:
            mx.clear_cache()
            before = gb(mx.get_active_memory())
            ok = self.can_resident(weight_gb, before)
        if not ok:
            raise MemoryError(
                f"阶段 {stage} 需要 {weight_gb:.2f}GB，但当前预算 {self.budget_gb:.2f}GB 不足"
            )
        mx.reset_peak_memory()
        obj = loader()
        mx.synchronize()
        peak = gb(mx.get_peak_memory())
        after = gb(mx.get_active_memory())
        return obj, StageDecision(
            stage=stage,
            action="load" if ok else "load_after_evict",
            weight_gb=weight_gb,
            active_before_gb=before,
            active_after_gb=after,
            peak_gb=peak,
            budget_ok=ok,
            cache_cleared=not ok,
        )

    def stage_exit(
        self,
        decision: StageDecision,
        release: Callable[[], None] | None = None,
    ) -> None:
        """退出阶段: 先清掉调用方持有的模型引用, 再执行释放四步曲。"""
        if release is not None:
            release()
        gc.collect()
        mx.synchronize()
        mx.clear_cache()
        cache = gb(mx.get_cache_memory())
        decision.active_after_gb = gb(mx.get_active_memory())
        decision.cache_cleared = cache == 0.0

    def report(self) -> MemoryReport:
        r = MemoryReport(profile_name=self.profile.get("device", "?"))
        return r

    # -------------------------------------------------- helpers
    @staticmethod
    def _load_profile(profile_path: Optional[Path]) -> dict:
        if profile_path is None:
            default = Path(__file__).resolve().parent.parent / "profiles" / "mac16.yaml"
            profile_path = default
        if not profile_path.exists():
            return {"memory_budget_gb": 11.0, "device": "unknown"}
        with open(profile_path) as f:
            return yaml.safe_load(f) or {}


def current_usage_gb() -> float:
    return gb(mx.get_active_memory())


def current_peak_gb() -> float:
    return gb(mx.get_peak_memory())

	"""runtime/memory_manager.py — Milestone 4: Memory Manager (PM v0.2 §5)。

	职责: 基于 mac16.yaml 的 memory_budget, 动态决定 load/unload/evict。
	当前阶段驻留管线的决策模型是确定性的 (AR → unload → Flow), 本模块把
	"决策" 从管线代码中抽出来, 并提供真实内存遥测的校验闭环:
	1. 读取 profile 的 memory_budget_breakdown
	2. 决策器: 根据 active/cache/峰值与预算判断某阶段是否可以驻留
	3. 阶段会话: enter → materialize, exit → 释放并校验回落
	4. 测量: 每阶段峰值/耗时/缓存归零检查, 输出 memory report

	后续 (M4 扩展) 可加:
	- 动态 eviction: 当 Flow 阶段 chunk 峰值超预算时, 降低 chunk 帧数
	- 双预算档位: preview (更激进) vs create (更保守)
	"""
	from __future__ import annotations

	import gc
	import time
	from dataclasses import dataclass, field
	from pathlib import Path
	from typing import Callable, Optional

	import mlx.core as mx
	import yaml

	from runtime.profiler import gb


	@dataclass
	class StageDecision:
	"""一次 load/unload 决策的记录。"""

	stage: str
	action: str # load \| unload \| skip
	weight_gb: float
	active_before_gb: float
	active_after_gb: float
	peak_gb: float
	budget_ok: bool
	cache_cleared: bool
	note: str = ""


	@dataclass
	class MemoryReport:
	stages: list[StageDecision] = field(default_factory=list)
	max_peak_gb: float = 0.0
	profile_name: str = "?"

	def add(self, d: StageDecision) -> None:
	self.stages.append(d)
	self.max_peak_gb = max(self.max_peak_gb, d.peak_gb)

	def to_dict(self) -> dict:
	max_peak_gb = max(
	(stage.peak_gb for stage in self.stages), default=self.max_peak_gb
	)
	return {
	"profile": self.profile_name,
	"max_peak_gb": round(max_peak_gb, 3),
	"stages": [
	{
	"stage": s.stage,
	"action": s.action,
	"weight_gb": round(s.weight_gb, 3),
	"active_before_gb": round(s.active_before_gb, 3),
	"active_after_gb": round(s.active_after_gb, 3),
	"peak_gb": round(s.peak_gb, 3),
	"budget_ok": s.budget_ok,
	"cache_cleared": s.cache_cleared,
	}
	for s in self.stages
	],
	}


	class MemoryManager:
	"""基于 profile 预算的确定性阶段决策器 + 遥测。"""

	def __init__(
	self,
	profile_path: Optional[Path] = None,
	budget_gb: Optional[float] = None,
	) -> None:
	self.profile = self._load_profile(profile_path)
	self.budget_gb = budget_gb or float(self.profile.get("memory_budget_gb", 11.0))

	# -------------------------------------------------- 公共 API
	def can_resident(self, weight_gb: float, current_active_gb: float) -> bool:
	"""决策: 当前 active + 待加载权重是否在预算内 (留 10% 余量)。"""
	return current_active_gb + weight_gb <= self.budget_gb * 0.9

	def stage_enter(self, stage: str, weight_gb: float, loader: Callable):
	"""进入阶段: 检查预算 → 执行加载 → 记录峰值。返回加载的模型。"""
	before = gb(mx.get_active_memory())
	ok = self.can_resident(weight_gb, before)
	if not ok:
	mx.clear_cache()
	before = gb(mx.get_active_memory())
	ok = self.can_resident(weight_gb, before)
	if not ok:
	raise MemoryError(
	f"阶段 {stage} 需要 {weight_gb:.2f}GB，但当前预算 {self.budget_gb:.2f}GB 不足"
	)
	mx.reset_peak_memory()
	obj = loader()
	mx.synchronize()
	peak = gb(mx.get_peak_memory())
	after = gb(mx.get_active_memory())
	return obj, StageDecision(
	stage=stage,
	action="load" if ok else "load_after_evict",
	weight_gb=weight_gb,
	active_before_gb=before,
	active_after_gb=after,
	peak_gb=peak,
	budget_ok=ok,
	cache_cleared=not ok,
	)

	def stage_exit(
	self,
	decision: StageDecision,
	release: Callable[[], None] \| None = None,
	) -> None:
	"""退出阶段: 先清掉调用方持有的模型引用, 再执行释放四步曲。"""
	if release is not None:
	release()
	gc.collect()
	mx.synchronize()
	mx.clear_cache()
	cache = gb(mx.get_cache_memory())
	decision.active_after_gb = gb(mx.get_active_memory())
	decision.cache_cleared = cache == 0.0

	def report(self) -> MemoryReport:
	r = MemoryReport(profile_name=self.profile.get("device", "?"))
	return r

	# -------------------------------------------------- helpers
	@staticmethod
	def _load_profile(profile_path: Optional[Path]) -> dict:
	if profile_path is None:
	default = Path(__file__).resolve().parent.parent / "profiles" / "mac16.yaml"
	profile_path = default
	if not profile_path.exists():
	return {"memory_budget_gb": 11.0, "device": "unknown"}
	with open(profile_path) as f:
	return yaml.safe_load(f) or {}


	def current_usage_gb() -> float:
	return gb(mx.get_active_memory())


	def current_peak_gb() -> float:
	return gb(mx.get_peak_memory())