#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-2.0 """ Cpu task migration overview toy Copyright (C) 2010 Frederic Weisbecker 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)