Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136"""Resource discovery for the test gate and standalone suite runners."""
from __future__ import annotations
import osfrom pathlib import Pathimport reimport subprocessimport sys
MIB = 1024 * 1024
def cpu_count(*, include_quota: bool = True) -> int: """Return CPUs available to this process, including a Linux quota.""" cpus = os.cpu_count() or 1 if hasattr(os, "sched_getaffinity"): cpus = min(cpus, len(os.sched_getaffinity(0))) for directory in cgroups() if include_quota else []: try: quota, period = (directory / "cpu.max").read_text().split() if quota != "max": cpus = min(cpus, max(1, (int(quota) + int(period) - 1) // int(period))) except (OSError, ValueError): pass return max(1, cpus)
def cgroups() -> list[Path]: """Return the Linux v2 cgroup and its ancestors.""" root = Path("/sys/fs/cgroup") try: entry = next( line[3:] for line in Path("/proc/self/cgroup").read_text().splitlines() if line.startswith("0::") ) except (OSError, StopIteration): return [] # A cgroup namespace may expose a path outside its mounted root. directory = root / entry.lstrip("/") if ".." in directory.parts or not directory.exists(): directory = root return [ directory, *[p for p in directory.parents if p == root or root in p.parents], ]
def available_memory() -> int: """Return reclaimable host memory limited by cgroup headroom.""" available = None try: for line in Path("/proc/meminfo").read_text().splitlines(): if line.startswith("MemAvailable:"): available = int(line.split()[1]) * 1024 break except OSError: pass if available is None and sys.platform == "darwin": try: stats = subprocess.check_output(["vm_stat"], text=True) page_size = int(re.search(r"page size of (\d+) bytes", stats)[1]) pages = sum( int(match[1]) for match in re.finditer( r"Pages (?:free|inactive|speculative):\s+(\d+)", stats ) ) available = pages * page_size except (OSError, subprocess.CalledProcessError, TypeError, ValueError): pass # Unknown hosts use a conservative budget, not total physical memory. available = 2048 * MIB if available is None else available for directory in cgroups(): try: limit = (directory / "memory.max").read_text().strip() if limit != "max": used = int((directory / "memory.current").read_text()) available = min(available, max(0, int(limit) - used)) except (OSError, ValueError): pass return available
FIELDS = ("shared", "exclusive", "go", "gleam", "harness", "build")ENVIRONMENT = ( "ALBEDO_TEST_JOBS", "ALBEDO_TEST_EXCLUSIVE_JOBS", "ALBEDO_TEST_GO_JOBS", "ALBEDO_TEST_GLEAM_JOBS", "ALBEDO_TEST_HARNESS_JOBS", "ALBEDO_TEST_BUILD_JOBS",)
def plan(cpus: int, memory: int, build_cpus: int) -> dict[str, int]: """Return lane limits, leaving a quarter of available memory unused.""" budget = memory * 3 // 4 if budget < 768 * MIB: raise ValueError("test gate needs at least 1 GiB of available memory") # Half is for both E2E lanes; the rest is shared by unit suites and builds. e2e_budget = budget // 2 unit_budget = (budget - e2e_budget) // 3 lanes = max(1, min(4, cpus // 4, e2e_budget // (384 * MIB))) return dict( shared=lanes, exclusive=2 * lanes, go=lanes, gleam=max(1, min(128, cpus * 8, unit_budget // (4 * MIB))), harness=max(1, min(32, cpus * 2, unit_budget // (32 * MIB))), build=max(1, min(8, build_cpus, unit_budget // (128 * MIB))), )
def limits() -> dict[str, int]: frozen = {f: os.environ.get(n) for f, n in zip(FIELDS, ENVIRONMENT)} if all(frozen.values()): return {f: max(1, int(v)) for f, v in frozen.items() if v} # Scenarios wait on I/O; quota limits active build threads instead. values = plan(cpu_count(include_quota=False), available_memory(), cpu_count()) for field, name in zip(FIELDS, ENVIRONMENT): if name in os.environ: values[field] = max(1, int(os.environ[name])) return values
if __name__ == "__main__": import argparse
parser = argparse.ArgumentParser() parser.add_argument("--get", choices=FIELDS) args = parser.parse_args() values = limits() print(values[args.get] if args.get else " ".join(str(values[f]) for f in FIELDS))