Mirrors_Framework/nuttx
1
0
Fork 0
mirror of https://github.com/apache/nuttx.git synced 2025-12-13 06:14:40 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity
Files
7927c8d105d6976147d76c5af81354e47cadc6a3
nuttx/tools/pynuttx/nxgdb/fs.py
dongjiuzhu1 7927c8d105 tools/pynuttx: update fs.py base on new structure fd and file 
Update the Python script based on the PR "Separate file
descriptors from file descriptions" in fs/vfs.

Signed-off-by: dongjiuzhu1 <dongjiuzhu1@xiaomi.com>
2025-06-12 18:12:42 +08:00

499 lines
15 KiB
Python
Raw Blame History

############################################################################
# tools/pynuttx/nxgdb/fs.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 argparse
import enum
from typing import Generator, Tuple
import gdb
from . import utils
from .protocols import fs as p
from .protocols.thread import Tcb
FSNODEFLAG_TYPE_MASK = utils.get_symbol_value("FSNODEFLAG_TYPE_MASK")
CONFIG_PSEUDOFS_FILE = utils.get_symbol_value("CONFIG_PSEUDOFS_FILE")
CONFIG_PSEUDOFS_ATTRIBUTES = utils.get_symbol_value("CONFIG_PSEUDOFS_ATTRIBUTES")
CONFIG_FS_BACKTRACE = utils.get_symbol_value("CONFIG_FS_BACKTRACE")
CONFIG_NFILE_DESCRIPTORS_PER_BLOCK = utils.get_field_nitems(
"struct fdlist", "fl_prefds"
)
CONFIG_FS_SHMFS = utils.get_symbol_value("CONFIG_FS_SHMFS")
g_special_inodes = {} # Map of the special inodes including epoll, inotify, etc.
class InodeType(enum.Enum):
# define FSNODEFLAG_TYPE_PSEUDODIR 0x00000000 /* Pseudo dir (default) */
# define FSNODEFLAG_TYPE_DRIVER 0x00000001 /* Character driver */
# define FSNODEFLAG_TYPE_BLOCK 0x00000002 /* Block driver */
# define FSNODEFLAG_TYPE_MOUNTPT 0x00000003 /* Mount point */
# define FSNODEFLAG_TYPE_NAMEDSEM 0x00000004 /* Named semaphore */
# define FSNODEFLAG_TYPE_MQUEUE 0x00000005 /* Message Queue */
# define FSNODEFLAG_TYPE_SHM 0x00000006 /* Shared memory region */
# define FSNODEFLAG_TYPE_MTD 0x00000007 /* Named MTD driver */
# define FSNODEFLAG_TYPE_SOFTLINK 0x00000008 /* Soft link */
# define FSNODEFLAG_TYPE_SOCKET 0x00000009 /* Socket */
# define FSNODEFLAG_TYPE_PIPE 0x0000000a /* Pipe */
# define FSNODEFLAG_TYPE_NAMEDEVENT 0x0000000b /* Named event group */
PSEUDODIR = 0
DRIVER = 1
BLOCK = 2
MOUNTPT = 3
NAMEDSEM = 4
MQUEUE = 5
SHM = 6
MTD = 7
SOFTLINK = 8
SOCKET = 9
PIPE = 10
NAMEDEVENT = 11
UNKNOWN = 12
class Inode(utils.Value, p.Inode):
def __init__(self, obj: gdb.Value | utils.Value):
super().__init__(obj)
def __str__(self) -> str:
name = get_inode_name(self)
type = inode_gettype(self)
return f"{name}{'/' if type == InodeType.PSEUDODIR else ''}{int(self): #x} {type.name}"
def __repr__(self) -> str:
return self.__str__()
def details(self) -> str:
details = (
f"ino: {self.i_ino}, crefs: {self.i_crefs}"
f", flags: {self.i_flags}, private: {self.i_private}"
)
if CONFIG_PSEUDOFS_FILE:
details += f", i_size: {self.i_size}"
if CONFIG_PSEUDOFS_ATTRIBUTES:
details += f", i_mode: {self.i_mode}, i_owner: {self.i_owner}, i_group: {self.i_group}"
details += f", i_atime: {self.i_atime}, i_mtime: {self.i_mtime}, i_ctime: {self.i_ctime}"
return details
@property
def nodetype(self) -> InodeType:
return inode_gettype(self)
def get_inode_name(inode: p.Inode):
try:
if not inode:
return ""
def special_inode_name(inode):
global g_special_inodes
if not g_special_inodes:
g_special_inodes = {}
for special in (
"perf",
"timerfd",
"signalfd",
"dir",
"inotify",
"epoll",
"eventfd",
"sock",
):
value = utils.gdb_eval_or_none(f"g_{special}_inode")
if value:
g_special_inodes[special] = value.address
for name, value in g_special_inodes.items():
if value == inode:
return name
return None
if name := special_inode_name(inode):
return name
ptr = inode.i_name.cast(gdb.lookup_type("char").pointer())
return ptr.string()
except gdb.MemoryError:
return "<MemoryError>"
def inode_getpath(inode: p.Inode):
"""get path from inode"""
try:
if not inode:
return ""
name = get_inode_name(inode)
if inode.i_parent:
return inode_getpath(inode.i_parent) + "/" + name
return name
except gdb.MemoryError:
return "<MemoryError>"
def inode_gettype(inode: p.Inode) -> InodeType:
if not inode:
return InodeType.UNKNOWN
type = int(inode.i_flags & FSNODEFLAG_TYPE_MASK)
# check if it's a valid type in InodeType
if type in [e.value for e in InodeType]:
return InodeType(type)
return InodeType.UNKNOWN
def get_file(tcb: Tcb, fd):
group = tcb.group
fdlist = group.tg_fdlist
fl_fds = fdlist.fl_fds
fl_rows = fdlist.fl_rows
row = fd // CONFIG_NFILE_DESCRIPTORS_PER_BLOCK
col = fd % CONFIG_NFILE_DESCRIPTORS_PER_BLOCK
if row >= fl_rows:
return None
return fl_fds[row][col]
def foreach_inode(root=None, path="") -> Generator[Tuple[p.Inode, str], None, None]:
node: p.Inode = root or utils.parse_and_eval("g_root_inode").i_child
while node:
newpath = path + "/" + get_inode_name(node)
yield node, newpath
if node.i_child:
yield from foreach_inode(node.i_child, newpath)
node = node.i_peer
def foreach_file(tcb: Tcb):
"""Iterate over all file descriptors in a tcb"""
group = tcb.group
fdlist = group.tg_fdlist
fl_fds = fdlist.fl_fds
fl_rows = fdlist.fl_rows
for row in range(fl_rows):
for col in range(CONFIG_NFILE_DESCRIPTORS_PER_BLOCK):
fdp = fl_fds[row][col]
if not fdp or not fdp.f_file:
continue
fd = row * CONFIG_NFILE_DESCRIPTORS_PER_BLOCK + col
file = fdp.f_file
yield fd, fdp, file
class Fdinfo(gdb.Command):
"""Dump fd info information of process"""
def __init__(self):
super().__init__("fdinfo", gdb.COMMAND_DATA, gdb.COMPLETE_EXPRESSION)
self.total_fd_count = 0
def print_file_info(self, fd, fdp: p.Fd, file: p.File, formatter: str):
backtrace_formatter = "{0:<5} {1:<36} {2}"
oflags = int(file.f_oflags)
pos = int(file.f_pos)
path = inode_getpath(file.f_inode)
output = []
if CONFIG_FS_BACKTRACE:
backtrace = utils.Backtrace(
utils.ArrayIterator(fdp.f_backtrace), formatter=backtrace_formatter
)
output.append(
formatter.format(
fd, hex(file.address), oflags, pos, path, backtrace.formatted[0]
)
)
output.extend(
formatter.format("", "", "", "", "", bt)
for bt in backtrace.formatted[1:]
)
output.append("") # separate each backtrace
else:
output = [
formatter.format(fd, hex(file.address), hex(oflags), pos, path, "")
]
gdb.write("\n".join(output).rstrip()) # strip trailing spaces
gdb.write("\n")
def print_fdinfo_by_tcb(self, tcb):
"""print fdlist from tcb"""
gdb.write(f"PID: {tcb['pid']}\n")
group = tcb.group
if not group:
return
if group in self.processed_groups:
return
self.processed_groups.add(group)
headers = ["FD", "FILEP", "OFLAGS", "POS", "PATH", "BACKTRACE"]
formatter = "{:<4}{:<12}{:<12}{:<10}{:<48}{:<50}"
gdb.write(formatter.format(*headers) + "\n")
fd_count = 0
for fd, fdp, file in foreach_file(tcb):
self.print_file_info(fd, fdp, file, formatter)
fd_count += 1
self.total_fd_count += fd_count
gdb.write("\n")
def diagnose(self, *args, **kwargs):
output = gdb.execute("fdinfo", to_string=True)
return {
"title": "fdinfo report",
"summary": f"Total files opened:{self.total_fd_count}",
"result": "info",
"command": "fdinfo",
"message": output,
}
def invoke(self, arg, from_tty):
parser = argparse.ArgumentParser(
description="Get fdinfo for a process or all processes."
)
parser.add_argument("-p", "--pid", type=int, help="Optional process ID")
try:
args = parser.parse_args(gdb.string_to_argv(arg))
except SystemExit:
gdb.write("Invalid arguments.\n")
return
self.processed_groups = set()
self.total_fd_count = 0
tcbs = [utils.get_tcb(args.pid)] if args.pid else utils.get_tcbs()
for tcb in tcbs:
self.print_fdinfo_by_tcb(tcb)
class Mount(gdb.Command):
def __init__(self):
if not utils.get_symbol_value("CONFIG_DISABLE_MOUNTPOINT"):
super().__init__("mount", gdb.COMMAND_USER)
self.mount_count = 0
def diagnose(self, *args, **kwargs):
output = gdb.execute("mount", to_string=True)
return {
"title": "File system mount information",
"summary": f"Total {self.mount_count} mount points",
"command": "mount",
"result": "info",
"message": output or "No mount",
}
def invoke(self, args, from_tty):
self.mount_count = 0
nodes = filter(
lambda x: inode_gettype(x[0]) == InodeType.MOUNTPT, foreach_inode()
)
for node, path in nodes:
statfs = node.u.i_mops.statfs
funcname = gdb.block_for_pc(int(statfs)).function.print_name
fstype = funcname.split("_")[0]
gdb.write(" %s type %s\n" % (path, fstype))
self.mount_count += 1
class ForeachInode(gdb.Command):
"""Dump each inode info"""
def __init__(self):
super().__init__("foreach inode", gdb.COMMAND_USER)
def get_root_inode(self, addr_or_expr):
try:
return utils.Value(int(addr_or_expr, 0)).cast(
gdb.lookup_type("struct inode").pointer()
)
except ValueError:
return utils.gdb_eval_or_none(addr_or_expr)
def parse_arguments(self, argv):
parser = argparse.ArgumentParser(description="foreach inode command")
parser.add_argument(
"-L",
"--level",
type=int,
default=4096,
help="Only render the tree to a specific depth",
)
parser.add_argument(
"--nodetype",
type=str,
choices=[e.name.lower() for e in InodeType],
default=None,
help="Only show the specific type of inode",
)
parser.add_argument(
"-v",
"--verbose",
action="store_true",
help="Show more information",
)
parser.add_argument(
"addr_or_expr",
type=str,
nargs="?",
default=None,
help="set the start inode to be tranversed",
)
try:
args = parser.parse_args(argv)
except SystemExit:
return None
return args
def print_inode_info(
self, node: Inode, level=1, prefix="", maxlevel=4096, type=None, verbose=False
):
if level > maxlevel:
return
while node:
if node.i_peer:
initial_indent = prefix + "├── "
newprefix = prefix + "│ "
else:
initial_indent = prefix + "└── "
newprefix = prefix + " "
if (
not type
or node.nodetype == type
or any(n.i_parent == node for n, _ in self.nodes)
):
gdb.write(
f"{initial_indent}{node} {node.details() if verbose else ''}\n"
)
if node.i_child:
self.print_inode_info(
Inode(node.i_child), level + 1, newprefix, maxlevel, type, verbose
)
node = Inode(node.i_peer)
def diagnose(self, *args, **kwargs):
output = gdb.execute("foreach inode", to_string=True)
return {
"title": "File node information",
"summary": "inode formation dump",
"command": "foreach inode",
"result": "info",
"message": output,
}
def invoke(self, args, from_tty):
args = self.parse_arguments(args.split(" "))
if not args:
return
root = (
self.get_root_inode(args.addr_or_expr)
if args.addr_or_expr
else utils.gdb_eval_or_none("g_root_inode")
)
nodetype = InodeType[args.nodetype.upper()] if args.nodetype else None
if nodetype:
self.nodes = list(
filter(lambda x: inode_gettype(x[0]) == nodetype, foreach_inode(root))
)
self.print_inode_info(
Inode(root), maxlevel=args.level, type=nodetype, verbose=args.verbose
)
class InfoShmfs(gdb.Command):
"""Show share memory usage"""
def __init__(self):
if CONFIG_FS_SHMFS:
super().__init__("info shm", gdb.COMMAND_USER)
self.total_size = 0
self.block_count = 0
def shm_filter(self, node: p.Inode, path):
if inode_gettype(node) != InodeType.SHM:
return
obj = node.i_private.cast(gdb.lookup_type("struct shmfs_object_s").pointer())
length = obj.length
paddr = obj.paddr
print(f" {path} memsize: {length}, paddr: {paddr}")
self.total_size += length / 1024
self.block_count += 1
def diagnose(self, *args, **kwargs):
output = gdb.execute("info shm", to_string=True)
return {
"title": "Share memory usage",
"summary": f"Total used:{self.total_size}kB, {self.block_count}blocks",
"result": "info",
"command": "info shm",
"message": output or "No InfoShmfs",
}
def invoke(self, args, from_tty):
self.total_size = 0
self.block_count = 0
nodes = filter(lambda x: inode_gettype(x[0]) == InodeType.SHM, foreach_inode())
for node, path in nodes:
obj = node.i_private.cast(
gdb.lookup_type("struct shmfs_object_s").pointer()
)
length = obj.length
paddr = obj.paddr
print(f" {path} memsize: {length}, paddr: {paddr}")
self.total_size += length / 1024
self.block_count += 1
Reference in New Issue View Git Blame Copy Permalink