linux/tools/perf/python/sched-migration.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
"""
Cpu task migration overview toy

Copyright (C) 2010 Frederic Weisbecker <fweisbec@gmail.com>
Ported to modern directory structure and refactored to use class.
"""
from __future__ import annotations

import argparse
from collections import defaultdict, UserList
import importlib
import sys
from typing import Any
import perf


# Global threads dictionary
threads = defaultdict(lambda: "unknown")
threads[0] = "idle"


def thread_name(pid: int) -> str:
    """Return thread name formatted with pid."""
    return f"{threads[pid]}:{pid}"


def task_state(state: int) -> str:
    """Map task state integer to string."""
    states = {
        0: "R",
        1: "S",
        2: "D",
        64: "DEAD"
    }
    return states.get(state, "Unknown")


class RunqueueEventUnknown:
    """Unknown runqueue event."""
    @staticmethod
    def color():
        """Return color for event."""
        return None

    def __repr__(self):
        return "unknown"


class RunqueueEventSleep:
    """Sleep runqueue event."""
    @staticmethod
    def color():
        """Return color for event."""
        return 0, 0, 0xff

    def __init__(self, sleeper: int):
        self.sleeper = sleeper

    def __repr__(self):
        return f"{thread_name(self.sleeper)} gone to sleep"


class RunqueueEventWakeup:
    """Wakeup runqueue event."""
    @staticmethod
    def color():
        """Return color for event."""
        return 0xff, 0xff, 0

    def __init__(self, wakee: int):
        self.wakee = wakee

    def __repr__(self):
        return f"{thread_name(self.wakee)} woke up"


class RunqueueEventFork:
    """Fork runqueue event."""
    @staticmethod
    def color():
        """Return color for event."""
        return 0, 0xff, 0

    def __init__(self, child: int):
        self.child = child

    def __repr__(self):
        return f"new forked task {thread_name(self.child)}"


class RunqueueMigrateIn:
    """Migrate in runqueue event."""
    @staticmethod
    def color():
        """Return color for event."""
        return 0, 0xf0, 0xff

    def __init__(self, new: int):
        self.new = new

    def __repr__(self):
        return f"task migrated in {thread_name(self.new)}"


class RunqueueMigrateOut:
    """Migrate out runqueue event."""
    @staticmethod
    def color():
        """Return color for event."""
        return 0xff, 0, 0xff

    def __init__(self, old: int):
        self.old = old

    def __repr__(self):
        return f"task migrated out {thread_name(self.old)}"


class RunqueueSnapshot:
    """Snapshot of runqueue state."""

    def __init__(self, tasks=None, event=None):
        if tasks is None:
            tasks = (0,)
        if event is None:
            event = RunqueueEventUnknown()
        self.tasks = tuple(tasks)
        self.event = event

    def sched_switch(self, prev: int, prev_state: int, next_pid: int):
        """Handle sched switch in snapshot."""
        if task_state(prev_state) == "R" and next_pid in self.tasks \
                and prev in self.tasks:
            return self

        event = (
            RunqueueEventSleep(prev)
            if task_state(prev_state) != "R"
            else RunqueueEventUnknown()
        )

        next_tasks = list(self.tasks[:])
        if prev in self.tasks:
            if task_state(prev_state) != "R":
                next_tasks.remove(prev)
        elif task_state(prev_state) == "R":
            next_tasks.append(prev)

        if next_pid not in next_tasks:
            next_tasks.append(next_pid)

        return RunqueueSnapshot(next_tasks, event)

    def migrate_out(self, old: int):
        """Handle task migrate out in snapshot."""
        if old not in self.tasks:
            return self
        next_tasks = [task for task in self.tasks if task != old]

        return RunqueueSnapshot(next_tasks, RunqueueMigrateOut(old))

    def __migrate_in(self, new: int, event):
        if new in self.tasks:
            return RunqueueSnapshot(self.tasks, event)
        next_tasks = self.tasks + tuple([new])

        return RunqueueSnapshot(next_tasks, event)

    def migrate_in(self, new: int):
        """Handle task migrate in snapshot."""
        return self.__migrate_in(new, RunqueueMigrateIn(new))

    def wake_up(self, new: int):
        """Handle task wakeup in snapshot."""
        return self.__migrate_in(new, RunqueueEventWakeup(new))

    def wake_up_new(self, new: int):
        """Handle task fork in snapshot."""
        return self.__migrate_in(new, RunqueueEventFork(new))

    def load(self) -> int:
        """Provide the number of tasks on the runqueue. Don't count idle"""
        return len(self.tasks) - 1

    def __repr__(self):
        return self.tasks.__repr__()


class TimeSlice:
    """Represents a time slice of execution."""

    def __init__(self, start: int, prev):
        self.start = start
        self.prev = prev
        self.end = start
        # cpus that triggered the event
        self.event_cpus: list[int] = []
        if prev is not None:
            self.total_load = prev.total_load
            self.rqs = prev.rqs.copy()
        else:
            self.rqs = defaultdict(RunqueueSnapshot)
            self.total_load = 0

    def __update_total_load(self, old_rq: RunqueueSnapshot, new_rq: RunqueueSnapshot):
        diff = new_rq.load() - old_rq.load()
        self.total_load += diff

    def sched_switch(self, ts_list, prev: int, prev_state: int, next_pid: int, cpu: int):
        """Process sched_switch in time slice."""
        old_rq = self.prev.rqs[cpu]
        new_rq = old_rq.sched_switch(prev, prev_state, next_pid)

        if old_rq is new_rq:
            return

        self.rqs[cpu] = new_rq
        self.__update_total_load(old_rq, new_rq)
        ts_list.append(self)
        self.event_cpus = [cpu]

    def migrate(self, ts_list, new: int, old_cpu: int, new_cpu: int):
        """Process task migration in time slice."""
        if old_cpu == new_cpu:
            return
        old_rq = self.prev.rqs[old_cpu]
        out_rq = old_rq.migrate_out(new)
        self.rqs[old_cpu] = out_rq
        self.__update_total_load(old_rq, out_rq)

        new_rq = self.prev.rqs[new_cpu]
        in_rq = new_rq.migrate_in(new)
        self.rqs[new_cpu] = in_rq
        self.__update_total_load(new_rq, in_rq)

        ts_list.append(self)

        if old_rq is not out_rq:
            self.event_cpus.append(old_cpu)
        self.event_cpus.append(new_cpu)

    def wake_up(self, ts_list, pid: int, cpu: int, fork: bool):
        """Process wakeup in time slice."""
        old_rq = self.prev.rqs[cpu]
        if fork:
            new_rq = old_rq.wake_up_new(pid)
        else:
            new_rq = old_rq.wake_up(pid)

        if new_rq is old_rq:
            return
        self.rqs[cpu] = new_rq
        self.__update_total_load(old_rq, new_rq)
        ts_list.append(self)
        self.event_cpus = [cpu]

    def next(self, t: int):
        """Create next time slice."""
        self.end = t
        return TimeSlice(t, self)


class TimeSliceList(UserList):
    """List of time slices with search capabilities."""

    def __init__(self, arg=None):
        super().__init__(arg if arg is not None else [])
        self.root_win = None

    def get_time_slice(self, ts: int) -> TimeSlice:
        """Get or create time slice for timestamp."""
        if len(self.data) == 0:
            ts_slice = TimeSlice(ts, TimeSlice(-1, None))
        else:
            ts_slice = self.data[-1].next(ts)
        return ts_slice

    def find_time_slice(self, ts: int) -> int:
        """Binary search for time slice containing timestamp."""
        if not self.data:
            return -1
        start = 0
        end = len(self.data)
        found = -1
        searching = True
        while searching:
            if start in (end, end - 1):
                searching = False

            i = (end + start) // 2
            if self.data[i].start <= ts <= self.data[i].end:
                found = i
                break

            if self.data[i].end < ts:
                start = i
            elif self.data[i].start > ts:
                end = i

        return found

    def set_root_win(self, win):
        """Set root window for GUI."""
        self.root_win = win

    def mouse_down(self, cpu: int, t: int):
        """Handle mouse down event from GUI."""
        idx = self.find_time_slice(t)
        if idx == -1:
            return

        ts = self[idx]
        rq = ts.rqs[cpu]
        raw = f"CPU: {cpu}\n"
        raw += f"Last event : {repr(rq.event)}\n"
        raw += f"Timestamp : {ts.start // (10 ** 9)}.{ts.start % (10 ** 9) // 1000:06d}\n"
        raw += f"Duration : {(ts.end - ts.start) // (10 ** 3):6d} us\n"
        raw += f"Load = {rq.load()}\n"
        for task in rq.tasks:
            raw += f"{thread_name(task)} \n"

        if self.root_win:
            self.root_win.update_summary(raw)

    def update_rectangle_cpu(self, slice_obj: TimeSlice, cpu: int):
        """Update rectangle for CPU in GUI."""
        rq = slice_obj.rqs[cpu]

        if slice_obj.total_load != 0:
            load_rate = rq.load() / float(slice_obj.total_load)
        else:
            load_rate = 0

        red_power = int(0xff - (0xff * load_rate))
        color = (0xff, red_power, red_power)

        top_color = None
        if cpu in slice_obj.event_cpus:
            top_color = rq.event.color()

        if self.root_win:
            self.root_win.paint_rectangle_zone(cpu, color, top_color,
                                               slice_obj.start, slice_obj.end)

    def fill_zone(self, start: int, end: int):
        """Fill zone in GUI."""
        i = self.find_time_slice(start)
        if i == -1:
            return

        for idx in range(i, len(self.data)):
            timeslice = self.data[idx]
            if timeslice.start > end:
                return

            for cpu in timeslice.rqs:
                self.update_rectangle_cpu(timeslice, cpu)

    def interval(self) -> tuple[int, int]:
        """Return start and end timestamps."""
        if len(self.data) == 0:
            return 0, 0
        return self.data[0].start, self.data[-1].end

    def nr_rectangles(self) -> int:
        """Return maximum CPU number."""
        if not self.data:
            return 0
        last_ts = self.data[-1]
        max_cpu = 0
        for cpu in last_ts.rqs:
            max_cpu = max(max_cpu, cpu)
        return max_cpu


class SchedMigrationAnalyzer:
    """Analyzes task migrations and manages time slices."""

    def __init__(self):
        self.current_tsk = defaultdict(lambda: -1)
        self.timeslices = TimeSliceList()

    def sched_switch(self, time: int, cpu: int, prev_comm: str, prev_pid: int, prev_state: int,
                     next_comm: str, next_pid: int):
        """Handle sched_switch event."""
        on_cpu_task = self.current_tsk[cpu]

        if on_cpu_task not in (-1, prev_pid):
            print(f"Sched switch event rejected ts: {time} cpu: {cpu} "
                  f"prev: {prev_comm}({prev_pid}) next: {next_comm}({next_pid})")

        threads[prev_pid] = prev_comm
        threads[next_pid] = next_comm
        self.current_tsk[cpu] = next_pid

        ts = self.timeslices.get_time_slice(time)
        ts.sched_switch(self.timeslices, prev_pid, prev_state, next_pid, cpu)

    def migrate(self, time: int, pid: int, orig_cpu: int, dest_cpu: int):
        """Handle sched_migrate_task event."""
        ts = self.timeslices.get_time_slice(time)
        ts.migrate(self.timeslices, pid, orig_cpu, dest_cpu)

    def wake_up(self, time: int, pid: int, success: int, target_cpu: int, fork: bool):
        """Handle wakeup event."""
        if success == 0:
            return
        ts = self.timeslices.get_time_slice(time)
        ts.wake_up(self.timeslices, pid, target_cpu, fork)

    def process_event(self, sample: perf.sample_event) -> None:
        """Collect events and pass to analyzer."""
        name = str(sample.evsel)
        if name.startswith("evsel(") and name.endswith(")"):
            name = name[6:-1]
        if name.count(":") >= 2:
            name = ":".join(name.split(":")[:2])
        time = sample.sample_time
        cpu = sample.sample_cpu
        _pid = sample.sample_pid
        _comm = "Unknown"

        if name.startswith("sched:sched_switch"):
            prev_comm = getattr(sample, "prev_comm", "Unknown")
            prev_pid = getattr(sample, "prev_pid", -1)
            prev_state = getattr(sample, "prev_state", 0)
            next_comm = getattr(sample, "next_comm", "Unknown")
            next_pid = getattr(sample, "next_pid", -1)
            self.sched_switch(time, cpu, prev_comm, prev_pid, prev_state, next_comm, next_pid)
        elif name.startswith("sched:sched_migrate_task"):
            task_pid = getattr(sample, "pid", -1)
            orig_cpu = getattr(sample, "orig_cpu", -1)
            dest_cpu = getattr(sample, "dest_cpu", -1)
            self.migrate(time, task_pid, orig_cpu, dest_cpu)
        elif name.startswith("sched:sched_wakeup_new"):
            task_pid = getattr(sample, "pid", -1)
            success = getattr(sample, "success", 1)
            target_cpu = getattr(sample, "target_cpu", -1)
            self.wake_up(time, task_pid, success, target_cpu, True)
        elif name.startswith("sched:sched_wakeup"):
            task_pid = getattr(sample, "pid", -1)
            success = getattr(sample, "success", 1)
            target_cpu = getattr(sample, "target_cpu", -1)
            self.wake_up(time, task_pid, success, target_cpu, False)

    def run_gui(self):
        """Start wxPython GUI."""
        try:
            wx_mod: Any = importlib.import_module("wx")
            sched_gui: Any = importlib.import_module("SchedGui")
        except ImportError:
            print("wxPython is not available. Cannot start GUI.")
            return
        app = wx_mod.App(False)
        _frame = sched_gui.RootFrame(self.timeslices, "Migration")
        app.MainLoop()


if __name__ == "__main__":
    ap = argparse.ArgumentParser(description="Cpu task migration overview toy")
    ap.add_argument("-i", "--input", default="perf.data", help="Input file name")
    ap.add_argument("-v", "--verbose", action="store_true",
                    help="Print parsed migration summary")
    ap.add_argument("--gui", action="store_true", default=True,
                    help="Start the GUI (default)")
    ap.add_argument("--no-gui", action="store_true", help="Do not start the GUI")
    args = ap.parse_args()

    analyzer = SchedMigrationAnalyzer()
    session = None

    try:
        try:
            session = perf.session(perf.data(args.input), sample=analyzer.process_event)
            try:
                session.process_events()
            except KeyboardInterrupt:
                pass
        finally:
            # Release the perf.session reference before launching the long-running
            # wxPython GUI event loop so the underlying C session and callback
            # reference cycle are freed immediately.
            session = None
        if args.verbose:
            start_ts, end_ts = analyzer.timeslices.interval()
            print(f"Timeslices: {len(analyzer.timeslices.data)} "
                  f"(interval: {start_ts}..{end_ts})")
        if args.gui and not args.no_gui:
            analyzer.run_gui()
    except KeyboardInterrupt:
        pass
    except OSError as e:
        print(f"Error processing events: {e}")
        sys.exit(1)