path: root/tools/perf/python/task-analyzer.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# task-analyzer.py - comprehensive perf tasks analysis
# Copyright (c) 2022, Hagen Paul Pfeifer <hagen@jauu.net>
# Licensed under the terms of the GNU GPL License version 2
#
# Usage:
#
#     perf record -e sched:sched_switch -a -- sleep 10
#     ./task-analyzer.py
#
"""Comprehensive perf tasks analysis."""
from __future__ import annotations

import argparse
from contextlib import contextmanager
import decimal

from typing import List, Dict, Union

def _median(numbers: List[decimal.Decimal]) -> decimal.Decimal:
    """phython3 hat statistics module - we have nothing"""
    n = len(numbers)
    index = n // 2
    if n % 2:
        return sorted(numbers)[index]
    return sum(sorted(numbers)[index - 1 : index + 1]) / decimal.Decimal(2)

def _mean(numbers: List[decimal.Decimal]) -> decimal.Decimal:
    return sum(numbers) / decimal.Decimal(len(numbers))

import os
import string
import sys
from typing import Any, Optional
import perf


# Columns will have a static size to align everything properly
# Support of 116 days of active update with nano precision
LEN_SWITCHED_IN = len("9999999.999999999")
LEN_SWITCHED_OUT = len("9999999.999999999")
LEN_CPU = len("000")
LEN_PID = len("maxvalue")
LEN_TID = len("maxvalue")
LEN_COMM = len("max-comms-length")
LEN_RUNTIME = len("999999.999")
# Support of 3.45 hours of timespans
LEN_OUT_IN = len("99999999999.999")
LEN_OUT_OUT = len("99999999999.999")
LEN_IN_IN = len("99999999999.999")
LEN_IN_OUT = len("99999999999.999")

class Timespans:
    """Tracks elapsed time between occurrences of the same task."""
    def __init__(self, args: argparse.Namespace, time_unit: str) -> None:
        self.args = args
        self.time_unit = time_unit
        self._last_start: Optional[decimal.Decimal] = None
        self._last_finish: Optional[decimal.Decimal] = None
        self.current = {
            'out_out': decimal.Decimal(-1),
            'in_out': decimal.Decimal(-1),
            'out_in': decimal.Decimal(-1),
            'in_in': decimal.Decimal(-1)
        }
        if args.summary_extended:
            self._time_in: decimal.Decimal = decimal.Decimal(-1)
            self.max_vals = {
                'out_in': decimal.Decimal(-1),
                'at': decimal.Decimal(-1),
                'in_out': decimal.Decimal(-1),
                'in_in': decimal.Decimal(-1),
                'out_out': decimal.Decimal(-1)
            }

    def feed(self, task: 'Task') -> None:
        """Calculate timespans from chronological task occurrences."""
        if not self._last_finish:
            self._last_start = task.time_in(self.time_unit)
            self._last_finish = task.time_out(self.time_unit)
            return
        assert self._last_start is not None
        assert self._last_finish is not None
        self._time_in = task.time_in()
        time_in = task.time_in(self.time_unit)
        time_out = task.time_out(self.time_unit)
        self.current['in_in'] = time_in - self._last_start
        self.current['out_in'] = time_in - self._last_finish
        self.current['in_out'] = time_out - self._last_start
        self.current['out_out'] = time_out - self._last_finish
        if self.args.summary_extended:
            self.update_max_entries()
        self._last_finish = task.time_out(self.time_unit)
        self._last_start = task.time_in(self.time_unit)

    def update_max_entries(self) -> None:
        """Update maximum timespans."""
        self.max_vals['in_in'] = max(self.max_vals['in_in'], self.current['in_in'])
        self.max_vals['out_out'] = max(self.max_vals['out_out'], self.current['out_out'])
        self.max_vals['in_out'] = max(self.max_vals['in_out'], self.current['in_out'])
        if self.current['out_in'] > self.max_vals['out_in']:
            self.max_vals['out_in'] = self.current['out_in']
            self.max_vals['at'] = self._time_in

class Task:
    """Handles information of a given task."""
    def __init__(self, task_id: str, tid: int, cpu: int, comm: str) -> None:
        self.id = task_id
        self.tid = tid
        self.cpu = cpu
        self.comm = comm
        self.pid: Optional[int] = None
        self._time_in: Optional[decimal.Decimal] = None
        self._time_out: Optional[decimal.Decimal] = None

    def schedule_in_at(self, time_ns: int) -> None:
        """Set schedule in time."""
        self._time_in = decimal.Decimal(time_ns) / decimal.Decimal(1e9)

    def schedule_out_at(self, time_ns: int) -> None:
        """Set schedule out time."""
        self._time_out = decimal.Decimal(time_ns) / decimal.Decimal(1e9)

    def time_out(self, unit: str = "s") -> decimal.Decimal:
        """Return schedule out time."""
        factor = TaskAnalyzer.time_uniter(unit)
        return self._time_out * decimal.Decimal(factor) if self._time_out else decimal.Decimal(0)

    def time_in(self, unit: str = "s") -> decimal.Decimal:
        """Return schedule in time."""
        factor = TaskAnalyzer.time_uniter(unit)
        return self._time_in * decimal.Decimal(factor) if self._time_in else decimal.Decimal(0)

    def runtime(self, unit: str = "us") -> decimal.Decimal:
        """Return runtime."""
        factor = TaskAnalyzer.time_uniter(unit)
        if self._time_out is not None and self._time_in is not None:
            return (self._time_out - self._time_in) * decimal.Decimal(factor)
        return decimal.Decimal(0)

    def update_pid(self, pid: int) -> None:
        """Update PID."""
        self.pid = pid

class Summary:
    """
    Primary instance for calculating the summary output. Processes the whole trace to
    find and memorize relevant data such as mean, max et cetera. This instance handles
    dynamic alignment aspects for summary output.
    """

    def __init__(self, analyzer):
        self.analyzer = analyzer
        self.args = analyzer.args
        self.db = analyzer.db
        self.time_unit = analyzer.time_unit
        self.fd_sum = analyzer.fd_sum
        self._body = []

    class AlignmentHelper:
        """
        Used to calculated the alignment for the output of the summary.
        """
        def __init__(self, pid, tid, comm, runs, acc, mean,
                    median, min_val, max_val, max_at):
            self.pid = pid
            self.tid = tid
            self.comm = comm
            self.runs = runs
            self.acc = acc
            self.mean = mean
            self.median = median
            self.min = min_val
            self.max = max_val
            self.max_at = max_at
            self.out_in = None
            self.inter_at = None
            self.out_out = None
            self.in_in = None
            self.in_out = None

    def _print_header(self):
        '''
        Output is aligned using the formatted column widths in self.db.
        '''
        len_tasks = max(
            len("Task Information"),
            sum(self.db["task_info"].values()) + len(self.db["task_info"]) - 1,
        )
        fmt = "{{:^{}}}".format(len_tasks)
        fmt += " {{:^{}}}".format(
            sum(self.db["runtime_info"].values()) + len(self.db["runtime_info"]) - 1
        )
        _header = ("Task Information", "Runtime Information")

        if self.args.summary_extended:
            fmt += " {{:^{}}}".format(
                sum(self.db["inter_times"].values()) + len(self.db["inter_times"]) - 1
            )
            _header += ("Max Inter Task Times",)
        self.fd_sum.write(fmt.format(*_header) + "\n")

    def _column_titles(self):
        """
        Cells are being processed and displayed in different way so an alignment adjust
        is implemented depeding on the choice of the timeunit. The positions of the max
        values are being displayed in grey. Thus in their format two additional {},
        are placed for color set and reset.
        """
        separator, fix_csv_align = self.analyzer.prepare_fmt_sep(is_summary=True)
        fmt = "{{:>{}}}".format(self.db["task_info"]["pid"] * fix_csv_align)
        fmt += "{}{{:>{}}}".format(separator, self.db["task_info"]["tid"] * fix_csv_align)
        fmt += "{}{{:>{}}}".format(separator, self.db["task_info"]["comm"] * fix_csv_align)
        fmt += "{}{{:>{}}}".format(separator, self.db["runtime_info"]["runs"] * fix_csv_align)
        fmt += "{}{{:>{}}}".format(separator, self.db["runtime_info"]["acc"] * fix_csv_align)
        fmt += "{}{{:>{}}}".format(separator, self.db["runtime_info"]["mean"] * fix_csv_align)
        fmt += "{}{{:>{}}}".format(
            separator, self.db["runtime_info"]["median"] * fix_csv_align
        )
        fmt += "{}{{:>{}}}".format(
            separator, self.db["runtime_info"]["min"] * fix_csv_align
        )
        fmt += "{}{{:>{}}}".format(
            separator, self.db["runtime_info"]["max"] * fix_csv_align
        )
        fmt += "{}{{}}{{:>{}}}{{}}".format(
            separator, self.db["runtime_info"]["max_at"] * fix_csv_align
        )

        grey = "" if self.args.csv_summary else TaskAnalyzer.COLORS["grey"]
        reset = "" if self.args.csv_summary else TaskAnalyzer.COLORS["reset"]
        column_titles = ("PID", "TID", "Comm")
        column_titles += ("Runs", "Accumulated", "Mean", "Median", "Min", "Max")
        column_titles += (grey, "Max At", reset)

        if self.args.summary_extended:
            fmt += "{}{{:>{}}}".format(
                separator,
                self.db["inter_times"]["out_in"] * fix_csv_align
            )
            fmt += "{}{{}}{{:>{}}}{{}}".format(
                separator,
                self.db["inter_times"]["inter_at"] * fix_csv_align
            )
            fmt += "{}{{:>{}}}".format(
                separator,
                self.db["inter_times"]["out_out"] * fix_csv_align
            )
            fmt += "{}{{:>{}}}".format(
                separator,
                self.db["inter_times"]["in_in"] * fix_csv_align
            )
            fmt += "{}{{:>{}}}".format(
                separator,
                self.db["inter_times"]["in_out"] * fix_csv_align
            )

            column_titles += (
                "Out-In", grey, "Max At",
                reset, "Out-Out", "In-In", "In-Out"
            )

        self.fd_sum.write(fmt.format(*column_titles) + "\n")

    def _task_stats(self):
        """calculates the stats of every task and constructs the printable summary"""
        grey = "" if self.args.csv_summary else TaskAnalyzer.COLORS["grey"]
        reset = "" if self.args.csv_summary else TaskAnalyzer.COLORS["reset"]
        for tid in sorted(self.db["tid"]):
            color_one_sample = grey
            color_reset = reset
            no_executed = 0
            runtimes = []
            time_in = []
            timespans = Timespans(self.args, self.time_unit)
            for task in self.db["tid"][tid]:
                pid = task.pid
                comm = task.comm
                no_executed += 1
                runtimes.append(task.runtime(self.time_unit))
                time_in.append(task.time_in())
                timespans.feed(task)
            if len(runtimes) > 1:
                color_one_sample = ""
                color_reset = ""
            time_max = max(runtimes)
            time_min = min(runtimes)
            max_at = time_in[runtimes.index(max(runtimes))]

            # The size of the decimal after sum,mean and median varies, thus we cut
            # the decimal number, by rounding it. It has no impact on the output,
            # because we have a precision of the decimal points at the output.
            time_sum = round(sum(runtimes), 3)
            time_mean = round(_mean(runtimes), 3)
            time_median = round(_median(runtimes), 3)

            align_helper = self.AlignmentHelper(pid, tid, comm, no_executed, time_sum,
                                    time_mean, time_median, time_min, time_max, max_at)
            self._body.append([
                pid, tid, comm, no_executed, time_sum, color_one_sample,
                time_mean, time_median, time_min, time_max,
                grey, max_at,
                reset, color_reset
            ])
            if self.args.summary_extended:
                self._body[-1].extend([timespans.max_vals['out_in'],
                                grey, timespans.max_vals['at'],
                                reset, timespans.max_vals['out_out'],
                                timespans.max_vals['in_in'],
                                timespans.max_vals['in_out']])
                align_helper.out_in = timespans.max_vals['out_in']
                align_helper.inter_at = timespans.max_vals['at']
                align_helper.out_out = timespans.max_vals['out_out']
                align_helper.in_in = timespans.max_vals['in_in']
                align_helper.in_out = timespans.max_vals['in_out']
            self._calc_alignments_summary(align_helper)

    def _format_stats(self):
        separator, fix_csv_align = self.analyzer.prepare_fmt_sep(is_summary=True)
        decimal_precision, time_precision = self.analyzer.prepare_fmt_precision()
        len_pid = self.db["task_info"]["pid"] * fix_csv_align
        len_tid = self.db["task_info"]["tid"] * fix_csv_align
        len_comm = self.db["task_info"]["comm"] * fix_csv_align
        len_runs = self.db["runtime_info"]["runs"] * fix_csv_align
        len_acc = self.db["runtime_info"]["acc"] * fix_csv_align
        len_mean = self.db["runtime_info"]["mean"] * fix_csv_align
        len_median = self.db["runtime_info"]["median"] * fix_csv_align
        len_min = self.db["runtime_info"]["min"] * fix_csv_align
        len_max = self.db["runtime_info"]["max"] * fix_csv_align
        len_max_at = self.db["runtime_info"]["max_at"] * fix_csv_align
        if self.args.summary_extended:
            len_out_in = self.db["inter_times"]["out_in"] * fix_csv_align
            len_inter_at = self.db["inter_times"]["inter_at"] * fix_csv_align
            len_out_out = self.db["inter_times"]["out_out"] * fix_csv_align
            len_in_in = self.db["inter_times"]["in_in"] * fix_csv_align
            len_in_out = self.db["inter_times"]["in_out"] * fix_csv_align

        fmt = "{{:{}d}}".format(len_pid)
        fmt += "{}{{:{}d}}".format(separator, len_tid)
        fmt += "{}{{:>{}}}".format(separator, len_comm)
        fmt += "{}{{:{}d}}".format(separator, len_runs)
        fmt += "{}{{:{}.{}f}}".format(separator, len_acc, time_precision)
        fmt += "{}{{}}{{:{}.{}f}}".format(separator, len_mean, time_precision)
        fmt += "{}{{:{}.{}f}}".format(separator, len_median, time_precision)
        fmt += "{}{{:{}.{}f}}".format(separator, len_min, time_precision)
        fmt += "{}{{:{}.{}f}}".format(separator, len_max, time_precision)
        fmt += "{}{{}}{{:{}.{}f}}{{}}{{}}".format(
            separator, len_max_at, decimal_precision
        )
        if self.args.summary_extended:
            fmt += "{}{{:{}.{}f}}".format(separator, len_out_in, time_precision)
            fmt += "{}{{}}{{:{}.{}f}}{{}}".format(
                separator, len_inter_at, decimal_precision
            )
            fmt += "{}{{:{}.{}f}}".format(separator, len_out_out, time_precision)
            fmt += "{}{{:{}.{}f}}".format(separator, len_in_in, time_precision)
            fmt += "{}{{:{}.{}f}}".format(separator, len_in_out, time_precision)
        return fmt

    def _calc_alignments_summary(self, align_helper):
        # Measure the formatted string widths (e.g. f"{val:.{decimal_precision}f}"
        # and f"{val:.{time_precision}f}") rather than str(val), because Decimal's
        # default string representation may include unrounded fractional digits or
        # omit trailing zeros that _format_stats() renders with fixed precision.
        decimal_precision, time_precision = self.analyzer.prepare_fmt_precision()
        for key in self.db["task_info"]:
            val_len = len(str(getattr(align_helper, key)))
            if val_len > self.db["task_info"][key]:
                self.db["task_info"][key] = val_len
        for key in self.db["runtime_info"]:
            val = getattr(align_helper, key)
            if key == "runs":
                val_len = len(str(val))
            elif key == "max_at":
                val_len = len(f"{val:.{decimal_precision}f}")
            else:
                val_len = len(f"{val:.{time_precision}f}")
            if val_len > self.db["runtime_info"][key]:
                self.db["runtime_info"][key] = val_len
        if self.args.summary_extended:
            for key in self.db["inter_times"]:
                val = getattr(align_helper, key)
                if key == "inter_at":
                    val_len = len(f"{val:.{decimal_precision}f}")
                else:
                    val_len = len(f"{val:.{time_precision}f}")
                if val_len > self.db["inter_times"][key]:
                    self.db["inter_times"][key] = val_len

    def print(self):
        self._task_stats()
        fmt = self._format_stats()

        if not self.args.csv_summary:
            print("\nSummary", file=self.fd_sum)
            self._print_header()
        self._column_titles()
        for i in range(len(self._body)):
            self.fd_sum.write(fmt.format(*tuple(self._body[i])) + "\n")


class TaskAnalyzer:

    """Main class for task analysis."""

    COLORS = {
        "grey": "\033[90m",
        "red": "\033[91m",
        "green": "\033[92m",
        "yellow": "\033[93m",
        "blue": "\033[94m",
        "violet": "\033[95m",
        "reset": "\033[0m",
    }

    def __init__(self, args: argparse.Namespace) -> None:
        self.args = args
        self.db: Dict[str, Any] = {}
        self.session: Optional[perf.session] = None
        self._tgid_cache: Dict[int, int] = {}
        self.time_unit = "us"
        if args.ns:
            self.time_unit = "ns"
        elif args.ms:
            self.time_unit = "ms"
        self._init_db()
        self._check_color()
        self.fd_task = sys.stdout
        self.fd_sum = sys.stdout

    @contextmanager
    def open_output(self, filename: str, default: Any):
        """Context manager for file or stdout."""
        if filename:
            with open(filename, "w", encoding="utf-8") as f:
                yield f
        else:
            yield default

    def _init_db(self) -> None:
        self.db["running"] = {}
        self.db["tid"] = {}
        self.db["global"] = []
        if (self.args.summary or self.args.summary_extended or
                self.args.summary_only or self.args.csv_summary):
            self.db["task_info"] = {}
            self.db["runtime_info"] = {}
            self.db["task_info"]["pid"] = len("PID")
            self.db["task_info"]["tid"] = len("TID")
            self.db["task_info"]["comm"] = len("Comm")
            self.db["runtime_info"]["runs"] = len("Runs")
            self.db["runtime_info"]["acc"] = len("Accumulated")
            self.db["runtime_info"]["max"] = len("Max")
            self.db["runtime_info"]["max_at"] = len("Max At")
            self.db["runtime_info"]["min"] = len("Min")
            self.db["runtime_info"]["mean"] = len("Mean")
            self.db["runtime_info"]["median"] = len("Median")
            if self.args.summary_extended:
                self.db["inter_times"] = {}
                self.db["inter_times"]["out_in"] = len("Out-In")
                self.db["inter_times"]["inter_at"] = len("Max At")
                self.db["inter_times"]["out_out"] = len("Out-Out")
                self.db["inter_times"]["in_in"] = len("In-In")
                self.db["inter_times"]["in_out"] = len("In-Out")

    def _check_color(self) -> None:
        """Check if color should be enabled."""
        if self.args.csv:
            TaskAnalyzer.COLORS = {k: "" for k in TaskAnalyzer.COLORS}
            return
        if sys.stdout.isatty() and self.args.stdio_color != "never":
            return
        if self.args.stdio_color == "always":
            return
        TaskAnalyzer.COLORS = {k: "" for k in TaskAnalyzer.COLORS}

    @staticmethod
    def time_uniter(unit: str) -> float:
        """Return time unit factor."""
        picker = {"s": 1, "ms": 1e3, "us": 1e6, "ns": 1e9}
        return picker[unit]

    def _task_id(self, pid: int, cpu: int) -> str:
        return f"{pid}-{cpu}"

    def _filter_non_printable(self, unfiltered: Union[str, bytearray]) -> str:
        # Strip non-printable characters, whitespace control characters
        # (\r\n\t\x0b\x0c), quotes (", '), and ';' because --csv and
        # --csv-summary emit unquoted ';'-delimited columns. Also replace a
        # leading '=', '+', '-', or '@' with '_' to mitigate CSV formula / DDE
        # injection attacks if an attacker-controlled process comm is opened in
        # a spreadsheet program.
        if isinstance(unfiltered, (bytearray, bytes)):
            unfiltered = unfiltered.decode('utf-8', 'ignore')
        filtered = ""
        for char in unfiltered:
            if char in string.printable and char not in "\r\n\t\x0b\x0c;\"'":
                filtered += char
        stripped = filtered.lstrip()
        if stripped and stripped[0] in "=+-@":
            filtered = "_" + stripped[1:]
        return filtered

    def prepare_fmt_precision(self) -> tuple[int, int]:
        if self.args.ns:
            return 9, 0
        return 6, 3

    def prepare_fmt_sep(self, is_summary: bool = False) -> tuple[str, int]:
        # fix_csv_align multiplies format field widths: 0 in CSV mode collapses
        # {:>N} padding to {:>0} (no extra spaces around ';'), while 1 in
        # standard text mode preserves fixed-width column alignment.
        if (is_summary and self.args.csv_summary) or (not is_summary and self.args.csv):
            return ";", 0
        return " ", 1

    def _fmt_header(self) -> str:
        separator, fix_csv_align = self.prepare_fmt_sep()
        fmt = f"{{:>{LEN_SWITCHED_IN*fix_csv_align}}}"
        fmt += f"{separator}{{:>{LEN_SWITCHED_OUT*fix_csv_align}}}"
        fmt += f"{separator}{{:>{LEN_CPU*fix_csv_align}}}"
        fmt += f"{separator}{{:>{LEN_PID*fix_csv_align}}}"
        fmt += f"{separator}{{:>{LEN_TID*fix_csv_align}}}"
        fmt += f"{separator}{{:>{LEN_COMM*fix_csv_align}}}"
        fmt += f"{separator}{{:>{LEN_RUNTIME*fix_csv_align}}}"
        fmt += f"{separator}{{:>{LEN_OUT_IN*fix_csv_align}}}"
        if self.args.extended_times:
            fmt += f"{separator}{{:>{LEN_OUT_OUT*fix_csv_align}}}"
            fmt += f"{separator}{{:>{LEN_IN_IN*fix_csv_align}}}"
            fmt += f"{separator}{{:>{LEN_IN_OUT*fix_csv_align}}}"
        return fmt

    def _fmt_body(self) -> str:
        separator, fix_csv_align = self.prepare_fmt_sep()
        decimal_precision, time_precision = self.prepare_fmt_precision()
        fmt = f"{{}}{{:{LEN_SWITCHED_IN*fix_csv_align}.{decimal_precision}f}}"
        fmt += f"{separator}{{:{LEN_SWITCHED_OUT*fix_csv_align}.{decimal_precision}f}}"
        fmt += f"{separator}{{:{LEN_CPU*fix_csv_align}d}}"
        fmt += f"{separator}{{:{LEN_PID*fix_csv_align}d}}"
        fmt += f"{separator}{{}}{{:{LEN_TID*fix_csv_align}d}}{{}}"
        fmt += f"{separator}{{}}{{:>{LEN_COMM*fix_csv_align}}}"
        fmt += f"{separator}{{:{LEN_RUNTIME*fix_csv_align}.{time_precision}f}}"
        if self.args.extended_times:
            fmt += f"{separator}{{:{LEN_OUT_IN*fix_csv_align}.{time_precision}f}}"
            fmt += f"{separator}{{:{LEN_OUT_OUT*fix_csv_align}.{time_precision}f}}"
            fmt += f"{separator}{{:{LEN_IN_IN*fix_csv_align}.{time_precision}f}}"
            fmt += f"{separator}{{:{LEN_IN_OUT*fix_csv_align}.{time_precision}f}}{{}}"
        else:
            fmt += f"{separator}{{:{LEN_OUT_IN*fix_csv_align}.{time_precision}f}}{{}}"
        return fmt

    def _print_header(self) -> None:
        fmt = self._fmt_header()
        header = ["Switched-In", "Switched-Out", "CPU", "PID", "TID", "Comm",
                  "Runtime", "Time Out-In"]
        if self.args.extended_times:
            header += ["Time Out-Out", "Time In-In", "Time In-Out"]
        self.fd_task.write(fmt.format(*header) + "\n")

    def _print_task_finish(self, task: Task) -> None:
        c_row_set = ""
        c_row_reset = ""
        out_in: Any = -1
        out_out: Any = -1
        in_in: Any = -1
        in_out: Any = -1
        fmt = self._fmt_body()

        if str(task.tid) in self.args.highlight_tasks_map:
            c_row_set = TaskAnalyzer.COLORS.get(self.args.highlight_tasks_map[str(task.tid)], '')
            c_row_reset = TaskAnalyzer.COLORS["reset"]
        if task.comm in self.args.highlight_tasks_map:
            c_row_set = TaskAnalyzer.COLORS.get(self.args.highlight_tasks_map[task.comm], '')
            c_row_reset = TaskAnalyzer.COLORS["reset"]

        c_tid_set = ""
        c_tid_reset = ""
        if task.pid == task.tid:
            c_tid_set = TaskAnalyzer.COLORS["grey"]
            c_tid_reset = TaskAnalyzer.COLORS["reset"]

        if task.tid in self.db["tid"]:
            last_tid_task = self.db["tid"][task.tid][-1]
            timespan_gap_tid = Timespans(self.args, self.time_unit)
            timespan_gap_tid.feed(last_tid_task)
            timespan_gap_tid.feed(task)
            out_in = timespan_gap_tid.current['out_in']
            out_out = timespan_gap_tid.current['out_out']
            in_in = timespan_gap_tid.current['in_in']
            in_out = timespan_gap_tid.current['in_out']

        if self.args.extended_times:
            line_out = fmt.format(c_row_set, task.time_in(), task.time_out(), task.cpu,
                            task.pid, c_tid_set, task.tid, c_tid_reset, c_row_set, task.comm,
                            task.runtime(self.time_unit), out_in, out_out, in_in, in_out,
                            c_row_reset) + "\n"
        else:
            line_out = fmt.format(c_row_set, task.time_in(), task.time_out(), task.cpu,
                            task.pid, c_tid_set, task.tid, c_tid_reset, c_row_set, task.comm,
                            task.runtime(self.time_unit), out_in, c_row_reset) + "\n"
        self.fd_task.write(line_out)

    def _record_cleanup(self, _list: list[Any]) -> list[Any]:
        need_summary = (self.args.summary or self.args.summary_extended or
                        self.args.summary_only or self.args.csv_summary)
        if not need_summary and len(_list) > 1:
            return _list[len(_list) - 1:]
        return _list

    def _record_by_tid(self, task: Task) -> None:
        tid = task.tid
        if tid not in self.db["tid"]:
            self.db["tid"][tid] = []
        self.db["tid"][tid].append(task)
        self.db["tid"][tid] = self._record_cleanup(self.db["tid"][tid])

    def _record_global(self, task: Task) -> None:
        self.db["global"].append(task)
        self.db["global"] = self._record_cleanup(self.db["global"])

    def _handle_task_finish(self, tid: int, cpu: int, time_ns: int, pid: int) -> None:
        if tid == 0 and (not self.args.tid_renames or 0 not in self.args.tid_renames):
            return
        _id = self._task_id(tid, cpu)
        if _id not in self.db["running"]:
            return
        task = self.db["running"][_id]
        task.schedule_out_at(time_ns)
        task.update_pid(pid)
        del self.db["running"][_id]

        if not self._limit_filtered(tid, pid, task.comm):
            if not self.args.summary_only:
                self._print_task_finish(task)
            self._record_by_tid(task)
            self._record_global(task)

    def _handle_task_start(self, tid: int, cpu: int, comm: str, time_ns: int) -> None:
        if tid == 0 and (not self.args.tid_renames or 0 not in self.args.tid_renames):
            return
        if tid in self.args.tid_renames:
            comm = self._filter_non_printable(self.args.tid_renames[tid])
        _id = self._task_id(tid, cpu)
        if _id in self.db["running"]:
            return
        task = Task(_id, tid, cpu, comm)
        task.schedule_in_at(time_ns)
        self.db["running"][_id] = task

    def _limit_filtered(self, tid: int, pid: int, comm: str) -> bool:
        """Filter tasks based on CLI arguments."""
        match_filter = False
        if self.args.filter_tasks:
            if (str(tid) in self.args.filter_tasks or
                str(pid) in self.args.filter_tasks or
                comm in self.args.filter_tasks):
                match_filter = True

        match_limit = False
        if self.args.limit_to_tasks:
            if (str(tid) in self.args.limit_to_tasks or
                str(pid) in self.args.limit_to_tasks or
                comm in self.args.limit_to_tasks):
                match_limit = True

        if self.args.filter_tasks and match_filter:
            return True
        if self.args.limit_to_tasks and not match_limit:
            return True
        return False

    def _is_within_timelimit(self, time_ns: int) -> bool:
        if not self.args.time_limit:
            return True
        time_s = decimal.Decimal(time_ns) / decimal.Decimal(1e9)
        bounds = self.args.time_limit.split(":")
        lower_bound = bounds[0] if len(bounds) > 0 else ""
        upper_bound = bounds[1] if len(bounds) > 1 else ""
        if lower_bound and time_s < decimal.Decimal(lower_bound):
            return False
        if upper_bound and time_s > decimal.Decimal(upper_bound):
            return False
        return True

    def process_event(self, sample: perf.sample_event) -> None:
        """Process sched:sched_switch events."""
        if "sched:sched_switch" not in str(sample.evsel):
            return

        time_ns = sample.sample_time
        if not self._is_within_timelimit(time_ns):
            return

        # Access tracepoint fields directly from sample object
        try:
            prev_pid = sample.prev_pid
            next_pid = sample.next_pid
            next_comm = sample.next_comm
            common_cpu = sample.sample_cpu
        except AttributeError:
            if not self.db.get("warned_missing_fields"):
                self.db["warned_missing_fields"] = True
                print("Warning: sched:sched_switch sample missing tracepoint fields "
                      "(is libtraceevent support enabled?).", file=sys.stderr)
            return

        next_comm = self._filter_non_printable(next_comm)

        # Task finish for previous task. Resolve TGID (process ID) from TID:
        # in file mode we query perf.session.find_thread() (checking thread.pid > 0
        # so an unresolved -1 does not overwrite the prev_pid fallback), whereas in
        # live mode (self.session is None) we read /proc/<tid>/status and cache
        # the lookup in _tgid_cache.
        prev_tgid = prev_pid  # Fallback
        if prev_pid != 0:
            if self.session:
                try:
                    thread = self.session.find_thread(-1, prev_pid)
                    if thread and thread.pid > 0:
                        prev_tgid = thread.pid
                except (OSError, ValueError, KeyError, RuntimeError, TypeError, AttributeError):
                    pass
            elif prev_pid in self._tgid_cache:
                prev_tgid = self._tgid_cache[prev_pid]
            else:
                if len(self._tgid_cache) >= 4096:
                    self._tgid_cache.pop(next(iter(self._tgid_cache)))
                self._tgid_cache[prev_pid] = prev_tgid
                try:
                    with open(f"/proc/{prev_pid}/status", encoding="utf-8") as f:
                        for line in f:
                            if line.startswith("Tgid:"):
                                prev_tgid = int(line.split()[1])
                                self._tgid_cache[prev_pid] = prev_tgid
                                break
                except (OSError, ValueError, KeyError, RuntimeError):
                    pass
        self._handle_task_finish(prev_pid, common_cpu, time_ns, prev_tgid)
        # Task start for next task
        self._handle_task_start(next_pid, common_cpu, next_comm, time_ns)

    def print_summary(self) -> None:
        """Calculate and print summary."""
        need_summary = (self.args.summary or self.args.summary_extended or
                        self.args.summary_only or self.args.csv_summary)
        if not need_summary:
            return

        Summary(self).print()

    def _run_file(self) -> None:
        if not self.args.summary_only:
            self._print_header()

        session = perf.session(perf.data(self.args.input), sample=self.process_event)
        self.session = session
        try:
            session.process_events()
        except KeyboardInterrupt:
            pass
        finally:
            # Break the reference cycle between self.session and the bound
            # self.process_event callback so the C perf.session object is freed.
            self.session = None

        if not self.db["global"]:
            print(f"Warning: No sched:sched_switch trace events found in '{self.args.input}'.",
                  file=sys.stderr)

        self.print_summary()

    def _run_live(self) -> None:
        if not self.args.summary_only:
            self._print_header()

        cpus = perf.cpu_map()
        threads = perf.thread_map(-1)
        evlist = perf.parse_events("sched:sched_switch", cpus, threads)
        evlist.config()

        evlist.open()
        evlist.mmap()
        evlist.enable()

        # Per-CPU perf ring buffers are drained sequentially in a loop, so events
        # read from CPU N may have timestamps slightly earlier than events just
        # read from CPU 0. Buffer events in pending_events and only dispatch those
        # older than a 50 ms watermark behind max_ts so cross-CPU migrations and
        # switches are processed in global timestamp order.
        pending_events: list[perf.sample_event] = []
        print("Live mode started. Press Ctrl+C to stop.", file=sys.stderr)
        try:
            while True:
                try:
                    evlist.poll(timeout=100)
                except InterruptedError:
                    continue
                for cpu in cpus:
                    while True:
                        event = evlist.read_on_cpu(cpu)
                        if not event:
                            break
                        if not isinstance(event, perf.sample_event):
                            continue
                        pending_events.append(event)
                if pending_events:
                    pending_events.sort(
                        key=lambda e: getattr(e, 'sample_time', getattr(e, 'time', 0))
                    )
                    max_ts = getattr(
                        pending_events[-1], 'sample_time', getattr(pending_events[-1], 'time', 0)
                    )
                    cutoff_ts = max_ts - 50_000_000
                    ready_idx = 0
                    for event in pending_events:
                        ev_ts = getattr(event, 'sample_time', getattr(event, 'time', 0))
                        if ev_ts <= cutoff_ts:
                            self.process_event(event)
                            ready_idx += 1
                        else:
                            break
                    del pending_events[:ready_idx]
        except KeyboardInterrupt:
            print("\nStopping live mode...", file=sys.stderr)
        finally:
            pending_events.sort(key=lambda e: getattr(e, 'sample_time', getattr(e, 'time', 0)))
            for event in pending_events:
                self.process_event(event)
            evlist.close()
            self.print_summary()

    def run(self) -> None:
        """Run the session."""
        is_live = (self.args.live or
                   (not os.path.exists(self.args.input) and self.args.input == "perf.data"))
        with self.open_output(self.args.csv, sys.stdout) as fd_task:
            if self.args.csv and self.args.csv == self.args.csv_summary:
                self.fd_task = fd_task
                self.fd_sum = fd_task
                if is_live:
                    self._run_live()
                else:
                    self._run_file()
            else:
                with self.open_output(self.args.csv_summary, sys.stdout) as fd_sum:
                    self.fd_task = fd_task
                    self.fd_sum = fd_sum
                    if is_live:
                        self._run_live()
                    else:
                        self._run_file()

def main() -> None:
    """Main function."""
    parser = argparse.ArgumentParser(description="Analyze tasks behavior")
    parser.add_argument("-i", "--input", default="perf.data", help="Input file name")
    parser.add_argument("--time-limit", default="", help="print tasks only in time window")
    parser.add_argument("--summary", action="store_true",
                        help="print additional runtime information")
    parser.add_argument("--summary-only", action="store_true",
                        help="print only summary without traces")
    parser.add_argument("--summary-extended", action="store_true",
                        help="print extended summary")
    parser.add_argument("--ns", action="store_true", help="show timestamps in nanoseconds")
    parser.add_argument("--ms", action="store_true", help="show timestamps in milliseconds")
    parser.add_argument("--live", action="store_true",
                        help="force live mode (ignores -i/perf.data)")
    parser.add_argument("--extended-times", action="store_true",
                        help="Show elapsed times between schedule in/out")
    parser.add_argument("--filter-tasks", default="", help="filter tasks by tid, pid or comm")
    parser.add_argument("--limit-to-tasks", default="", help="limit output to selected tasks")
    parser.add_argument("--highlight-tasks", default="", help="colorize special tasks")
    parser.add_argument("--rename-comms-by-tids", default="", help="rename task names by using tid")
    parser.add_argument("--stdio-color", default="auto", choices=["always", "never", "auto"],
                        help="configure color output")
    parser.add_argument("--csv", default="", help="Write trace to file")
    parser.add_argument("--csv-summary", default="", help="Write summary to file")

    args = parser.parse_args()
    args.tid_renames = {}
    args.highlight_tasks_map = {}
    args.filter_tasks = args.filter_tasks.split(",") if args.filter_tasks else []
    args.limit_to_tasks = args.limit_to_tasks.split(",") if args.limit_to_tasks else []

    if args.rename_comms_by_tids:
        for item in args.rename_comms_by_tids.split(","):
            try:
                tid, name = item.split(":", 1)
                args.tid_renames[int(tid)] = name
            except ValueError:
                print(f"Error: Invalid format for --rename-comms-by-tids '{item}', "
                      "expected tid:name", file=sys.stderr)
                sys.exit(1)

    if args.highlight_tasks:
        for item in args.highlight_tasks.split(","):
            parts = item.split(":")
            if len(parts) == 1:
                parts.append("red")
            key, color = parts[0], parts[1]
            args.highlight_tasks_map[key] = color

    analyzer = TaskAnalyzer(args)
    analyzer.run()

if __name__ == "__main__":
    main()