nuttx/tools/gdb/thread.py

############################################################################
# tools/gdb/thread.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.  The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################

import gdb
import utils
from stack import Stack
from enum import Enum, auto

UINT16_MAX = 0xFFFF
SEM_TYPE_MUTEX = 4

saved_regs = None


def save_regs():
    global saved_regs
    tcbinfo = gdb.parse_and_eval("g_tcbinfo")

    if saved_regs is not None:
        return
    arch = gdb.selected_frame().architecture()
    saved_regs = []
    i = 0
    for reg in arch.registers():
        if i >= tcbinfo["regs_num"]:
            break

        saved_regs.append(gdb.parse_and_eval("$%s" % reg.name))
        i += 1


def restore_regs():
    tcbinfo = gdb.parse_and_eval("g_tcbinfo")
    global saved_regs

    if saved_regs is None:
        return

    arch = gdb.selected_frame().architecture()
    i = 0
    for reg in arch.registers():
        if i >= tcbinfo["regs_num"]:
            break

        gdb.execute("set $%s=%d" % (reg.name, int(saved_regs[i])))
        i += 1

    saved_regs = None


class Nxsetregs(gdb.Command):
    """
    Set registers to the specified values.
    Usage: nxsetregs [regs]

    Etc: nxsetregs
         nxsetregs g_current_regs[0]
         nxsetregs tcb->xcp.regs
         Nxsetregs g_pidhash[0].tcb->xcp.regs

    Default regs is g_current_regs[0],if regs is NULL, it will not set registers.

    """

    def __init__(self):
        super(Nxsetregs, self).__init__("nxsetregs", gdb.COMMAND_USER)

    def invoke(self, args, from_tty):
        gdb.execute("set $_current_regs=tcbinfo_running_regs()")
        current_regs = gdb.parse_and_eval("$_current_regs")
        tcbinfo = gdb.parse_and_eval("g_tcbinfo")
        arg = args.split(" ")

        if arg[0] != "":
            regs = gdb.parse_and_eval("%s" % arg[0]).cast(
                gdb.lookup_type("char").pointer()
            )
        else:
            gdb.execute("set $_current_regs=tcbinfo_current_regs()")
            current_regs = gdb.parse_and_eval("$_current_regs")
            regs = current_regs.cast(gdb.lookup_type("char").pointer())

        if regs == 0:
            gdb.write("regs is NULL\n")
            return

        save_regs()
        arch = gdb.selected_frame().architecture()
        i = 0
        for reg in arch.registers():
            if i >= tcbinfo["regs_num"]:
                return

            gdb.execute("select-frame 0")
            if tcbinfo["reg_off"]["p"][i] != UINT16_MAX:
                value = gdb.Value(regs + tcbinfo["reg_off"]["p"][i]).cast(
                    gdb.lookup_type("uintptr_t").pointer()
                )[0]
                gdb.execute("set $%s = 0x%x" % (reg.name, value))

            i += 1


class Nxinfothreads(gdb.Command):
    def __init__(self):
        super(Nxinfothreads, self).__init__("info threads", gdb.COMMAND_USER)

    def invoke(self, args, from_tty):
        npidhash = gdb.parse_and_eval("g_npidhash")
        pidhash = gdb.parse_and_eval("g_pidhash")
        statenames = gdb.parse_and_eval("g_statenames")

        if utils.is_target_smp():
            gdb.write(
                "%-5s %-4s %-4s %-4s %-21s %-80s %-30s\n"
                % ("Index", "Tid", "Pid", "Cpu", "Thread", "Info", "Frame")
            )
        else:
            gdb.write("%-5s %-4s %-4s %-21s %-80s %-30s\n" % ("Index", "Tid", "Pid", "Thread", "Info", "Frame"))

        for i in range(0, npidhash):
            if pidhash[i] == 0:
                continue

            pid = pidhash[i]["group"]["tg_pid"]
            tid = pidhash[i]["pid"]

            if pidhash[i]["task_state"] == gdb.parse_and_eval("TSTATE_TASK_RUNNING"):
                index = "*%s" % i
                pc = utils.get_register_byname(utils.get_arch_pc_name(), tcb=None)
            else:
                index = " %s" % i
                pc = utils.get_register_byname(utils.get_arch_pc_name(), tcb=pidhash[i])

            thread = "Thread 0x%x" % pidhash[i]

            statename = statenames[pidhash[i]["task_state"]].string()
            if statename == "Running":
                statename = "\x1b[32;1m%s\x1b[m" % statename
            else:
                statename = "\x1b[33;1m%s\x1b[m" % statename

            if pidhash[i]["task_state"] == gdb.parse_and_eval("TSTATE_WAIT_SEM"):
                mutex = pidhash[i]["waitobj"].cast(gdb.lookup_type("sem_t").pointer())
                if mutex["flags"] & SEM_TYPE_MUTEX:
                    mutex = pidhash[i]["waitobj"].cast(gdb.lookup_type("mutex_t").pointer())
                    statename = "Waiting,Mutex:%d" % (mutex["holder"])

            try:
                """Maybe tcb not have name member, or name is not utf-8"""
                info = "(Name: \x1b[31;1m%s\x1b[m, State: %s, Priority: %d, Stack: %d)" % (
                    pidhash[i]["name"].string(),
                    statename,
                    pidhash[i]["sched_priority"],
                    pidhash[i]["adj_stack_size"],
                )
            except gdb.error and UnicodeDecodeError:
                info = "(Name: Not utf-8, State: %s, Priority: %d, Stack: %d)" % (
                    statename,
                    pidhash[i]["sched_priority"],
                    pidhash[i]["adj_stack_size"],
                )

            line = gdb.find_pc_line(pc)
            if line.symtab:
                func = gdb.execute("info symbol %d " % pc, to_string=True)
                frame = "\x1b[34;1m0x%x\x1b[\t\x1b[33;1m%s\x1b[m at %s:%d" % (
                    pc,
                    func.split()[0] + "()",
                    line.symtab,
                    line.line,
                )
            else:
                frame = "No symbol with pc"

            if utils.is_target_smp():
                cpu = "%d" % pidhash[i]["cpu"]
                gdb.write(
                    "%-5s %-4s %-4s %-4s %-21s %-80s %-30s\n" % (index, tid, pid, cpu, thread, info, frame)
                )
            else:
                gdb.write("%-5s %-4s %-4s %-21s %-80s %-30s\n" % (index, tid, pid, thread, info, frame))


class Nxthread(gdb.Command):
    def __init__(self):
        super(Nxthread, self).__init__("thread", gdb.COMMAND_USER)

    def invoke(self, args, from_tty):
        npidhash = gdb.parse_and_eval("g_npidhash")
        pidhash = gdb.parse_and_eval("g_pidhash")
        arg = args.split(" ")
        arglen = len(arg)

        if arg[0] == "":
            pass
        elif arg[0] == "apply":
            if arglen <= 1:
                gdb.write("Please specify a thread ID list\n")
            elif arglen <= 2:
                gdb.write("Please specify a command following the thread ID list\n")

            elif arg[1] == "all":
                for i in range(0, npidhash):
                    if pidhash[i] == 0:
                        continue
                    try:
                        gdb.write("Thread %d %s\n" % (i, pidhash[i]["name"].string()))
                    except gdb.error and UnicodeDecodeError:
                        gdb.write("Thread %d\n" % (i))

                    gdb.execute("nxsetregs g_pidhash[%d]->xcp.regs" % i)
                    cmd_arg = ""
                    for cmd in arg[2:]:
                        cmd_arg += cmd + " "

                    gdb.execute("%s\n" % cmd_arg)
                    restore_regs()
            else:
                threadlist = []
                i = 0
                cmd = ""
                for i in range(1, arglen):
                    if arg[i].isnumeric():
                        threadlist.append(int(arg[i]))
                    else:
                        cmd += arg[i] + " "

                if len(threadlist) == 0 or cmd == "":
                    gdb.write("Please specify a thread ID list and command\n")
                else:
                    for i in threadlist:
                        if i >= npidhash:
                            break

                        if pidhash[i] == 0:
                            continue

                        try:
                            gdb.write(
                                "Thread %d %s\n" % (i, pidhash[i]["name"].string())
                            )
                        except gdb.error and UnicodeDecodeError:
                            gdb.write("Thread %d\n" % (i))

                        gdb.execute("nxsetregs g_pidhash[%d]->xcp.regs" % i)
                        gdb.execute("%s\n" % cmd)
                        restore_regs()

        else:
            if arg[0].isnumeric() and pidhash[int(arg[0])] != 0:
                if pidhash[int(arg[0])]["task_state"] == gdb.parse_and_eval("TSTATE_TASK_RUNNING"):
                    restore_regs()
                else:
                    gdb.execute("nxsetregs g_pidhash[%s]->xcp.regs" % arg[0])
            else:
                gdb.write("Invalid thread id %s\n" % arg[0])


class Nxcontinue(gdb.Command):
    def __init__(self):
        super(Nxcontinue, self).__init__("nxcontinue", gdb.COMMAND_USER)

    def invoke(self, args, from_tty):
        restore_regs()
        gdb.execute("continue")


class Nxstep(gdb.Command):
    def __init__(self):
        super(Nxstep, self).__init__("nxstep", gdb.COMMAND_USER)

    def invoke(self, args, from_tty):
        restore_regs()
        gdb.execute("step")

class TaskType(Enum):
    TASK = 0
    PTHREAD = 1
    KTHREAD = 2

class TaskSchedPolicy(Enum):
    FIFO = 0
    RR = 1
    SPORADIC = 2

class TaskState(Enum):
    Invalid = 0
    Waiting_Unlock = auto()
    Ready = auto()
    if utils.get_symbol_value("CONFIG_SMP"):
        Assigned = auto()
    Running = auto()
    Inactive = auto()
    Waiting_Semaphore = auto()
    Waiting_Signal = auto()
    if not utils.get_symbol_value("CONFIG_DISABLE_MQUEUE") or \
        not utils.get_symbol_value("CONFIG_DISABLE_MQUEUE_SYSV"):
        Waiting_MQEmpty = auto()
        Waiting_MQFull = auto()
    if utils.get_symbol_value("CONFIG_PAGING"):
        Waiting_PagingFill = auto()
    if utils.get_symbol_value("CONFIG_SIG_SIGSTOP_ACTION"):
        Stopped = auto()

class Ps(gdb.Command):
    def __init__(self):
        super(Ps, self).__init__("ps", gdb.COMMAND_USER)
        self._fmt_wxl = "{0: <{width}}"
        # By default we align to the right, whcih respects the nuttx foramt
        self._fmt_wx = "{0: >{width}}"

    def parse_and_show_info(self, tcb):
        get_macro = lambda x : utils.get_symbol_value(x)
        eval2str = lambda cls, x : cls(int(x)).name
        cast2ptr = lambda x, t : x.cast(gdb.lookup_type(t).pointer())

        pid = int(tcb["pid"])
        group = int(tcb["group"]["tg_pid"])
        priority = int(tcb["sched_priority"])

        policy = eval2str(TaskSchedPolicy, (tcb["flags"] & get_macro("TCB_FLAG_POLICY_MASK")) \
            >> get_macro("TCB_FLAG_POLICY_SHIFT"))

        task_type = eval2str(TaskType, (tcb["flags"] & get_macro("TCB_FLAG_TTYPE_MASK")) \
            >> get_macro("TCB_FLAG_TTYPE_SHIFT"))

        npx = 'P' if (tcb["flags"] & get_macro("TCB_FLAG_EXIT_PROCESSING")) else '-'

        waiter = str(int(cast2ptr(tcb["waitobj"], "mutex_t")["holder"])) \
        if tcb["waitobj"] and cast2ptr(tcb["waitobj"], "sem_t")["flags"] & get_macro("SEM_TYPE_MUTEX") else ""
        state_and_event = eval2str(TaskState, (tcb["task_state"])) + ("@Mutex_Holder: " + waiter if waiter else "")
        state_and_event = state_and_event.split("_")

        # Append a null str here so we don't need to worry
        # about the number of elements as we only want the first two
        state, event = state_and_event if len(state_and_event) > 1 else state_and_event + [""]

        sigmask = "{0:#0{1}x}".format( \
            sum([int(tcb["sigprocmask"]['_elem'][i] << i) for i in range(get_macro("_SIGSET_NELEM"))]), \
            get_macro("_SIGSET_NELEM") * 8 + 2 \
        )[2:] # exclude "0x"

        st = Stack(
            tcb["name"].string(),
            hex(tcb["entry"]['pthread']), # should use main?
            int(tcb["stack_base_ptr"]),
            int(tcb["stack_alloc_ptr"]),
            int(tcb["adj_stack_size"]),
            utils.get_register_byname("sp", tcb),
            4,)

        stacksz = st._stack_size
        used = st.max_usage()
        filled = "{0:.2%}".format(st.max_usage() / st._stack_size)

        cpu = tcb['cpu'] if get_macro("CONFIG_SMP") else 0

        # For a task we need to display its cmdline arguments, while for a thread we display
        # pointers to its entry and argument
        cmd = ""
        name = tcb['name'].string()

        if int(tcb["flags"] & get_macro("TCB_FLAG_TTYPE_MASK")) == int(get_macro("TCB_FLAG_TTYPE_PTHREAD")):
            entry = tcb["entry"]["main"]
            ptcb = cast2ptr(tcb, "struct pthread_tcb_s")
            arg = ptcb["arg"]
            cmd = " ".join((name, hex(entry), hex(arg)))
        else:
            argv = tcb["group"]["tg_info"]["ta_argv"] + 1

            args = []
            parg = argv
            while (parg.dereference()):
                args.append(parg.dereference().string())
                parg += 1

            cmd = " ".join([name] + args)

        if not utils.get_symbol_value("CONFIG_SCHED_CPULOAD_NONE"):
            load = "{0:.1%}".format(int(tcb['ticks']) / int(gdb.parse_and_eval("g_cpuload_total")))
        else:
            load = "Dis."

        gdb.write(" ".join((
            self._fmt_wx.format(pid, width=5), self._fmt_wx.format(group, width=5), \
            self._fmt_wx.format(cpu, width=3), \
            self._fmt_wx.format(priority, width=3), self._fmt_wxl.format(policy, width=8), self._fmt_wxl.format(task_type, width=7), \
            self._fmt_wx.format(npx, width=3), self._fmt_wxl.format(state, width=8), self._fmt_wxl.format(event, width=9), \
            self._fmt_wxl.format(sigmask, width=8), \
            self._fmt_wx.format(stacksz, width=7), self._fmt_wx.format(used, width=7), self._fmt_wx.format(filled, width=6), \
            self._fmt_wx.format(load, width=6), cmd,
        )))
        gdb.write("\n")

    def invoke(self, args, from_tty):
        gdb.write(" ".join((
            self._fmt_wx.format("PID", width=5), self._fmt_wx.format("GROUP", width=5), \
            self._fmt_wx.format("CPU", width=3), \
            self._fmt_wx.format("PRI", width=3), self._fmt_wxl.format("POLICY", width=8), self._fmt_wxl.format("TYPE", width=7), \
            self._fmt_wx.format("NPX", width=3), self._fmt_wxl.format("STATE", width=8), self._fmt_wxl.format("EVENT", width=9), \
            self._fmt_wxl.format("SIGMASK", width=utils.get_symbol_value("_SIGSET_NELEM") * 8), \
            self._fmt_wx.format("STACK", width=7), self._fmt_wx.format("USED", width=7), self._fmt_wx.format("FILLED", width=3), \
            self._fmt_wx.format("LOAD", width=6), "COMMAND",
        )))
        gdb.write("\n")

        for tcb in utils.get_tcbs():
            self.parse_and_show_info(tcb)

# We can't use a user command to rename continue it will recursion

gdb.execute("define c\n nxcontinue \n end\n")
gdb.execute("define s\n nxstep \n end\n")
gdb.write("\n\x1b[31;1m if use thread command, please don't use 'continue', use 'c' instead !!!\x1b[m\n")
gdb.write("\x1b[31;1m if use thread command, please don't use 'step', use 's' instead !!!\x1b[m\n")

Nxsetregs()
Nxinfothreads()
Nxthread()
Nxcontinue()
Nxstep()
Ps()