emdbg.debug.px4.system_load

 1# Copyright (c) 2023, Auterion AG
 2# SPDX-License-Identifier: BSD-3-Clause
 3
 4from __future__ import annotations
 5from functools import cached_property
 6from . import utils
 7from .base import Base
 8from .device import Device
 9import logging
10LOGGER = logging.getLogger(__name__)
11
12
13class SystemLoad(Base):
14    """
15    PX4 system load monitor using the `cpuload.cpp` module inside PX4.
16    Do not use this class directly, instead access the cpu load information via
17    the `emdbg.debug.px4.task.all_tasks_as_table()` function.
18    """
19    def __init__(self, gdb):
20        super().__init__(gdb)
21        self._device = Device(gdb)
22        self._system_load = self.lookup_global_symbol_ptr("system_load")
23        self._monitor = self.lookup_static_symbol_ptr("cpuload_monitor_all_count")
24        self._psl = None
25        self.restart()
26
27    def _load(self):
28        sl = self._system_load.dereference()
29        tasks = {"hrt": self._device.uptime, "start": int(sl["start_time"]), "tasks": {}}
30        for task in utils.gdb_iter(sl["tasks"]):
31            if not task["valid"]: continue
32            tcb = int(task["tcb"])
33            total = int(task["total_runtime"])
34            tasks["tasks"][tcb] = total
35        return tasks
36
37    def restart(self):
38        """
39        Enables the cpuload monitor in PX4 and loads the first reference value.
40        """
41        if self._system_load is None: return
42        if self._monitor.dereference()["_value"] == 0:
43            self._gdb.execute("set cpuload_monitor_all_count = 1")
44            self._gdb.execute(f"set system_load.start_time = {self._device.uptime}")
45            for ii in range(1, self._system_load["tasks"].type.range()[1]):
46                self._gdb.execute(f"set system_load.tasks[{ii}].total_runtime = 0")
47                self._gdb.execute(f"set system_load.tasks[{ii}].curr_start_time = 0")
48        self._psl = self._load()
49
50
51    def stop(self):
52        """Disables the cpuload monitor"""
53        if self._monitor is None: return
54        self._gdb.execute("set cpuload_monitor_all_count = 0")
55
56    @cached_property
57    def sample(self) -> tuple[int, int, dict[int, tuple[int, int]]]:
58        """
59        Samples the cpuload monitor and computes the difference to the last
60        sample.
61        :return: a tuple of (start time since enabled [µs], interval from last
62                 sample [µs], tasks dict[tcb [ptr] -> (total runtime [µs], difference
63                 from last sample [µs])]).
64        """
65        if self._system_load is None or not self._system_load["initialized"]:
66            return (0, 0, {})
67
68        sl = self._load()
69        # compute the delta running times
70        sample = {}
71        for tcb, total in sl["tasks"].items():
72            ptotal = self._psl["tasks"].get(tcb, total)
73            sample[tcb] = (total, total - ptotal)
74        interval = self._device.uptime - self._psl["hrt"]
75        # remember the new system load as previous load
76        self._psl = sl
77        return (sl["start"], interval, sample)
78
79
80_SYSTEM_LOAD = None
81def system_load(gdb):
82    """
83    :return: The SystemLoad singleton object.
84    """
85    global _SYSTEM_LOAD
86    if _SYSTEM_LOAD is None:
87        _SYSTEM_LOAD = SystemLoad(gdb)
88    return _SYSTEM_LOAD
89
90
91def restart_system_load_monitor(gdb):
92    """
93    Starts the system load monitor if not started, else resets the sample
94    interval to right now.
95    """
96    system_load(gdb).sample
97    system_load(gdb).restart()