emdbg.debug.px4.semaphore

 1# Copyright (c) 2023, Auterion AG
 2# SPDX-License-Identifier: BSD-3-Clause
 3
 4from __future__ import annotations
 5from functools import cached_property
 6from dataclasses import dataclass
 7from . import utils
 8from .base import Base
 9
10
11class Semaphore(Base):
12    """
13    Accessor for a NuttX semaphore
14    """
15
16    @dataclass
17    class Holder:
18        task: "emdbg.debug.px4.task.Task"
19        """The task this holder belongs to"""
20        count: int
21        """How many times the task is holding this semaphore"""
22
23    def __init__(self, gdb, sem_ptr: "gdb.Value"):
24        super().__init__(gdb)
25        self._sem = sem_ptr
26        self.count: int = self._sem["semcount"]
27        """Semaphore count, positive if available, 0 if taken, negative if tasks are waiting"""
28
29    @cached_property
30    def flags(self) -> int:
31        """:return: The flags field of the semaphore"""
32        return utils.gdb_getfield(self._sem, "flags", 0b1)
33
34    @cached_property
35    def holders(self) -> list[Holder]:
36        """:return: a list of semaphore holders"""
37        from .task import Task
38
39        #if CONFIG_SEM_PREALLOCHOLDERS == 0
40        if holder := utils.gdb_getfield(self._sem, "holder"):
41            return [self.Holder(Task(self._gdb, holder[0]["htcb"]), int(holder[0]["counts"]))]
42
43        #if CONFIG_SEM_PREALLOCHOLDERS > 0
44        holders = []
45        holder = self._sem["hhead"]
46        while(holder):
47            holders.append(self.Holder(Task(self._gdb, holder["htcb"]), int(holder["counts"])))
48            holder = holder["flink"]
49        return holders
50
51    @property
52    def has_priority_inheritance(self) -> bool:
53        """:return: `True` if the semaphore supports priority inheritance"""
54        return not (self.flags & 0b1)
55
56    def to_string(self) -> str:
57        sc = self.count
58        state = "taken" if sc <= 0 else "available"
59        ostr = f"{state} ({sc}{'i' if self.has_priority_inheritance else ''})"
60        if hstr := [f"{h.count}x {h.task.name}" for h in self.holders]:
61            ostr += f" holders: {', '.join(hstr)}"
62        return ostr