48c102d2a4
- add process_metrics function to record peak RSS and CPU time - update JSON converter to include metrics in completion marker - modify SQLite converter to utilize new metrics for performance tracking - enhance storage size updates with locking mechanism for concurrent access
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""Task-orchestration helpers: CLI shape, dependency markers, completion
|
|
markers, lab config loading (docs/rules/build-pipeline-tasks.md,
|
|
docs/rules/code-python-style.md section 3).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import psutil
|
|
import yaml
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def parse_task_args(description: str) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=description)
|
|
parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"])
|
|
parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)")
|
|
args = parser.parse_args()
|
|
logging.basicConfig(
|
|
level=args.log_level,
|
|
stream=sys.stderr,
|
|
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
|
|
)
|
|
return args
|
|
|
|
|
|
def check_dependencies(out_root: Path, deps: list[str]) -> None:
|
|
for dep in deps:
|
|
marker = out_root / ".done" / f"{dep}.ok"
|
|
if not marker.exists():
|
|
raise FileNotFoundError(f"dependency marker missing: {marker} - run task {dep} first")
|
|
|
|
|
|
def remove_stale_marker(out_root: Path, task_id: str) -> Path:
|
|
marker = out_root / ".done" / f"{task_id}.ok"
|
|
if marker.exists():
|
|
logger.info("re-run: removing stale marker %s", marker)
|
|
marker.unlink()
|
|
return marker
|
|
|
|
|
|
def write_marker(out_root: Path, task_id: str, entries: dict) -> Path:
|
|
marker = out_root / ".done" / f"{task_id}.ok"
|
|
lines = [f"task: '{task_id}'", "status: ok"] + [f"{k}: {v}" for k, v in entries.items()]
|
|
marker.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(marker, "w", encoding="utf-8", newline="\n") as fh:
|
|
fh.write("\n".join(lines) + "\n")
|
|
return marker
|
|
|
|
|
|
def load_lab_config(out_root: Path) -> dict:
|
|
with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh:
|
|
return yaml.safe_load(fh)
|
|
|
|
|
|
def process_metrics() -> dict:
|
|
"""Peak RSS and CPU time of the current process.
|
|
|
|
Every converter records its own build cost in its marker so the task 10
|
|
hardware-sizing model can extrapolate per-format RAM/CPU needs, not just
|
|
disk. peak_wset exists on Windows; the rss fallback covers Linux, where
|
|
the value is the current (not peak) RSS - good enough for streaming
|
|
converters whose RSS is flat by design.
|
|
"""
|
|
proc = psutil.Process()
|
|
mem = proc.memory_info()
|
|
peak = getattr(mem, "peak_wset", 0) or mem.rss
|
|
cpu = proc.cpu_times()
|
|
return {
|
|
"proc_peak_rss_mb": round(peak / 1048576, 1),
|
|
"proc_cpu_seconds": round(cpu.user + cpu.system, 1),
|
|
}
|