Skip to content

API Reference

Allocation dataclass

User-declared resource request, mirroring typical Slurm flags.

Source code in src/prevc/checks.py 
25
26
27
28
29
30
31
32
@dataclass
class Allocation:
    """User-declared resource request, mirroring typical Slurm flags."""

    cpus: int | None = None
    mem_gb: float | None = None
    gpu: bool = False
    timeout_s: float | None = None

Finding dataclass

A single diagnostic message produced by a check.

Source code in src/prevc/report.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@dataclass
class Finding:
    """A single diagnostic message produced by a check."""

    level: str  # "info" | "warning" | "critical"
    category: (
        str  # "memory" | "parallelism" | "network" | "io" | "gpu" | "runtime"
    )
    message: str
    suggestion: str | None = None

    def __str__(self) -> str:
        tag = {"info": "INFO", "warning": "WARN", "critical": "CRIT"}[
            self.level
        ]
        s = f"[{tag}][{self.category}] {self.message}"
        if self.suggestion:
            s += f"\n    -> suggestion: {self.suggestion}"
        return s

Snapshot dataclass

One sample of the whole process tree at a point in time.

Source code in src/prevc/report.py 
12
13
14
15
16
17
18
19
20
21
22
23
@dataclass
class Snapshot:
    """One sample of the whole process tree at a point in time."""

    t: float  # seconds since dry-run start
    n_processes: int
    total_threads: int
    threads_per_process: list[int]
    rss_bytes: int  # summed RSS across all processes
    open_files: int
    external_connections: int  # connections to non-local/non-private IPs
    gpu_mem_bytes: int = 0

prevalidate(command, *, cwd=None, env=None, timeout=60.0, sample_interval=0.2, cpus_allocated=None, mem_allocated_gb=None, gpu_allocated=False, print_report=True)

Run command as a subprocess, sample its whole process tree while it runs, and return a Report with peak resource usage and diagnostic Findings relevant to running this on a shared Slurm cluster.

command can be a shell string or an argv list -- this is what makes the tool language-agnostic: it wraps any executable via the OS process tree, not a specific language runtime.

Parameters mirroring Slurm allocation flags (cpus_allocated, mem_allocated_gb, gpu_allocated) are optional but strongly recommended: without them, prevc can only report what happened, not whether it fits your intended #SBATCH request.

Source code in src/prevc/core.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def prevalidate(
    command: CommandLike,
    *,
    cwd: str | None = None,
    env: dict | None = None,
    timeout: float = 60.0,
    sample_interval: float = 0.2,
    cpus_allocated: int | None = None,
    mem_allocated_gb: float | None = None,
    gpu_allocated: bool = False,
    print_report: bool = True,
) -> Report:
    """
    Run `command` as a subprocess, sample its whole process tree while it
    runs, and return a Report with peak resource usage and diagnostic
    Findings relevant to running this on a shared Slurm cluster.

    `command` can be a shell string or an argv list -- this is what makes
    the tool language-agnostic: it wraps any executable via the OS process
    tree, not a specific language runtime.

    Parameters mirroring Slurm allocation flags (cpus_allocated,
    mem_allocated_gb, gpu_allocated) are optional but strongly recommended:
    without them, prevc can only report *what happened*, not whether
    it fits your intended #SBATCH request.
    """
    is_shell = isinstance(command, str)
    t_start = time.monotonic()

    proc = subprocess.Popen(
        command,
        shell=is_shell,
        cwd=cwd,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    monitor = ProcessTreeMonitor(proc.pid, interval=sample_interval)
    monitor.start()

    timed_out = False
    try:
        proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True
        proc.kill()
        proc.wait()
    finally:
        monitor.stop()

    wall_time = time.monotonic() - t_start
    cmd_str = command if is_shell else " ".join(command)

    report = Report(
        command=cmd_str,
        exit_code=proc.returncode,
        timed_out=timed_out,
        wall_time_s=wall_time,
        snapshots=monitor.snapshots,
    )
    _aggregate(report)

    alloc = Allocation(
        cpus=cpus_allocated,
        mem_gb=mem_allocated_gb,
        gpu=gpu_allocated,
        timeout_s=timeout,
    )
    for check_fn in ALL_CHECKS:
        report.findings.extend(check_fn(report, alloc))

    if print_report:
        print(report.summary())

    return report