fasta_lake.capacity

fasta_lake.capacity

Conservative laptop planning within visible host, container and scheduler limits.

detect_capacity

detect_capacity(*, proc=Path('/proc'), environ=None)

Observe capacity available to this process, not the cluster node's full RAM.

Source code in fasta_lake/capacity.py

def detect_capacity(*, proc=Path("/proc"), environ=None):
    """Observe capacity available to this process, not the cluster node's full RAM."""
    environ = os.environ if environ is None else environ
    total, available = _host_memory(proc)
    cpus = os.cpu_count() or 1
    affinity = None
    if hasattr(os, "sched_getaffinity"):
        affinity = len(os.sched_getaffinity(0))
        cpus = min(cpus, affinity)
    slurm_cpus = _integer(environ.get("SLURM_CPUS_PER_TASK"))
    if slurm_cpus:
        cpus = min(cpus, slurm_cpus)
    slurm_memory = _integer(environ.get("SLURM_MEM_PER_NODE"))
    if slurm_memory is None and slurm_cpus:
        per_cpu = _integer(environ.get("SLURM_MEM_PER_CPU"))
        if per_cpu:
            slurm_memory = per_cpu * slurm_cpus
    memory_limits = [total]
    if slurm_memory:
        memory_limits.append(slurm_memory * MIB)
    observations = _cgroup_limits(proc) if sys.platform.startswith("linux") else []
    for observation in observations:
        if observation["memory_limit_bytes"] is not None:
            memory_limits.append(observation["memory_limit_bytes"])
        if observation["memory_available_bytes"] is not None:
            available = min(available, observation["memory_available_bytes"])
        if observation["cpu_quota"] is not None:
            cpus = min(cpus, max(1, math.floor(observation["cpu_quota"])))
    capacity = min(memory_limits)
    return {
        "memory_capacity_bytes": capacity,
        "memory_available_bytes": min(available, capacity),
        "cpu_slots": max(1, cpus),
        "host_memory_bytes": total,
        "affinity_cpus": affinity,
        "slurm_cpus_per_task": slurm_cpus,
        "slurm_memory_mib": slurm_memory,
        "cgroup_observations": observations,
        "availability_note": "Point-in-time estimate; inactive file cache can be reclaimed",
    }

plan_laptop

plan_laptop(*, memory_fraction=0.35, cpu_fraction=0.5, memory_gib=None, threads=None, chunk_mib=None, capacity=None)

Choose shared-process threads and reference pieces; never infer jobs per GB.

Source code in fasta_lake/capacity.py

def plan_laptop(
    *,
    memory_fraction=0.35,
    cpu_fraction=0.5,
    memory_gib=None,
    threads=None,
    chunk_mib=None,
    capacity=None,
):
    """Choose shared-process threads and reference pieces; never infer jobs per GB."""
    for name, fraction in (("memory", memory_fraction), ("CPU", cpu_fraction)):
        if not math.isfinite(fraction) or not 0 < fraction <= 1:
            raise ValueError(f"{name} fraction must be in (0, 1]")
    for name, value in (("threads", threads), ("chunk MiB", chunk_mib)):
        if value is not None and (
            not isinstance(value, int) or isinstance(value, bool) or value < 1
        ):
            raise ValueError(f"{name} must be a positive integer")
    if memory_gib is not None and (not math.isfinite(memory_gib) or memory_gib <= 0):
        raise ValueError("Memory budget must be finite and positive")
    capacity = detect_capacity() if capacity is None else capacity
    available = capacity["memory_available_bytes"]
    ceiling = min(capacity["memory_capacity_bytes"], int(available * 0.8))
    budget = (
        int(memory_gib * GIB)
        if memory_gib is not None
        else min(
            int(capacity["memory_capacity_bytes"] * memory_fraction),
            ceiling,
        )
    )
    if budget > ceiling:
        raise ValueError("Memory budget exceeds available headroom; close other work or lower it")
    if budget < 128 * MIB:
        raise ValueError("Insufficient available memory for laptop planning")
    slots = capacity["cpu_slots"]
    if threads is not None and threads > slots:
        raise ValueError("Requested threads exceed the CPU slots available to this process")
    threads = threads or max(1, math.floor(slots * cpu_fraction))
    chunk_ceiling = max(1, budget // (16 * MIB))
    if chunk_mib is not None and chunk_mib > chunk_ceiling:
        raise ValueError("Reference piece is too large for the planning budget; lower chunk MiB")
    return {
        "schema": "fastalake.laptop-plan.v1",
        "capacity": capacity,
        "memory_budget_bytes": budget,
        "memory_fraction": memory_fraction,
        "cpu_fraction": cpu_fraction,
        "threads_per_process": threads,
        "parallel_processes": 1,
        "reference_chunk_mib": chunk_mib or min(256, chunk_ceiling),
        "blas_threads": 1,
        "scope": (
            "Planning target, not an enforced RAM cap. Steps and acquisitions stay sequential; "
            "Rust threads share process data."
        ),
        "enforcement": (
            "For a hard job limit use Docker --memory or Slurm --mem. "
            "Chunk size does not bound razor, Sage or quantification RAM."
        ),
    }