Skip to content

Monitor

This page documents the monitoring stack under jumper_extension.monitor, including the core monitor and metric backends. High‑level usage is described in the Public API and Jupyter API sections; the content below is generated directly from the Python code.

Core

MonitorUnavailableError

Bases: RuntimeError

This monitor is a stub and cannot be used.

Source code in jumper_extension/monitor/common.py 
32
33
class MonitorUnavailableError(RuntimeError):
    """This monitor is a stub and cannot be used."""

OfflinePerformanceMonitor

Offline monitor that satisfies MonitorProtocol.

Holds static DataFrames plus metadata from a manifest; does not collect live data.

Source code in jumper_extension/monitor/common.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class OfflinePerformanceMonitor:
    """Offline monitor that satisfies MonitorProtocol.

    Holds static DataFrames plus metadata from a manifest; does not collect live data.
    """

    def __init__(
        self,
        *,
        manifest: Dict,
        perf_dfs: Dict[str, pd.DataFrame],
        source: Optional[str] = None,
    ):
        monitor_info = manifest.get("monitor", {})

        self.interval = float(monitor_info.get("interval", 1.0) or 1.0)
        self.running = False
        self.start_time = monitor_info.get("start_time")
        self.stop_time = monitor_info.get("stop_time")
        self.wallclock_start_time = monitor_info.get("wallclock_start_time")
        self.wallclock_stop_time = monitor_info.get("wallclock_stop_time")

        num_cpus = int(monitor_info.get("num_cpus", 0) or 0)
        num_system_cpus = int(monitor_info.get("num_system_cpus", num_cpus) or num_cpus)
        num_gpus = int(monitor_info.get("num_gpus", 0) or 0)

        node_info = NodeInfo(
            node="local",
            num_cpus=num_cpus,
            num_system_cpus=num_system_cpus,
            num_gpus=num_gpus,
            gpu_memory=float(monitor_info.get("gpu_memory", 0.0) or 0.0),
            gpu_name=monitor_info.get("gpu_name", "") or "",
            memory_limits=monitor_info.get("memory_limits", {}) or {},
            cpu_handles=monitor_info.get("cpu_handles", []) or [],
        )

        self.nodes = NodeDataStore()
        self.nodes.register_node(node_info)
        self.nodes.load_frames("local", perf_dfs or {})

        self.is_imported = True
        self.session_source = source

    def start(self, interval: float = 1.0) -> None:
        self.interval = interval
        self.running = False

    def stop(self) -> None:
        self.running = False

UnavailablePerformanceMonitor

A stub that type-checks against MonitorProtocol but fails at runtime.

Any attribute access or method call raises MonitorUnavailableError, except 'running', which always returns False.

Source code in jumper_extension/monitor/common.py 
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
class UnavailablePerformanceMonitor:
    """
    A stub that type-checks against MonitorProtocol but fails at runtime.

    Any attribute access or method call raises MonitorUnavailableError,
    except 'running', which always returns False.
    """

    interval: float
    running: bool
    start_time: Optional[float]
    wallclock_start_time: Optional[float]
    wallclock_stop_time: Optional[float]
    nodes: NodeDataStore
    is_imported: bool
    session_source: Optional[str]

    def start(self, interval: float = 1.0) -> None: ...
    def stop(self) -> None: ...

    def __init__(self, reason: str = "Performance monitor is not available"):
        object.__setattr__(self, "_reason", reason)

    def __getattribute__(self, name: str) -> Any:
        if name in {
            "_reason", "__class__", "__repr__", "__str__",
            "__init__", "__getattribute__", "__setattr__",
            "__dict__", "__annotations__"
        }:
            return object.__getattribute__(self, name)

        if name == "running":
            return False

        reason = object.__getattribute__(self, "_reason")
        raise MonitorUnavailableError(f"Access to '{name}' is not allowed: {reason}")

    def __setattr__(self, name: str, value):
        if name in {"_reason", "__dict__", "__annotations__"}:
            return object.__setattr__(self, name, value)
        reason = object.__getattribute__(self, "_reason")
        raise MonitorUnavailableError(f"Setting '{name}' is not allowed: {reason}")

    def __repr__(self) -> str:
        return f"<UnavailablePerformanceMonitor: {self._reason}>"

CPU

PsutilCpuCollector

Bases: CpuCollectorBackend

CPU backend implemented via psutil.

Source code in jumper_extension/monitor/metrics/cpu/psutil.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class PsutilCpuCollector(CpuCollectorBackend):
    """CPU backend implemented via psutil."""

    name = "cpu-psutil"

    def collect(self, level: str, context: CollectionContext) -> list[float]:
        num_cpus = self._node_info.num_cpus
        if level == "system":
            return psutil.cpu_percent(percpu=True)
        elif level == "process":
            cpu_total = sum(
                context["cpu"].get(pid, 0.0) for pid in context["process_pids"]
            )
            return [cpu_total / num_cpus] * num_cpus
        elif level == "user":
            cpu_total = sum(
                context["cpu"].get(pid, 0.0) for pid in context["user_pids"]
            )
            return [cpu_total / num_cpus] * num_cpus
        else:  # slurm
            cpu_total = sum(
                context["cpu"].get(pid, 0.0) for pid in context["slurm_pids"]
            )
            return [cpu_total / num_cpus] * num_cpus

Memory

PsutilMemoryCollector

Bases: MemoryCollectorBackend

Memory backend implemented via psutil.

Source code in jumper_extension/monitor/metrics/memory/psutil.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class PsutilMemoryCollector(MemoryCollectorBackend):
    """Memory backend implemented via psutil."""

    name = "memory-psutil"

    def collect(self, level: str, context: CollectionContext) -> float:
        if level == "system":
            return (
                psutil.virtual_memory().total
                - psutil.virtual_memory().available
            ) / (1024**3)
        elif level == "process":
            memory_total = sum(
                context["rss"].get(pid, 0) for pid in context["process_pids"]
            )
            return memory_total / (1024**3)
        elif level == "user":
            memory_total = sum(
                context["rss"].get(pid, 0) for pid in context["user_pids"]
            )
            return memory_total / (1024**3)
        else:  # slurm
            memory_total = sum(
                context["rss"].get(pid, 0) for pid in context["slurm_pids"]
            )
            return memory_total / (1024**3)

IO

PsutilIoCollector

Bases: IoCollectorBackend

I/O backend implemented via psutil.

Source code in jumper_extension/monitor/metrics/io/psutil.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class PsutilIoCollector(IoCollectorBackend):
    """I/O backend implemented via psutil."""

    name = "io-psutil"

    def _add_io(self, totals: list[int], io_data: Any):
        if io_data:
            totals[0] += io_data.read_count
            totals[1] += io_data.write_count
            totals[2] += io_data.read_bytes
            totals[3] += io_data.write_bytes

    def collect(self, level: str, context: CollectionContext) -> list[int]:
        totals = [0, 0, 0, 0]
        if level == "process":
            for pid in context["process_pids"]:
                self._add_io(totals, context["io"].get(pid))
        elif level == "system":
            # Use disk_io_counters for a single-syscall system total
            dio = psutil.disk_io_counters()
            if dio:
                totals = [
                    dio.read_count, dio.write_count,
                    dio.read_bytes, dio.write_bytes,
                ]
        elif level == "user":
            for pid in context["user_pids"]:
                self._add_io(totals, context["io"].get(pid))
        else:  # slurm
            for pid in context["slurm_pids"]:
                self._add_io(totals, context["io"].get(pid))
        return totals

Process

PsutilProcessCollector

Bases: ProcessCollectorBackend

Process backend implemented via psutil.

Source code in jumper_extension/monitor/metrics/process/psutil.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class PsutilProcessCollector(ProcessCollectorBackend):
    """Process backend implemented via psutil."""

    name = "process-psutil"

    def setup(self) -> None:
        self._process_cache: dict[int, psutil.Process] = {}

    def _get_or_create_process(self, pid: int) -> psutil.Process:
        """Return a cached Process object for *pid*, creating one if needed."""
        proc = self._process_cache.get(pid)
        if proc is None:
            proc = psutil.Process(pid)
            self._process_cache[pid] = proc
        return proc

    def get_process_pids(self) -> set[int]:
        """Get current process PID and all its children PIDs."""
        pids = {self._pid}
        try:
            pids.update(
                child.pid for child in self._process.children(recursive=True)
            )
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            pass
        # prune cache: drop PIDs that are no longer alive
        self._process_cache = {
            p: obj for p, obj in self._process_cache.items() if p in pids
        }
        return pids

    def snapshot(self, context: CollectionContext) -> None:
        """Compute process_pids then collect cpu/memory/io for every known PID.

        Populates context["process_pids"], context["cpu"], context["rss"],
        context["io"], context["user_pids"], and context["slurm_pids"].
        """
        context["process_pids"] = self.get_process_pids()

        cpu = context["cpu"]
        rss = context["rss"]
        io = context["io"]
        process_pids = context["process_pids"]

        # 1) Snapshot process-level PIDs
        for pid in process_pids:
            proc = self._get_or_create_process(pid)
            try:
                cpu[pid] = proc.cpu_percent()
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                cpu[pid] = 0.0
            try:
                rss[pid] = proc.memory_info().rss
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                rss[pid] = 0
            try:
                io[pid] = proc.io_counters()
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                io[pid] = None

        # 2) Snapshot user/slurm-filtered processes (only the ones
        #    not already covered by the process-level set above)
        user_pids = set(process_pids)
        slurm_pids = set(process_pids)
        try:
            for proc in psutil.process_iter(["pid", "uids"]):
                pid = proc.pid
                if pid in cpu:
                    continue  # already collected above
                try:
                    is_user = proc.uids().real == self._uid
                except (psutil.AccessDenied, psutil.NoSuchProcess, IndexError):
                    is_user = False
                if not is_user:
                    continue
                user_pids.add(pid)
                # Collect metrics for this PID too
                try:
                    cpu[pid] = proc.cpu_percent()
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    cpu[pid] = 0.0
                try:
                    rss[pid] = proc.memory_info().rss
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    rss[pid] = 0
                try:
                    io[pid] = proc.io_counters()
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    io[pid] = None
                # Check slurm membership
                if is_slurm_available():
                    try:
                        if proc.environ().get("SLURM_JOB_ID") == str(
                            self._slurm_job
                        ):
                            slurm_pids.add(pid)
                    except (psutil.AccessDenied, psutil.NoSuchProcess):
                        pass
        except (psutil.NoSuchProcess, psutil.AccessDenied, IndexError):
            pass

        context["user_pids"] = user_pids
        context["slurm_pids"] = slurm_pids

    def collect(self, level: str, context: CollectionContext) -> None:
        return None

    def filter_process(self, proc: psutil.Process, mode: str) -> bool:
        """Check if process matches the filtering mode."""
        try:
            if mode == "user":
                return proc.uids().real == self._uid
            elif mode == "slurm":
                if not is_slurm_available():
                    return False
                return proc.environ().get("SLURM_JOB_ID") == str(
                    self._slurm_job
                )
        except (psutil.AccessDenied, psutil.NoSuchProcess):
            pass
        return False

    def get_filtered_processes(
        self,
        level: str = "user",
        mode: str = "cpu",
        handle: Optional[object] = None,
    ) -> list[psutil.Process]:
        """Get filtered processes for CPU monitoring."""
        if mode == "cpu":
            return [
                proc
                for proc in psutil.process_iter(["pid", "uids"])
                if self.safe_proc_call(
                    proc, lambda p: self.filter_process(p, level), False
                )
            ]
        else:
            raise ValueError(f"Unknown mode: {mode}")

    def safe_proc_call(
        self,
        proc,
        proc_func: Callable[[psutil.Process], Any],
        default=0,
    ) -> Any:
        """Safely call a process method and return default on error."""
        try:
            if not isinstance(proc, psutil.Process):
                # proc might be a pid — use cache so objects persist
                # across ticks (needed for delta-based cpu_percent)
                proc = self._get_or_create_process(proc)
            result = proc_func(proc)
            return result if result is not None else default
        except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError):
            return default
        except TypeError:
            # in test case, where psutil is a mock
            if isinstance(psutil.Process, unittest.mock.MagicMock):
                return default

filter_process(proc, mode)

Check if process matches the filtering mode.

Source code in jumper_extension/monitor/metrics/process/psutil.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def filter_process(self, proc: psutil.Process, mode: str) -> bool:
    """Check if process matches the filtering mode."""
    try:
        if mode == "user":
            return proc.uids().real == self._uid
        elif mode == "slurm":
            if not is_slurm_available():
                return False
            return proc.environ().get("SLURM_JOB_ID") == str(
                self._slurm_job
            )
    except (psutil.AccessDenied, psutil.NoSuchProcess):
        pass
    return False

get_filtered_processes(level='user', mode='cpu', handle=None)

Get filtered processes for CPU monitoring.

Source code in jumper_extension/monitor/metrics/process/psutil.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def get_filtered_processes(
    self,
    level: str = "user",
    mode: str = "cpu",
    handle: Optional[object] = None,
) -> list[psutil.Process]:
    """Get filtered processes for CPU monitoring."""
    if mode == "cpu":
        return [
            proc
            for proc in psutil.process_iter(["pid", "uids"])
            if self.safe_proc_call(
                proc, lambda p: self.filter_process(p, level), False
            )
        ]
    else:
        raise ValueError(f"Unknown mode: {mode}")

get_process_pids()

Get current process PID and all its children PIDs.

Source code in jumper_extension/monitor/metrics/process/psutil.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def get_process_pids(self) -> set[int]:
    """Get current process PID and all its children PIDs."""
    pids = {self._pid}
    try:
        pids.update(
            child.pid for child in self._process.children(recursive=True)
        )
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        pass
    # prune cache: drop PIDs that are no longer alive
    self._process_cache = {
        p: obj for p, obj in self._process_cache.items() if p in pids
    }
    return pids

safe_proc_call(proc, proc_func, default=0)

Safely call a process method and return default on error.

Source code in jumper_extension/monitor/metrics/process/psutil.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def safe_proc_call(
    self,
    proc,
    proc_func: Callable[[psutil.Process], Any],
    default=0,
) -> Any:
    """Safely call a process method and return default on error."""
    try:
        if not isinstance(proc, psutil.Process):
            # proc might be a pid — use cache so objects persist
            # across ticks (needed for delta-based cpu_percent)
            proc = self._get_or_create_process(proc)
        result = proc_func(proc)
        return result if result is not None else default
    except (psutil.NoSuchProcess, psutil.AccessDenied, AttributeError):
        return default
    except TypeError:
        # in test case, where psutil is a mock
        if isinstance(psutil.Process, unittest.mock.MagicMock):
            return default

snapshot(context)

Compute process_pids then collect cpu/memory/io for every known PID.

Populates context["process_pids"], context["cpu"], context["rss"], context["io"], context["user_pids"], and context["slurm_pids"].

Source code in jumper_extension/monitor/metrics/process/psutil.py 
 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
107
108
109
110
111
112
113
def snapshot(self, context: CollectionContext) -> None:
    """Compute process_pids then collect cpu/memory/io for every known PID.

    Populates context["process_pids"], context["cpu"], context["rss"],
    context["io"], context["user_pids"], and context["slurm_pids"].
    """
    context["process_pids"] = self.get_process_pids()

    cpu = context["cpu"]
    rss = context["rss"]
    io = context["io"]
    process_pids = context["process_pids"]

    # 1) Snapshot process-level PIDs
    for pid in process_pids:
        proc = self._get_or_create_process(pid)
        try:
            cpu[pid] = proc.cpu_percent()
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            cpu[pid] = 0.0
        try:
            rss[pid] = proc.memory_info().rss
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            rss[pid] = 0
        try:
            io[pid] = proc.io_counters()
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            io[pid] = None

    # 2) Snapshot user/slurm-filtered processes (only the ones
    #    not already covered by the process-level set above)
    user_pids = set(process_pids)
    slurm_pids = set(process_pids)
    try:
        for proc in psutil.process_iter(["pid", "uids"]):
            pid = proc.pid
            if pid in cpu:
                continue  # already collected above
            try:
                is_user = proc.uids().real == self._uid
            except (psutil.AccessDenied, psutil.NoSuchProcess, IndexError):
                is_user = False
            if not is_user:
                continue
            user_pids.add(pid)
            # Collect metrics for this PID too
            try:
                cpu[pid] = proc.cpu_percent()
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                cpu[pid] = 0.0
            try:
                rss[pid] = proc.memory_info().rss
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                rss[pid] = 0
            try:
                io[pid] = proc.io_counters()
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                io[pid] = None
            # Check slurm membership
            if is_slurm_available():
                try:
                    if proc.environ().get("SLURM_JOB_ID") == str(
                        self._slurm_job
                    ):
                        slurm_pids.add(pid)
                except (psutil.AccessDenied, psutil.NoSuchProcess):
                    pass
    except (psutil.NoSuchProcess, psutil.AccessDenied, IndexError):
        pass

    context["user_pids"] = user_pids
    context["slurm_pids"] = slurm_pids

GPU

GpuCollectorBackend

Bases: CollectorBackend

A pluggable backend that provides GPU discovery and metric collection.

Source code in jumper_extension/monitor/metrics/gpu/common.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
class GpuCollectorBackend(CollectorBackend):
    """A pluggable backend that provides GPU discovery and metric collection."""

    name = "gpu-base"

    def __init__(self, uid: int, slurm_job: str):
        self._uid = uid
        self._slurm_job = slurm_job

    def shutdown(self) -> None:
        """Clean up resources if needed."""
        return None

    def _iter_handles(self) -> Iterable[object]:
        return []

    def _filter_process(self, pid: int, mode: str) -> bool:
        """Filter a GPU process by user/slurm membership."""
        try:
            proc = psutil.Process(pid)
            if mode == "user":
                return proc.uids().real == self._uid
            elif mode == "slurm":
                if not is_slurm_available():
                    return False
                return proc.environ().get("SLURM_JOB_ID") == str(self._slurm_job)
        except (psutil.AccessDenied, psutil.NoSuchProcess):
            pass
        return False

    def _collect_system(self, handle: object) -> tuple[float, float, float]:
        raise NotImplementedError

    def _collect_process(
        self, handle: object, context: CollectionContext
    ) -> tuple[float, float, float]:
        raise NotImplementedError

    def _collect_other(
        self, handle: object, level: str, context: CollectionContext
    ) -> tuple[float, float, float]:
        raise NotImplementedError

    def collect(
        self,
        level: str,
        context: CollectionContext,
    ) -> tuple[list[float], list[float], list[float]]:
        """Collect metrics for the given level.

        Returns: (gpu_util, gpu_band, gpu_mem)
        """
        gpu_util, gpu_band, gpu_mem = [], [], []

        for handle in self._iter_handles():
            if level == "system":
                util, band, mem = self._collect_system(handle)
            elif level == "process":
                util, band, mem = self._collect_process(handle, context)
            else:  # user or slurm
                util, band, mem = self._collect_other(handle, level, context)
            gpu_util.append(util)
            gpu_band.append(band)
            gpu_mem.append(mem)

        return gpu_util, gpu_band, gpu_mem

collect(level, context)

Collect metrics for the given level.

Returns: (gpu_util, gpu_band, gpu_mem)

Source code in jumper_extension/monitor/metrics/gpu/common.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def collect(
    self,
    level: str,
    context: CollectionContext,
) -> tuple[list[float], list[float], list[float]]:
    """Collect metrics for the given level.

    Returns: (gpu_util, gpu_band, gpu_mem)
    """
    gpu_util, gpu_band, gpu_mem = [], [], []

    for handle in self._iter_handles():
        if level == "system":
            util, band, mem = self._collect_system(handle)
        elif level == "process":
            util, band, mem = self._collect_process(handle, context)
        else:  # user or slurm
            util, band, mem = self._collect_other(handle, level, context)
        gpu_util.append(util)
        gpu_band.append(band)
        gpu_mem.append(mem)

    return gpu_util, gpu_band, gpu_mem

shutdown()

Clean up resources if needed.

Source code in jumper_extension/monitor/metrics/gpu/common.py 
19
20
21
def shutdown(self) -> None:
    """Clean up resources if needed."""
    return None

GpuDiscovery

Finds all available GPU device backends at runtime.

Source code in jumper_extension/monitor/metrics/gpu/common.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class GpuDiscovery:
    """Finds all available GPU device backends at runtime."""

    def __init__(self, uid: int, slurm_job: str):
        self._uid = uid
        self._slurm_job = slurm_job

    def discover(self) -> list[GpuCollectorBackend]:
        from jumper_extension.monitor.metrics.gpu.nvml import NvmlGpuCollector
        from jumper_extension.monitor.metrics.gpu.adlx import AdlxGpuCollector

        return [
            NvmlGpuCollector(uid=self._uid, slurm_job=self._slurm_job),
            AdlxGpuCollector(uid=self._uid, slurm_job=self._slurm_job),
        ]

MultiGpuCollector

Combined GPU pipeline member — aggregates all discovered device backends.

Source code in jumper_extension/monitor/metrics/gpu/common.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class MultiGpuCollector:
    """Combined GPU pipeline member — aggregates all discovered device backends."""

    def __init__(self, uid: int, slurm_job: str):
        self._uid = uid
        self._slurm_job = slurm_job
        self._backends: list[GpuCollectorBackend] = []

    def setup(self) -> dict:
        self._backends = GpuDiscovery(self._uid, self._slurm_job).discover()
        gpu_memory = 0.0
        gpu_name_parts = []
        for backend in self._backends:
            meta = backend.setup() or {}
            if meta.get("gpu_memory", 0) > 0 and gpu_memory == 0:
                gpu_memory = meta["gpu_memory"]
            if meta.get("gpu_name"):
                gpu_name_parts.append(meta["gpu_name"])
        num_gpus = sum(len(b._handles) for b in self._backends)
        return {
            "num_gpus": num_gpus,
            "gpu_memory": gpu_memory,
            "gpu_name": ", ".join(gpu_name_parts),
        }

    def snapshot(self, context: CollectionContext) -> None:
        for backend in self._backends:
            backend.snapshot(context)

    def collect(
        self,
        level: str,
        context: CollectionContext,
    ) -> tuple[list[float], list[float], list[float]]:
        util, band, mem = [], [], []
        for backend in self._backends:
            backend_util, backend_band, backend_mem = backend.collect(level, context)
            util.extend(backend_util)
            band.extend(backend_band)
            mem.extend(backend_mem)
        return util, band, mem

NullGpuCollector

Bases: GpuCollectorBackend

A no-op backend used when no GPU backend is available.

Source code in jumper_extension/monitor/metrics/gpu/common.py 
78
79
80
81
82
83
84
class NullGpuCollector(GpuCollectorBackend):
    """A no-op backend used when no GPU backend is available."""

    name = "gpu-disabled"

    def _iter_handles(self) -> Iterable[object]:
        return []

NvmlGpuCollector

Bases: GpuCollectorBackend

NVIDIA NVML backend (uses pynvml).

Source code in jumper_extension/monitor/metrics/gpu/nvml.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class NvmlGpuCollector(GpuCollectorBackend):
    """NVIDIA NVML backend (uses pynvml)."""

    name = "nvidia-nvml"

    def __init__(self, uid: int, slurm_job: str):
        super().__init__(uid, slurm_job)
        self._pynvml = None
        self._handles = []

    def _iter_handles(self) -> Iterable[object]:
        return self._handles

    def _get_util_rates(self, handle: object):
        if self._pynvml is None:
            class DefaultUtilRates:
                gpu = 0.0
                memory = 0.0

            return DefaultUtilRates()
        try:
            return self._pynvml.nvmlDeviceGetUtilizationRates(handle)
        except self._pynvml.NVMLError:
            # If permission denied or other error, use default values
            class DefaultUtilRates:
                gpu = 0.0
                memory = 0.0

            return DefaultUtilRates()

    def setup(self) -> dict:
        # Logic is intentionally kept identical to the previous implementation.
        try:
            import pynvml

            pynvml.nvmlInit()
            self._pynvml = pynvml
            globals()["pynvml"] = pynvml
            ngpus = self._pynvml.nvmlDeviceGetCount()
            self._handles = [
                self._pynvml.nvmlDeviceGetHandleByIndex(i)
                for i in range(ngpus)
            ]
            if self._handles:
                handle = self._handles[0]
                gpu_mem = round(
                    self._pynvml.nvmlDeviceGetMemoryInfo(handle).total
                    / (1024**3),
                    2,
                    )
                name = self._pynvml.nvmlDeviceGetName(handle)
                gpu_name = name.decode() if isinstance(name, bytes) else name
                return {"gpu_memory": gpu_mem, "gpu_name": gpu_name}
        except ImportError:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.PYNVML_NOT_AVAILABLE
                ]
            )
            self._handles = []
        except Exception:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.NVIDIA_DRIVERS_NOT_AVAILABLE
                ]
            )
            self._handles = []
        return {}

    def _collect_system(self, handle: object) -> tuple[float, float, float]:
        util_rates = self._get_util_rates(handle)
        memory_info = self._pynvml.nvmlDeviceGetMemoryInfo(handle)
        return util_rates.gpu, 0.0, memory_info.used / (1024**3)

    def _collect_process(
        self,
        handle: object,
        context: CollectionContext,
    ) -> tuple[float, float, float]:
        util_rates = self._get_util_rates(handle)
        pids = context["process_pids"]
        process_mem = (
                sum(
                    p.usedGpuMemory
                    for p in self._pynvml.nvmlDeviceGetComputeRunningProcesses(
                        handle
                    )
                    if p.pid in pids and p.usedGpuMemory
                )
                / (1024**3)
        )
        return util_rates.gpu if process_mem > 0 else 0.0, 0.0, process_mem

    def _collect_other(
        self,
        handle: object,
        level: str,
        context: CollectionContext,
    ) -> tuple[float, float, float]:
        util_rates = self._get_util_rates(handle)
        if self._pynvml is None:
            return 0.0, 0.0, 0.0
        try:
            all_processes = self._pynvml.nvmlDeviceGetComputeRunningProcesses(
                handle
            )
            filtered_gpu_processes = [
                p for p in all_processes
                if self._filter_process(p.pid, level)
            ]
        except Exception:
            return 0.0, 0.0, 0.0
        filtered_mem = (
                sum(
                    p.usedGpuMemory
                    for p in filtered_gpu_processes
                    if p.usedGpuMemory
                )
                / (1024**3)
        )
        filtered_util = (
            (
                    util_rates.gpu
                    * len(filtered_gpu_processes)
                    / max(len(all_processes), 1)
            )
            if filtered_gpu_processes
            else 0.0
        )
        return filtered_util, 0.0, filtered_mem

    def shutdown(self) -> None:
        return None

AdlxGpuCollector

Bases: GpuCollectorBackend

AMD ADLX backend (uses ADLXPybind).

Source code in jumper_extension/monitor/metrics/gpu/adlx.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 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
107
108
109
110
class AdlxGpuCollector(GpuCollectorBackend):
    """AMD ADLX backend (uses ADLXPybind)."""

    name = "amd-adlx"

    def __init__(self, uid: int, slurm_job: str):
        super().__init__(uid, slurm_job)
        self._adlx_helper = None
        self._adlx_system = None
        self._handles = []

    def _iter_handles(self) -> Iterable[object]:
        return self._handles

    def setup(self) -> dict:
        # Logic is intentionally kept identical to the previous implementation.
        try:
            from ADLXPybind import ADLXHelper, ADLX_RESULT

            self._adlx_helper = ADLXHelper()
            if self._adlx_helper.Initialize() != ADLX_RESULT.ADLX_OK:
                self._handles = []
                return {}
            self._adlx_system = self._adlx_helper.GetSystemServices()
            gpus_list = self._adlx_system.GetGPUs()
            num_amd_gpus = gpus_list.Size()
            self._handles = [
                gpus_list.At(i) for i in range(num_amd_gpus)
            ]
            if self._handles:
                gpu = self._handles[0]
                # Get memory info
                gpu_mem_info = gpu.TotalVRAM()
                gpu_mem = round(gpu_mem_info / (1024**3), 2)
                # Get GPU name
                gpu_name = gpu.Name()
                return {"gpu_memory": gpu_mem, "gpu_name": gpu_name}
        except ImportError:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.ADLX_NOT_AVAILABLE
                ]
            )
        except Exception:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.AMD_DRIVERS_NOT_AVAILABLE
                ]
            )
        self._handles = []
        return {}

    def _collect_system(self, handle: object) -> tuple[float, float, float]:
        try:
            if self._adlx_system is None:
                return 0.0, 0.0, 0.0
            # Get performance metrics interface
            perf_monitoring = (
                self._adlx_system.GetPerformanceMonitoringServices()
            )

            # Get current metrics
            current_metrics = perf_monitoring.GetCurrentPerformanceMetrics(
                handle
            )

            # Get GPU utilization
            util = current_metrics.GPUUsage()

            # Get memory info
            mem_info = current_metrics.GPUVRAMUsage()

            # AMD ADLX doesn't provide memory bandwidth easily
            return util, 0.0, mem_info / 1024.0
        except Exception:
            # If we can't get metrics, return zeros
            return 0.0, 0.0, 0.0

    def _collect_process(
        self,
        handle: object,
        context: CollectionContext,
    ) -> tuple[float, float, float]:
        # AMD ADLX doesn't provide per-process metrics easily
        return 0.0, 0.0, 0.0

    def _collect_other(
        self,
        handle: object,
        level: str,
        context: CollectionContext,
    ) -> tuple[float, float, float]:
        # AMD ADLX doesn't provide per-user metrics easily
        return 0.0, 0.0, 0.0

    def shutdown(self) -> None:
        return None