#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
"""Failed system call counts."""
from __future__ import annotations

import argparse
from collections import defaultdict
import sys
from typing import Optional
import perf

class FailedSyscalls:
    """Tracks and displays failed system call totals."""
    def __init__(self, comm: Optional[str] = None) -> None:
        self.failed_syscalls: dict[str, int] = defaultdict(int)
        self.for_comm = comm
        self.session: Optional[perf.session] = None

    def process_event(self, sample: perf.sample_event) -> None:
        """Process sys_exit events."""
        event_name = str(sample.evsel)
        if not event_name.startswith("evsel(syscalls:sys_exit") and \
           not event_name.startswith("evsel(raw_syscalls:sys_exit"):
            return

        try:
            ret = sample.ret
        except AttributeError:
            print("ERROR: tracepoint fields missed", file=sys.stderr)
            sys.exit(1)

        if ret > 0:
            if ret >= 0xfffffffffffff000:  # 64-bit negative errors
                ret -= 0x10000000000000000
            elif 0xfffff000 <= ret <= 0xffffffff:  # 32-bit negative errors
                assert self.session is not None
                if not self.session.is_64_bit:
                    ret -= 0x100000000

        # Linux kernel syscalls only return error codes in the [-4095, -1] (-MAX_ERRNO)
        # range; other negative signed values are valid unsigned user-space addresses.
        if not -4095 <= ret < 0:
            return

        assert self.session is not None
        try:
            thread = self.session.find_thread(sample.sample_pid, sample.sample_tid)
            comm = (thread.comm() if thread else None) or "unknown"
        except (TypeError, AttributeError):
            # find_thread returns None when the thread isn't known.
            comm = "unknown"

        if self.for_comm and comm != self.for_comm:
            return

        self.failed_syscalls[comm] += 1

    def print_totals(self) -> None:
        """Print summary table."""
        print("\nfailed syscalls by comm:\n")
        print(f"{'comm':<20s}  {'# errors':>10s}")
        print(f"{'-'*20}  {'-'*10}")

        for comm, val in sorted(self.failed_syscalls.items(),
                                key=lambda kv: (-kv[1], kv[0])):
            print(f"{comm:<20s}  {val:10d}")

    def run(self, input_file: str) -> None:
        """Run the session."""
        self.session = perf.session(perf.data(input_file), sample=self.process_event)
        try:
            self.session.process_events()
        finally:
            # Break the reference cycle between perf.session and self.process_event
            # because perf.session lacks cyclic GC support (tp_traverse).
            self.session = None
        self.print_totals()

def main() -> None:
    """Main function."""
    parser = argparse.ArgumentParser(description="Trace failed syscalls")
    parser.add_argument("comm", nargs="?", help="Filter by command name")
    parser.add_argument("-i", "--input", default="perf.data", help="Input file")
    args = parser.parse_args()

    analyzer = FailedSyscalls(args.comm)
    try:
        analyzer.run(args.input)
    except IOError as e:
        print(e, file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()