linux/scripts/lib/abi/system_symbols.py

#!/usr/bin/env python3
# pylint: disable=R0902,R0912,R0914,R0915,R1702
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
# SPDX-License-Identifier: GPL-2.0

"""
Parse ABI documentation and produce results from it.
"""

import os
import re
import sys

from concurrent import futures
from datetime import datetime
from random import shuffle

from helpers import AbiDebug

class SystemSymbols:
    """Stores arguments for the class and initialize class vars"""

    def graph_add_file(self, path, link=None):
        """
        add a file path to the sysfs graph stored at self.root
        """

        if path in self.files:
            return

        name = ""
        ref = self.root
        for edge in path.split("/"):
            name += edge + "/"
            if edge not in ref:
                ref[edge] = {"__name": [name.rstrip("/")]}

            ref = ref[edge]

        if link and link not in ref["__name"]:
            ref["__name"].append(link.rstrip("/"))

        self.files.add(path)

    def print_graph(self, root_prefix="", root=None, level=0):
        """Prints a reference tree graph using UTF-8 characters"""

        if not root:
            root = self.root
            level = 0

        # Prevent endless traverse
        if level > 5:
            return

        if level > 0:
            prefix = "├──"
            last_prefix = "└──"
        else:
            prefix = ""
            last_prefix = ""

        items = list(root.items())

        names = root.get("__name", [])
        for k, edge in items:
            if k == "__name":
                continue

            if not k:
                k = "/"

            if len(names) > 1:
                k += " links: " + ",".join(names[1:])

            if edge == items[-1][1]:
                print(root_prefix + last_prefix + k)
                p = root_prefix
                if level > 0:
                    p += "   "
                self.print_graph(p, edge, level + 1)
            else:
                print(root_prefix + prefix + k)
                p = root_prefix + "│   "
                self.print_graph(p, edge, level + 1)

    def _walk(self, root):
        """
        Walk through sysfs to get all devnodes that aren't ignored.

        By default, uses /sys as sysfs mounting point. If another
        directory is used, it replaces them to /sys at the patches.
        """

        with os.scandir(root) as obj:
            for entry in obj:
                path = os.path.join(root, entry.name)
                if self.sysfs:
                    p = path.replace(self.sysfs, "/sys", count=1)
                else:
                    p = path

                if self.re_ignore.search(p):
                    return

                # Handle link first to avoid directory recursion
                if entry.is_symlink():
                    real = os.path.realpath(path)
                    if not self.sysfs:
                        self.aliases[path] = real
                    else:
                        real = real.replace(self.sysfs, "/sys", count=1)

                    # Add absfile location to graph if it doesn't exist
                    if not self.re_ignore.search(real):
                        # Add link to the graph
                        self.graph_add_file(real, p)

                elif entry.is_file():
                    self.graph_add_file(p)

                elif entry.is_dir():
                    self._walk(path)

    def __init__(self, abi, sysfs="/sys", hints=False):
        """
        Initialize internal variables and get a list of all files inside
        sysfs that can currently be parsed.

        Please notice that there are several entries on sysfs that aren't
        documented as ABI. Ignore those.

        The real paths will be stored under self.files. Aliases will be
        stored in separate, as self.aliases.
        """

        self.abi = abi
        self.log = abi.log

        if sysfs != "/sys":
            self.sysfs = sysfs.rstrip("/")
        else:
            self.sysfs = None

        self.hints = hints

        self.root = {}
        self.aliases = {}
        self.files = set()

        dont_walk = [
            # Those require root access and aren't documented at ABI
            f"^{sysfs}/kernel/debug",
            f"^{sysfs}/kernel/tracing",
            f"^{sysfs}/fs/pstore",
            f"^{sysfs}/fs/bpf",
            f"^{sysfs}/fs/fuse",

            # This is not documented at ABI
            f"^{sysfs}/module",

            f"^{sysfs}/fs/cgroup",  # this is big and has zero docs under ABI
            f"^{sysfs}/firmware",   # documented elsewhere: ACPI, DT bindings
            "sections|notes",       # aren't actually part of ABI

            # kernel-parameters.txt - not easy to parse
            "parameters",
        ]

        self.re_ignore = re.compile("|".join(dont_walk))

        print(f"Reading {sysfs} directory contents...", file=sys.stderr)
        self._walk(sysfs)

    def check_file(self, refs, found):
        """Check missing ABI symbols for a given sysfs file"""

        res_list = []

        try:
            for names in refs:
                fname = names[0]

                res = {
                    "found": False,
                    "fname": fname,
                    "msg": "",
                }
                res_list.append(res)

                re_what = self.abi.get_regexes(fname)
                if not re_what:
                    self.abi.log.warning(f"missing rules for {fname}")
                    continue

                for name in names:
                    for r in re_what:
                        if self.abi.debug & AbiDebug.UNDEFINED:
                            self.log.debug("check if %s matches '%s'", name, r.pattern)
                        if r.match(name):
                            res["found"] = True
                            if found:
                                res["msg"] += f"  {fname}: regex:\n\t"
                            continue

                if self.hints and not res["found"]:
                    res["msg"] += f"  {fname} not found. Tested regexes:\n"
                    for r in re_what:
                        res["msg"] += "    " + r.pattern + "\n"

        except KeyboardInterrupt:
            pass

        return res_list

    def _ref_interactor(self, root):
        """Recursive function to interact over the sysfs tree"""

        for k, v in root.items():
            if isinstance(v, dict):
                yield from self._ref_interactor(v)

            if root == self.root or k == "__name":
                continue

            if self.abi.re_string:
                fname = v["__name"][0]
                if self.abi.re_string.search(fname):
                    yield v
            else:
                yield v


    def get_fileref(self, all_refs, chunk_size):
        """Interactor to group refs into chunks"""

        n = 0
        refs = []

        for ref in all_refs:
            refs.append(ref)

            n += 1
            if n >= chunk_size:
                yield refs
                n = 0
                refs = []

        yield refs

    def check_undefined_symbols(self, max_workers=None, chunk_size=50,
                                found=None, dry_run=None):
        """Seach ABI for sysfs symbols missing documentation"""

        self.abi.parse_abi()

        if self.abi.debug & AbiDebug.GRAPH:
            self.print_graph()

        all_refs = []
        for ref in self._ref_interactor(self.root):
            all_refs.append(ref["__name"])

        if dry_run:
            print("Would check", file=sys.stderr)
            for ref in all_refs:
                print(", ".join(ref))

            return

        print("Starting to search symbols (it may take several minutes):",
              file=sys.stderr)
        start = datetime.now()
        old_elapsed = None

        # Python doesn't support multithreading due to limitations on its
        # global lock (GIL). While Python 3.13 finally made GIL optional,
        # there are still issues related to it. Also, we want to have
        # backward compatibility with older versions of Python.
        #
        # So, use instead multiprocess. However, Python is very slow passing
        # data from/to multiple processes. Also, it may consume lots of memory
        # if the data to be shared is not small.  So, we need to group workload
        # in chunks that are big enough to generate performance gains while
        # not being so big that would cause out-of-memory.

        num_refs = len(all_refs)
        print(f"Number of references to parse: {num_refs}", file=sys.stderr)

        if not max_workers:
            max_workers = os.cpu_count()
        elif max_workers > os.cpu_count():
            max_workers = os.cpu_count()

        max_workers = max(max_workers, 1)

        max_chunk_size = int((num_refs + max_workers - 1) / max_workers)
        chunk_size = min(chunk_size, max_chunk_size)
        chunk_size = max(1, chunk_size)

        if max_workers > 1:
            executor = futures.ProcessPoolExecutor

            # Place references in a random order. This may help improving
            # performance, by mixing complex/simple expressions when creating
            # chunks
            shuffle(all_refs)
        else:
            # Python has a high overhead with processes. When there's just
            # one worker, it is faster to not create a new process.
            # Yet, User still deserves to have a progress print. So, use
            # python's "thread", which is actually a single process, using
            # an internal schedule to switch between tasks. No performance
            # gains for non-IO tasks, but still it can be quickly interrupted
            # from time to time to display progress.
            executor = futures.ThreadPoolExecutor

        not_found = []
        f_list = []
        with executor(max_workers=max_workers) as exe:
            for refs in self.get_fileref(all_refs, chunk_size):
                if refs:
                    try:
                        f_list.append(exe.submit(self.check_file, refs, found))

                    except KeyboardInterrupt:
                        return

            total = len(f_list)

            if not total:
                if self.abi.re_string:
                    print(f"No ABI symbol matches {self.abi.search_string}")
                else:
                    self.abi.log.warning("No ABI symbols found")
                return

            print(f"{len(f_list):6d} jobs queued on {max_workers} workers",
                  file=sys.stderr)

            while f_list:
                try:
                    t = futures.wait(f_list, timeout=1,
                                     return_when=futures.FIRST_COMPLETED)

                    done = t[0]

                    for fut in done:
                        res_list = fut.result()

                        for res in res_list:
                            if not res["found"]:
                                not_found.append(res["fname"])
                            if res["msg"]:
                                print(res["msg"])

                        f_list.remove(fut)
                except KeyboardInterrupt:
                    return

                except RuntimeError as e:
                    self.abi.log.warning(f"Future: {e}")
                    break

                if sys.stderr.isatty():
                    elapsed = str(datetime.now() - start).split(".", maxsplit=1)[0]
                    if len(f_list) < total:
                        elapsed += f" ({total - len(f_list)}/{total} jobs completed).  "
                    if elapsed != old_elapsed:
                        print(elapsed + "\r", end="", flush=True,
                              file=sys.stderr)
                        old_elapsed = elapsed

        elapsed = str(datetime.now() - start).split(".", maxsplit=1)[0]
        print(elapsed, file=sys.stderr)

        for f in sorted(not_found):
            print(f"{f} not found.")