path: root/tools/perf/python/ttimechart.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
"""ttimechart.py - interactive perf timechart written using textual.

Reads a perf.data file, typically created with 'perf timechart record', and
displays per-CPU and per-task timelines in the terminal. Scheduler
(sched:sched_switch, sched:sched_wakeup), power (power:cpu_idle,
power:cpu_frequency) and I/O syscall (perf timechart record -I) tracepoints
are understood. Unlike 'perf timechart', which writes an SVG file, the
timeline can be zoomed, panned and queried interactively.

Usage:
    perf timechart record -- <workload>
    perf timechart --tui
or:
    perf script ttimechart [-i perf.data]
"""
from __future__ import annotations

from abc import ABC, abstractmethod
import argparse
import bisect
from collections import defaultdict
from dataclasses import dataclass, replace
import math
import os
import sys
import threading
from time import monotonic
from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple

import perf
from rich.segment import Segment
from rich.style import Style
from textual import events, on, work
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.color import Color
from textual.geometry import Size
from textual.message import Message
from textual.reactive import reactive
from textual.scroll_view import ScrollView
from textual.strip import Strip
from textual.widgets import DataTable, Footer, Header, Static, TabbedContent, TabPane
from textual.widgets.data_table import CellDoesNotExist, RowDoesNotExist

# Task states, SLEEPING and UNKNOWN aren't drawn.
STATE_UNKNOWN = -1
STATE_SLEEPING = 0
STATE_RUNNING = 1
STATE_WAITING = 2
STATE_BLOCKED = 3
NUM_STATES = 4
STATE_NAMES = {
    STATE_UNKNOWN: "unknown",
    STATE_SLEEPING: "sleeping",
    STATE_RUNNING: "running",
    STATE_WAITING: "runnable (waiting for a CPU)",
    STATE_BLOCKED: "blocked (uninterruptible)",
}

# I/O types, matching builtin-timechart.c.
IOTYPE_READ = 0
IOTYPE_WRITE = 1
IOTYPE_SYNC = 2
IOTYPE_TX = 3
IOTYPE_RX = 4
IOTYPE_POLL = 5
NUM_IOTYPES = 6
IOTYPE_NAMES = ["read", "write", "sync", "tx", "rx", "poll"]

IO_SYSCALLS = {
    "read": IOTYPE_READ, "pread64": IOTYPE_READ, "readv": IOTYPE_READ,
    "preadv": IOTYPE_READ,
    "write": IOTYPE_WRITE, "pwrite64": IOTYPE_WRITE, "writev": IOTYPE_WRITE,
    "pwritev": IOTYPE_WRITE,
    "sync": IOTYPE_SYNC, "sync_file_range": IOTYPE_SYNC, "fsync": IOTYPE_SYNC,
    "msync": IOTYPE_SYNC,
    "recvfrom": IOTYPE_RX, "recvmmsg": IOTYPE_RX, "recvmsg": IOTYPE_RX,
    "sendto": IOTYPE_TX, "sendmsg": IOTYPE_TX, "sendmmsg": IOTYPE_TX,
    "epoll_pwait": IOTYPE_POLL, "epoll_wait": IOTYPE_POLL, "poll": IOTYPE_POLL,
    "ppoll": IOTYPE_POLL, "pselect6": IOTYPE_POLL, "select": IOTYPE_POLL,
}

# Trace flags for interrupt context in the common_flags tracepoint field.
TRACE_FLAG_HARDIRQ = 0x08
TRACE_FLAG_SOFTIRQ = 0x10
# Value of state in power:cpu_idle when leaving idle.
PWR_EVENT_EXIT = 0xffffffff
# Width of the row label column.
LABEL_WIDTH = 28
# Characters for drawing fractional bars.
BARS = " ▁▂▃▄▅▆▇█"
NSEC_PER_SEC = 1_000_000_000


def fmt_duration(nsecs: float) -> str:
    """Format a duration in nanoseconds with an appropriate unit."""
    if nsecs >= NSEC_PER_SEC:
        return f"{nsecs / NSEC_PER_SEC:.3f}s"
    if nsecs >= 1_000_000:
        return f"{nsecs / 1_000_000:.3f}ms"
    if nsecs >= 1_000:
        return f"{nsecs / 1_000:.3f}us"
    return f"{nsecs:.0f}ns"


def escape(text: str) -> str:
    """Escape text, such as a task name, for use in textual markup.

    rich.markup.escape doesn't escape tags like "[1]", the name given to tasks
    with an unknown command, but textual fails to parse them.
    """
    return text.replace("[", "\\[")


def make_fixed_length_string(s: str, length: int, pad_char: str = ' ') -> str:
    """Truncate or right pad s so that it is length characters long."""
    return s[:length] if len(s) > length else s.ljust(length, pad_char)


def bar_char(frac: float) -> str:
    """A block character whose height represents frac, in the range [0, 1]."""
    if frac <= 0:
        return BARS[0]
    return BARS[max(1, min(8, round(frac * 8)))]


class SegmentList:
    """A time ordered list of non-overlapping [start, end) segments.

    Each segment has a small integer key, used for computing coverage, and
    arbitrary associated data.
    """
    def __init__(self) -> None:
        self.starts: List[int] = []
        self.ends: List[int] = []
        self.keys: List[int] = []
        self.data: List[Any] = []

    def __len__(self) -> int:
        return len(self.starts)

    def add(self, start: int, end: int, key: int, data: Any = None) -> None:
        """Append a segment, segments must be added in time order."""
        if end <= start:
            return
        if self.ends and start < self.ends[-1]:
            # Clip overlaps caused by inconsistent data.
            start = self.ends[-1]
            if end <= start:
                return
        self.starts.append(start)
        self.ends.append(end)
        self.keys.append(key)
        self.data.append(data)

    def find(self, time: float) -> int:
        """Index of the segment containing time or -1."""
        i = bisect.bisect_right(self.starts, time) - 1
        if i >= 0 and time < self.ends[i]:
            return i
        return -1

    def last_before(self, time: float) -> int:
        """Index of the last segment starting at or before time or -1."""
        return bisect.bisect_right(self.starts, time) - 1

    def next_change(self, time: float) -> Optional[int]:
        """The first segment start or end after time."""
        candidates = []
        i = bisect.bisect_right(self.starts, time)
        if i < len(self.starts):
            candidates.append(self.starts[i])
        i = bisect.bisect_right(self.ends, time)
        if i < len(self.ends):
            candidates.append(self.ends[i])
        return min(candidates) if candidates else None

    def prev_change(self, time: float) -> Optional[int]:
        """The last segment start or end before time."""
        candidates = []
        i = bisect.bisect_left(self.starts, time) - 1
        if i >= 0:
            candidates.append(self.starts[i])
        i = bisect.bisect_left(self.ends, time) - 1
        if i >= 0:
            candidates.append(self.ends[i])
        return max(candidates) if candidates else None

    def coverage(self, t0: float, dt: float, width: int, nkeys: int,
                 weight_fn=None) -> List[List[float]]:
        """Time covered by each key in each of width columns of size dt.

        If weight_fn is given then, rather than the time covered, the time
        covered multiplied by weight_fn(data) is accumulated.
        """
        cols = [[0.0] * nkeys for _ in range(width)]
        t1 = t0 + dt * width
        i = bisect.bisect_right(self.ends, t0)
        num = len(self.starts)
        while i < num and self.starts[i] < t1:
            start = max(self.starts[i], t0)
            end = min(self.ends[i], t1)
            key = self.keys[i]
            weight = weight_fn(self.data[i]) if weight_fn else 1.0
            x0 = min(int((start - t0) / dt), width - 1)
            x1 = min(int((end - t0) / dt), width - 1)
            if x0 == x1:
                cols[x0][key] += (end - start) * weight
            else:
                cols[x0][key] += (t0 + (x0 + 1) * dt - start) * weight
                for x in range(x0 + 1, x1):
                    cols[x][key] += dt * weight
                cols[x1][key] += (end - (t0 + x1 * dt)) * weight
            i += 1
        return cols


class Task:
    """Scheduling and I/O history of a single thread."""
    def __init__(self, tid: int, comm: str) -> None:
        self.tid = tid
        self.comm = comm
        self.comms: List[str] = [comm]
        self.state = STATE_UNKNOWN
        self.since = 0
        self.cpu = -1
        # Segments with a state key and the CPU as data.
        self.segs = SegmentList()
        # Segments with an I/O type key and (fd, ret) as data.
        self.io = SegmentList()
        self.io_pending: Optional[Tuple[int, int, int]] = None
        # Wakeups of this task as (time, waker tid) pairs.
        self.wakeups: List[Tuple[int, int]] = []
        self.totals = [0] * NUM_STATES
        self.switches = 0
        self.io_bytes = 0

    def name(self) -> str:
        """Name for the task used in labels."""
        return f"{self.comm} ({self.tid})"

    def set_comm(self, comm: Optional[str]) -> None:
        """Update the task's command name."""
        if not comm or comm == self.comm:
            return
        self.comm = comm
        if comm not in self.comms:
            self.comms.append(comm)

    def change_state(self, time: int, state: int, cpu: int = -1) -> None:
        """Record the current state as a segment and switch to a new state."""
        if self.state in (STATE_RUNNING, STATE_WAITING, STATE_BLOCKED) and time > self.since:
            self.segs.add(self.since, time, self.state, self.cpu)
            self.totals[self.state] += time - self.since
        self.state = state
        self.since = time
        if cpu >= 0:
            self.cpu = cpu

    def passes_filter(self, filters: Sequence[str]) -> bool:
        """Does the task match one of the process filters (pid or name)?"""
        if not filters:
            return True
        return any(f == str(self.tid) or f in self.comms for f in filters)


class Cpu:
    """Activity on a single CPU."""
    def __init__(self, cpu: int) -> None:
        self.cpu = cpu
        self.cur_tid = -1
        self.since = 0
        # Busy segments, key 1, with the running tid as data.
        self.run = SegmentList()
        # Idle state segments, key 1, with the C-state as data.
        self.cstate = SegmentList()
        self.cstate_cur: Optional[Tuple[int, int]] = None
        # Frequency segments, key 1, with the frequency in kHz as data.
        self.pstate = SegmentList()
        self.pstate_cur: Optional[Tuple[int, int]] = None


class LoadCancelled(Exception):
    """Raised from the sample callback to stop processing events early."""


class TimechartData:
    """Builds per-task and per-CPU timelines from perf events.

    The data may be displayed while it is loaded in another thread, lock must
    be held when modifying it or when reading it from another thread.
    """
    # Number of samples between checks for cancellation and progress.
    PROGRESS_INTERVAL = 1000
    # Minimum and maximum time between calls to the progress callback.
    PROGRESS_SECONDS = 1.0
    PROGRESS_MAX_SECONDS = 10.0
    # Updating the views costs more as more data is loaded, so the time between
    # progress calls grows as this fraction of the time spent loading. This
    # bounds the fraction of the load time spent updating the views.
    PROGRESS_FRACTION = 0.25

    def __init__(self) -> None:
        self.tasks: Dict[int, Task] = {}
        self.cpus: Dict[int, Cpu] = {}
        self.first_time = 0
        self.last_time = 0
        self.min_freq = 0
        self.max_freq = 0
        self.max_cstate = 0
        self.sched_events = 0
        self.power_events = 0
        self.io_events = 0
        self.nr_samples = 0
        self.unhandled: Dict[str, int] = defaultdict(int)
        self.session: Optional[perf.session] = None
        # Evsel name and event handler, keyed by sample ID.
        self._handlers: Dict[int, Tuple[str, Optional[Callable[[int, perf.sample_event],
                                                               None]]]] = {}
        self.lock = threading.Lock()
        # Set, possibly from another thread, to stop processing events.
        self.cancelled = False
        # Called periodically, see PROGRESS_FRACTION, while processing events.
        self.progress: Optional[Callable[[], None]] = None
        self.start_progress = monotonic()
        self.last_progress = self.start_progress

    def has_events(self) -> bool:
        """Were any events that can be displayed processed?"""
        return bool(self.sched_events or self.power_events or self.io_events)

    def task(self, tid: int, comm: Optional[str] = None) -> Task:
        """Find or create a task."""
        task = self.tasks.get(tid)
        if task is None:
            if not comm and self.session:
                try:
                    thread = self.session.find_thread(tid, tid)
                    comm = thread.comm() if thread else None
                except (OSError, ValueError, KeyError, RuntimeError, TypeError, AttributeError):
                    comm = None
            task = Task(tid, comm or f"[{tid}]")
            self.tasks[tid] = task
        else:
            task.set_comm(comm)
        return task

    def cpu(self, cpu: int) -> Cpu:
        """Find or create a CPU."""
        c = self.cpus.get(cpu)
        if c is None:
            c = Cpu(cpu)
            self.cpus[cpu] = c
        return c

    def sched_switch(self, time: int, cpu: int, prev_tid: int, prev_comm: Optional[str],
                     prev_state: int, next_tid: int, next_comm: Optional[str]) -> None:
        """Process a sched:sched_switch event."""
        self.sched_events += 1
        c = self.cpu(cpu)
        if c.cur_tid == -1 and prev_tid != 0:
            # Assume the task was running from the start of the trace.
            c.cur_tid = prev_tid
            c.since = self.first_time
        if c.cur_tid > 0:
            c.run.add(c.since, time, 1, c.cur_tid)
        c.cur_tid = next_tid
        c.since = time

        if prev_tid != 0:
            prev = self.task(prev_tid, prev_comm)
            if prev.state == STATE_UNKNOWN:
                prev.state = STATE_RUNNING
                prev.since = self.first_time
                prev.cpu = cpu
            # Ignore bits like TASK_REPORT_MAX used to report preemption.
            state = prev_state & 0xff
            if state == 0:
                new_state = STATE_WAITING
            elif state & 2:
                new_state = STATE_BLOCKED
            else:
                new_state = STATE_SLEEPING
            prev.change_state(time, new_state)
            prev.switches += 1

        if next_tid != 0:
            nxt = self.task(next_tid, next_comm)
            nxt.change_state(time, STATE_RUNNING, cpu)

    def sched_wakeup(self, time: int, wakee: int, comm: Optional[str], waker: int) -> None:
        """Process a sched:sched_wakeup or sched:sched_wakeup_new event."""
        self.sched_events += 1
        if wakee == 0:
            return
        task = self.task(wakee, comm)
        task.wakeups.append((time, waker))
        if task.state in (STATE_UNKNOWN, STATE_SLEEPING, STATE_BLOCKED):
            task.change_state(time, STATE_WAITING)

    def cstate_start(self, time: int, cpu: int, state: int) -> None:
        """Enter an idle state."""
        self.power_events += 1
        c = self.cpu(cpu)
        if c.cstate_cur:
            c.cstate.add(c.cstate_cur[0], time, 1, c.cstate_cur[1])
        c.cstate_cur = (time, state)
        self.max_cstate = max(self.max_cstate, state)

    def cstate_end(self, time: int, cpu: int) -> None:
        """Leave an idle state."""
        self.power_events += 1
        c = self.cpu(cpu)
        if c.cstate_cur:
            c.cstate.add(c.cstate_cur[0], time, 1, c.cstate_cur[1])
            c.cstate_cur = None

    def pstate_change(self, time: int, cpu: int, freq: int) -> None:
        """Change of CPU frequency, freq is in kHz."""
        if freq <= 0 or freq > 8000000:
            return
        self.power_events += 1
        c = self.cpu(cpu)
        if c.pstate_cur:
            c.pstate.add(c.pstate_cur[0], time, 1, c.pstate_cur[1])
        c.pstate_cur = (time, freq)
        self.max_freq = max(self.max_freq, freq)
        self.min_freq = freq if not self.min_freq else min(self.min_freq, freq)

    def io_enter(self, time: int, tid: int, iotype: int, fd: int) -> None:
        """Entry to an I/O syscall."""
        self.io_events += 1
        self.task(tid).io_pending = (time, iotype, fd)

    def io_exit(self, time: int, tid: int, iotype: int, ret: int) -> None:
        """Exit from an I/O syscall."""
        self.io_events += 1
        task = self.task(tid)
        pending = task.io_pending
        task.io_pending = None
        if not pending or pending[1] != iotype:
            return
        start = pending[0]
        task.io.add(start, max(time, start + 1), iotype, (pending[2], ret))
        if ret > 0 and iotype in (IOTYPE_READ, IOTYPE_WRITE, IOTYPE_TX, IOTYPE_RX):
            task.io_bytes += ret

    def _on_sched_switch(self, time: int, sample: perf.sample_event) -> None:
        self.sched_switch(time, sample.sample_cpu, sample.prev_pid,
                          getattr(sample, "prev_comm", None), sample.prev_state,
                          sample.next_pid, getattr(sample, "next_comm", None))

    def _on_sched_wakeup(self, time: int, sample: perf.sample_event) -> None:
        waker = getattr(sample, "common_pid", sample.sample_tid)
        flags = getattr(sample, "common_flags", 0)
        if flags & (TRACE_FLAG_HARDIRQ | TRACE_FLAG_SOFTIRQ):
            waker = -1
        self.sched_wakeup(time, sample.pid, getattr(sample, "comm", None), waker)

    def _cstate(self, time: int, cpu: int, state: int) -> None:
        if (state & 0xffffffff) == PWR_EVENT_EXIT:
            self.cstate_end(time, cpu)
        else:
            self.cstate_start(time, cpu, state)

    def _on_cpu_idle(self, time: int, sample: perf.sample_event) -> None:
        self._cstate(time, sample.cpu_id, sample.state)

    def _on_power_start(self, time: int, sample: perf.sample_event) -> None:
        self._cstate(time, sample.cpu_id, sample.value)

    def _on_power_end(self, time: int, sample: perf.sample_event) -> None:
        self.cstate_end(time, sample.sample_cpu)

    def _on_cpu_frequency(self, time: int, sample: perf.sample_event) -> None:
        self.pstate_change(time, sample.cpu_id, sample.state)

    def _on_power_frequency(self, time: int, sample: perf.sample_event) -> None:
        self.pstate_change(time, sample.cpu_id, sample.value)

    def _handler_for(self, name: str) -> Optional[Callable[[int, perf.sample_event], None]]:
        """Find the handler for events with the given evsel name."""
        handlers: Dict[str, Callable[[int, perf.sample_event], None]] = {
            "sched:sched_switch": self._on_sched_switch,
            "sched:sched_wakeup": self._on_sched_wakeup,
            "sched:sched_wakeup_new": self._on_sched_wakeup,
            "power:cpu_idle": self._on_cpu_idle,
            "power:power_start": self._on_power_start,
            "power:power_end": self._on_power_end,
            "power:cpu_frequency": self._on_cpu_frequency,
            "power:power_frequency": self._on_power_frequency,
        }
        if name in handlers:
            return handlers[name]
        if name.startswith("syscalls:sys_enter_") and name[19:] in IO_SYSCALLS:
            iotype = IO_SYSCALLS[name[19:]]
            return lambda time, sample: self.io_enter(time, sample.sample_tid, iotype,
                                                      getattr(sample, "fd", -1))
        if name.startswith("syscalls:sys_exit_") and name[18:] in IO_SYSCALLS:
            iotype = IO_SYSCALLS[name[18:]]
            return lambda time, sample: self.io_exit(time, sample.sample_tid, iotype,
                                                     sample.ret)
        return None

    def process_event(self, sample: perf.sample_event) -> None:
        """Callback from perf.session for each sample."""
        self.nr_samples += 1
        if self.nr_samples % self.PROGRESS_INTERVAL == 0:
            if self.cancelled:
                raise LoadCancelled()
            now = monotonic()
            interval = min(max(self.PROGRESS_SECONDS,
                               (now - self.start_progress) * self.PROGRESS_FRACTION),
                           self.PROGRESS_MAX_SECONDS)
            if self.progress and now - self.last_progress >= interval:
                with self.lock:
                    self.update_comms()
                # Must not hold the lock as the callback may read the data.
                self.progress()
                # Time from when the callback, that may block, returns.
                self.last_progress = monotonic()
        with self.lock:
            self._process_event(sample)

    def _process_event(self, sample: perf.sample_event) -> None:
        """Update the data from a sample, the lock must be held."""
        time = sample.sample_time
        if not self.first_time or time < self.first_time:
            self.first_time = time
        self.last_time = max(self.last_time, time)

        # Computing the evsel name and matching it is relatively expensive,
        # so cache the result by sample ID. Each ID belongs to a single evsel.
        sample_id = sample.sample_id
        cached = self._handlers.get(sample_id)
        if cached is None:
            name = str(sample.evsel)
            if name.startswith("evsel(") and name.endswith(")"):
                name = name[6:-1]
            cached = (name, self._handler_for(name))
            self._handlers[sample_id] = cached
        name, handler = cached
        if handler is None:
            self.unhandled[name] += 1
            return
        try:
            handler(time, sample)
        except AttributeError:
            self.unhandled[name] += 1

    def update_comms(self) -> None:
        """Refresh task command names from the session, the lock must be held."""
        if not self.session:
            return
        for tid, task in self.tasks.items():
            try:
                thread = self.session.find_thread(tid, tid)
                if thread:
                    task.set_comm(thread.comm())
            except (OSError, ValueError, KeyError, RuntimeError, TypeError, AttributeError):
                pass

    def finish(self) -> None:
        """Close open segments at the end of the trace."""
        with self.lock:
            self.update_comms()
            end = self.last_time
            for task in self.tasks.values():
                task.change_state(end, STATE_UNKNOWN)
            for c in self.cpus.values():
                if c.cur_tid > 0:
                    c.run.add(c.since, end, 1, c.cur_tid)
                if c.cstate_cur:
                    c.cstate.add(c.cstate_cur[0], end, 1, c.cstate_cur[1])
                    c.cstate_cur = None
                if c.pstate_cur:
                    c.pstate.add(c.pstate_cur[0], end, 1, c.pstate_cur[1])
                    c.pstate_cur = None

    def fmt_time(self, time: float) -> str:
        """Format an absolute timestamp relative to the trace start."""
        return f"{(time - self.first_time) / NSEC_PER_SEC:.6f}s"

    def task_name(self, tid: int) -> str:
        """Name of a task, or a description for special tids."""
        if tid == 0:
            return "idle"
        if tid < 0:
            return "interrupt"
        task = self.tasks.get(tid)
        return task.name() if task else f"[{tid}]"

    def sched_tasks(self, filters: Sequence[str]) -> List[Task]:
        """Tasks with scheduling history passing the filters, sorted by tid."""
        return sorted((t for t in self.tasks.values()
                       if len(t.segs) and t.passes_filter(filters)),
                      key=lambda t: t.tid)

    def io_tasks(self, filters: Sequence[str]) -> List[Task]:
        """Tasks with I/O passing the filters, most I/O first."""
        return sorted((t for t in self.tasks.values()
                       if len(t.io) and t.passes_filter(filters)),
                      key=lambda t: -len(t.io))

    def dump(self, power_only: bool, tasks_only: bool, filters: Sequence[str]) -> None:
        """Print a plain text summary, for use without a terminal UI."""
        duration = self.last_time - self.first_time
        span = max(duration, 1)
        tasks = self.sched_tasks(filters)
        io_tasks = self.io_tasks(filters)
        print(f"Duration: {fmt_duration(duration)}, CPUs: {len(self.cpus)}, "
              f"tasks: {len(tasks) or len(io_tasks)}, sched events: {self.sched_events}, "
              f"power events: {self.power_events}, I/O events: {self.io_events}")
        if not tasks_only:
            for c in sorted(self.cpus.values(), key=lambda c: c.cpu):
                busy = sum(e - s for s, e in zip(c.run.starts, c.run.ends))
                print(f"CPU {c.cpu}: busy {busy * 100 / span:.1f}%, "
                      f"{len(c.run)} runs, {len(c.cstate)} idle periods, "
                      f"{len(c.pstate)} frequency periods")
        if power_only:
            return
        print(f"{'Task':<24} {'TID':>8} {'Running':>12} {'Waiting':>12} {'Blocked':>12} "
              f"{'Switches':>9} {'Wakeups':>8}")
        for task in sorted(tasks, key=lambda t: -t.totals[STATE_RUNNING]):
            print(f"{task.comm[:24]:<24} {task.tid:>8} "
                  f"{fmt_duration(task.totals[STATE_RUNNING]):>12} "
                  f"{fmt_duration(task.totals[STATE_WAITING]):>12} "
                  f"{fmt_duration(task.totals[STATE_BLOCKED]):>12} "
                  f"{task.switches:>9} {len(task.wakeups):>8}")
        for task in io_tasks:
            print(f"I/O {task.name()}: {len(task.io)} syscalls, {task.io_bytes} bytes")


Cell = Tuple[str, Style]


class ThemeColors:
    """Colors and Rich styles derived from the active Textual theme."""
    def __init__(self, theme_variables: Mapping[str, str]) -> None:
        self.running = theme_variables["primary"]
        self.waiting = theme_variables["error"]
        self.blocked = theme_variables["warning"]
        self.idle_light = theme_variables["secondary-lighten-2"]
        self.idle_dark = theme_variables["secondary-darken-2"]
        self.freq_low = theme_variables["success"]
        self.freq_high = theme_variables["error"]
        self.io = [
            theme_variables["success"],
            theme_variables["error"],
            theme_variables["warning"],
            theme_variables["accent"],
            theme_variables["primary"],
            theme_variables["secondary-lighten-2"],
        ]
        run_color = Color.parse(self.running).rich_color
        wait_color = Color.parse(self.waiting).rich_color
        block_color = Color.parse(self.blocked).rich_color
        self.run_style = Style(color=run_color)
        self.wait_style = Style(color=run_color, bgcolor=wait_color)
        self.block_style = Style(color=run_color, bgcolor=block_color)
        low = Color.parse(self.freq_low)
        high = Color.parse(self.freq_high)
        self.freq_styles = [Style(color=low.blend(high, x / 8).rich_color) for x in range(9)]
        light = Color.parse(self.idle_light)
        dark = Color.parse(self.idle_dark)
        self.idle_styles = [Style(color=light.blend(dark, x / 8).rich_color) for x in range(9)]
        self.io_styles = [Style(color=Color.parse(c).rich_color) for c in self.io]
        self.io_err_styles = [Style(color=Color.parse(c).rich_color, underline=True)
                              for c in self.io]
        accent = Color.parse(theme_variables["accent"]).rich_color
        accent_muted = Color.parse(theme_variables["accent-muted"]).rich_color
        self.selected_style = Style(color=accent, bgcolor=accent_muted, bold=True)


class Row(ABC):
    """A row within a timeline view."""
    def __init__(self, label: str) -> None:
        self._label = label

    def label(self) -> str:
        """Label shown to the left of the row."""
        return self._label

    @abstractmethod
    def segments(self) -> SegmentList:
        """Segments used for navigating between changes."""

    @abstractmethod
    def cells(self, t0: float, dt: float, width: int, colors: ThemeColors) -> List[Cell]:
        """Cells for the columns starting at time t0 and dt wide."""

    @abstractmethod
    def describe(self, time: float, t0: float, t1: float) -> str:
        """Rich markup describing the row at time within the window [t0, t1)."""


class TaskRow(Row):
    """Row showing the running, waiting and blocked states of a task.

    The height of the bar shows the fraction of the time running, the
    background shows waiting for a CPU or being blocked.
    """
    def __init__(self, data: TimechartData, task: Task) -> None:
        super().__init__(task.name())
        self.data = data
        self.task = task

    def label(self) -> str:
        return self.task.name()

    def segments(self) -> SegmentList:
        return self.task.segs

    def cells(self, t0: float, dt: float, width: int, colors: ThemeColors) -> List[Cell]:
        result = []
        for col in self.task.segs.coverage(t0, dt, width, NUM_STATES):
            run = col[STATE_RUNNING] / dt
            wait = col[STATE_WAITING]
            block = col[STATE_BLOCKED]
            if wait > 0 and wait >= block:
                style = colors.wait_style
            elif block > 0:
                style = colors.block_style
            else:
                style = colors.run_style
            result.append((bar_char(run), style))
        return result

    def describe(self, time: float, t0: float, t1: float) -> str:
        task = self.task
        i = task.segs.find(time)
        if i >= 0:
            state = task.segs.keys[i]
            desc = STATE_NAMES[state]
            if state == STATE_RUNNING:
                desc += f" on CPU {task.segs.data[i]}"
            elif state == STATE_WAITING and task.segs.data[i] >= 0:
                desc += f", last ran on CPU {task.segs.data[i]}"
            start = task.segs.starts[i]
            end = task.segs.ends[i]
            desc += (f" from {self.data.fmt_time(start)} for "
                     f"{fmt_duration(end - start)}")
        else:
            desc = "sleeping or not traced"
        window = task.segs.coverage(t0, max(t1 - t0, 1), 1, NUM_STATES)[0]
        span = max(t1 - t0, 1)
        lines = [
            f"[b]{escape(task.name())}[/b]: {desc}",
            f"In view: running {window[STATE_RUNNING] * 100 / span:.1f}%, "
            f"waiting {window[STATE_WAITING] * 100 / span:.1f}%, "
            f"blocked {window[STATE_BLOCKED] * 100 / span:.1f}%",
            f"Total: running {fmt_duration(task.totals[STATE_RUNNING])}, "
            f"waiting {fmt_duration(task.totals[STATE_WAITING])}, "
            f"blocked {fmt_duration(task.totals[STATE_BLOCKED])}, "
            f"{task.switches} switches, {len(task.wakeups)} wakeups",
        ]
        idx = bisect.bisect_right(task.wakeups, (time, sys.maxsize)) - 1
        if idx >= 0:
            wake_time, waker = task.wakeups[idx]
            lines.append(f"Last woken at {self.data.fmt_time(wake_time)} by "
                         f"{escape(self.data.task_name(waker))} (press 'w' to go to waker)")
        if len(task.comms) > 1:
            lines.append(f"Names: {escape(', '.join(task.comms))}")
        return "\n".join(lines)


class CpuRow(Row):
    """Row showing how busy a CPU is, colored by frequency if known."""
    def __init__(self, data: TimechartData, cpu: Cpu) -> None:
        super().__init__(f"CPU {cpu.cpu}")
        self.data = data
        self.cpu = cpu

    def segments(self) -> SegmentList:
        return self.cpu.run

    def cells(self, t0: float, dt: float, width: int, colors: ThemeColors) -> List[Cell]:
        busy = self.cpu.run.coverage(t0, dt, width, 2)
        freqs: Optional[List[List[float]]] = None
        weighted: Optional[List[List[float]]] = None
        if len(self.cpu.pstate) and self.data.max_freq > self.data.min_freq:
            freqs = self.cpu.pstate.coverage(t0, dt, width, 2)
            weighted = self.cpu.pstate.coverage(t0, dt, width, 2, weight_fn=float)
        result = []
        for x in range(width):
            style = colors.run_style
            if freqs and weighted and freqs[x][1] > 0:
                freq = weighted[x][1] / freqs[x][1]
                frac = (freq - self.data.min_freq) / (self.data.max_freq - self.data.min_freq)
                style = colors.freq_styles[max(0, min(8, round(frac * 8)))]
            result.append((bar_char(busy[x][1] / dt), style))
        return result

    def describe(self, time: float, t0: float, t1: float) -> str:
        c = self.cpu
        i = c.run.find(time)
        if i >= 0:
            desc = (f"running {escape(self.data.task_name(c.run.data[i]))} from "
                    f"{self.data.fmt_time(c.run.starts[i])} for "
                    f"{fmt_duration(c.run.ends[i] - c.run.starts[i])}")
        else:
            desc = "idle"
        i = c.cstate.find(time)
        if i >= 0:
            desc += f", C-state C{c.cstate.data[i]}"
        i = c.pstate.find(time)
        if i >= 0:
            desc += f", {c.pstate.data[i] / 1000:.0f} MHz"
        span = max(t1 - t0, 1)
        window = c.run.coverage(t0, span, 1, 2)[0]
        return "\n".join([
            f"[b]CPU {c.cpu}[/b]: {desc}",
            f"In view: busy {window[1] * 100 / span:.1f}%",
            "Bar height is the fraction of time busy" +
            (", color is the frequency from low to high"
             if len(c.pstate) else ""),
        ])


class CStateRow(Row):
    """Row showing the idle states of a CPU."""
    def __init__(self, data: TimechartData, cpu: Cpu) -> None:
        super().__init__(f"  CPU {cpu.cpu} idle")
        self.data = data
        self.cpu = cpu

    def segments(self) -> SegmentList:
        return self.cpu.cstate

    def cells(self, t0: float, dt: float, width: int, colors: ThemeColors) -> List[Cell]:
        cov = self.cpu.cstate.coverage(t0, dt, width, 2)
        depth = self.cpu.cstate.coverage(t0, dt, width, 2, weight_fn=float)
        max_cstate = max(self.data.max_cstate, 1)
        result = []
        for x in range(width):
            frac = cov[x][1] / dt
            style = colors.idle_styles[0]
            if cov[x][1] > 0:
                avg = depth[x][1] / cov[x][1]
                style = colors.idle_styles[max(0, min(8, round(avg * 8 / max_cstate)))]
            result.append((bar_char(frac), style))
        return result

    def describe(self, time: float, t0: float, t1: float) -> str:
        c = self.cpu
        i = c.cstate.find(time)
        if i >= 0:
            desc = (f"C{c.cstate.data[i]} from {self.data.fmt_time(c.cstate.starts[i])} for "
                    f"{fmt_duration(c.cstate.ends[i] - c.cstate.starts[i])}")
        else:
            desc = "not idle"
        span = max(t1 - t0, 1)
        window = c.cstate.coverage(t0, span, 1, 2)[0]
        return "\n".join([
            f"[b]CPU {c.cpu} idle state[/b]: {desc}",
            f"In view: idle {window[1] * 100 / span:.1f}%",
            "Bar height is the fraction of time idle, darker colors are deeper C-states",
        ])


class FreqRow(Row):
    """Row showing the frequency of a CPU."""
    def __init__(self, data: TimechartData, cpu: Cpu) -> None:
        super().__init__(f"  CPU {cpu.cpu} freq")
        self.data = data
        self.cpu = cpu

    def segments(self) -> SegmentList:
        return self.cpu.pstate

    def cells(self, t0: float, dt: float, width: int, colors: ThemeColors) -> List[Cell]:
        cov = self.cpu.pstate.coverage(t0, dt, width, 2)
        weighted = self.cpu.pstate.coverage(t0, dt, width, 2, weight_fn=float)
        max_freq = max(self.data.max_freq, 1)
        min_freq = self.data.min_freq
        result = []
        for x in range(width):
            if cov[x][1] <= 0:
                result.append((" ", colors.freq_styles[0]))
                continue
            freq = weighted[x][1] / cov[x][1]
            frac = (freq - min_freq) / (max_freq - min_freq) if max_freq > min_freq else 1.0
            result.append((bar_char(max(freq / max_freq, 1 / 8)),
                           colors.freq_styles[max(0, min(8, round(frac * 8)))]))
        return result

    def describe(self, time: float, t0: float, t1: float) -> str:
        c = self.cpu
        i = c.pstate.find(time)
        if i >= 0:
            desc = (f"{c.pstate.data[i] / 1000:.0f} MHz from "
                    f"{self.data.fmt_time(c.pstate.starts[i])} for "
                    f"{fmt_duration(c.pstate.ends[i] - c.pstate.starts[i])}")
        else:
            desc = "unknown"
        return "\n".join([
            f"[b]CPU {c.cpu} frequency[/b]: {desc}",
            f"Range: {self.data.min_freq / 1000:.0f} - {self.data.max_freq / 1000:.0f} MHz",
            "Bar height is the frequency relative to the maximum",
        ])


class IoRow(Row):
    """Row showing the I/O syscalls of a task."""
    def __init__(self, data: TimechartData, task: Task) -> None:
        super().__init__(task.name())
        self.data = data
        self.task = task

    def label(self) -> str:
        return self.task.name()

    def segments(self) -> SegmentList:
        return self.task.io

    def cells(self, t0: float, dt: float, width: int, colors: ThemeColors) -> List[Cell]:
        errs = self.task.io.coverage(t0, dt, width, NUM_IOTYPES,
                                     weight_fn=lambda d: 1.0 if d[1] < 0 else 0.0)
        result = []
        for x, col in enumerate(self.task.io.coverage(t0, dt, width, NUM_IOTYPES)):
            total = sum(col)
            if total <= 0:
                result.append((" ", colors.io_styles[0]))
                continue
            iotype = col.index(max(col))
            styles = colors.io_err_styles if sum(errs[x]) > 0 else colors.io_styles
            result.append((bar_char(max(total / dt, 1 / 8)), styles[iotype]))
        return result

    def describe(self, time: float, t0: float, t1: float) -> str:
        task = self.task
        i = task.io.find(time)
        if i < 0:
            i = task.io.last_before(time)
            prefix = "last I/O"
        else:
            prefix = "in"
        if i >= 0:
            fd, ret = task.io.data[i]
            result = f"returned {ret}" if ret >= 0 else f"failed with error {-ret}"
            desc = (f"{prefix} {IOTYPE_NAMES[task.io.keys[i]]} fd={fd} at "
                    f"{self.data.fmt_time(task.io.starts[i])} for "
                    f"{fmt_duration(task.io.ends[i] - task.io.starts[i])}, {result}")
        else:
            desc = "no I/O"
        span = max(t1 - t0, 1)
        window = task.io.coverage(t0, span, 1, NUM_IOTYPES)[0]
        in_view = ", ".join(f"{IOTYPE_NAMES[t]} {window[t] * 100 / span:.1f}%"
                            for t in range(NUM_IOTYPES) if window[t] > 0)
        return "\n".join([
            f"[b]{escape(task.name())}[/b]: {desc}",
            f"In view: {in_view or 'no I/O'}",
            f"Total: {len(task.io)} I/O syscalls, {task.io_bytes} bytes",
        ])


@dataclass(frozen=True)
class TimeWindow:
    """The visible time range and the cursor, shared between views.

    Immutable so that it can be a reactive value, the methods return new windows.
    """
    first: int
    last: int
    start: float
    end: float
    cursor: float

    @staticmethod
    def whole(first: int, last: int) -> "TimeWindow":
        """A window showing the whole trace with the cursor at the start."""
        last = max(last, first + 1)
        return TimeWindow(first, last, first, last, first)

    def span(self) -> float:
        """Length of the visible time range."""
        return self.end - self.start

    def reset(self) -> "TimeWindow":
        """Show the whole trace."""
        return replace(self, start=self.first, end=self.last)

    def with_range(self, start: float, span: float) -> "TimeWindow":
        """Set the visible range clamped to the trace."""
        span = min(max(span, 100.0), self.last - self.first)
        start = min(max(start, self.first), self.last - span)
        return replace(self, start=start, end=start + span)

    def zoom(self, factor: float) -> "TimeWindow":
        """Zoom by factor keeping the cursor at the same position."""
        cursor = self.cursor
        if not self.start <= cursor <= self.end:
            cursor = (self.start + self.end) / 2
        rel = (cursor - self.start) / self.span()
        span = self.span() * factor
        return replace(self, cursor=cursor).with_range(cursor - rel * span, span)

    def pan(self, frac: float) -> "TimeWindow":
        """Pan the view by the fraction of the visible span."""
        delta = self.span() * frac
        win = self.with_range(self.start + delta, self.span())
        return replace(win, cursor=min(max(self.cursor + delta, win.start), win.end))

    def with_cursor(self, time: float) -> "TimeWindow":
        """Move the cursor, scrolling to keep it visible."""
        win = replace(self, cursor=min(max(time, self.first), self.last))
        if win.cursor < win.start:
            return win.with_range(win.cursor, win.span())
        if win.cursor >= win.end:
            return win.with_range(win.cursor - win.span() * 0.9, win.span())
        return win

    def extend(self, first: int, last: int) -> "TimeWindow":
        """Change the bounds of the trace as more of it is loaded.

        If the whole trace was visible then it still is, otherwise the visible
        range is kept.
        """
        last = max(last, first + 1)
        whole = self.start == self.first and self.end == self.last
        win = replace(self, first=first, last=last, cursor=min(max(self.cursor, first), last))
        return win.reset() if whole else win.with_range(self.start, self.span())


class TimelineView(ScrollView):
    """A scrollable view of rows against a time axis.

    Line 0 is a time ruler that stays at the top, the other lines are rows.
    """
    BINDINGS = [
        Binding("up,k", "row_up", "Up", show=False),
        Binding("down,j", "row_down", "Down", show=False),
        Binding("left,h", "cursor_left", "Cursor ←", key_display="←"),
        Binding("right,l", "cursor_right", "Cursor →", key_display="→"),
        Binding("shift+left,H", "pan_left", "Pan left", show=False),
        Binding("shift+right,L", "pan_right", "Pan right", show=False),
        Binding("plus,equals_sign", "zoom_in", "Zoom in", key_display="+"),
        Binding("minus", "zoom_out", "Zoom out", key_display="-"),
        Binding("0,escape", "zoom_reset", "Reset zoom", key_display="0"),
        Binding("n", "next_change", "Next change"),
        Binding("p", "prev_change", "Prev change"),
        Binding("pageup", "page_up", "Page up", show=False),
        Binding("pagedown", "page_down", "Page down", show=False),
    ]

    DEFAULT_CSS = """
    TimelineView {
        width: 100%;
        height: 1fr;
    }
    """

    class SelectionChanged(Message):
        """Posted when the selected row changes."""

    # Bound to TimechartApp.window, changing it repaints the view.
    window: reactive[TimeWindow] = reactive(TimeWindow.whole(0, 1))
    # Index of the selected row.
    selected: reactive[int] = reactive(0)

    def __init__(self, data: TimechartData, rows: List[Row], *pos_args, **kwargs) -> None:
        super().__init__(*pos_args, **kwargs)
        self.can_focus = True
        self.data = data
        self.rows = rows
        self.cache: Dict[int, List[Cell]] = {}
        self.cache_key: Tuple[float, float, int, Optional[ThemeColors]] = (0.0, 0.0, 0, None)
        self.label_style = Style()
        self.ruler_style = Style(dim=True)
        self.cursor_style = Style(reverse=True)

    def selected_row(self) -> Optional[Row]:
        """The selected row, if any."""
        if 0 <= self.selected < len(self.rows):
            return self.rows[self.selected]
        return None

    def timeline_width(self) -> int:
        """Number of columns used for the timeline."""
        return max(self.scrollable_content_region.width - LABEL_WIDTH, 1)

    def update_size(self) -> None:
        """Update the virtual size after rows change or the widget resizes."""
        self.virtual_size = Size(self.scrollable_content_region.width, len(self.rows) + 1)

    def on_mount(self) -> None:
        """Size the view when mounted."""
        self.update_size()

    def on_resize(self) -> None:
        """Size the view when resized."""
        self.update_size()
        self.refresh()

    def set_rows(self, rows: List[Row]) -> None:
        """Replace the rows, after sorting or as more data is loaded.

        The selected row is kept, if it moves watch_selected scrolls to it.
        """
        selected = self.selected_row()
        self.rows = rows
        self.cache.clear()
        self.update_size()
        self.selected = rows.index(selected) if selected in rows else 0
        self.refresh()

    def validate_selected(self, idx: int) -> int:
        """Keep the selection within the rows."""
        return max(0, min(idx, len(self.rows) - 1))

    def watch_selected(self) -> None:
        """Keep the selection visible and tell the app."""
        self.scroll_to_selected()
        self.post_message(self.SelectionChanged())

    def scroll_to_selected(self) -> None:
        """Scroll so that the selected row is visible below the ruler."""
        _, scroll_y = self.scroll_offset
        visible = max(self.scrollable_content_region.height - 1, 1)
        if self.selected < scroll_y:
            self.scroll_to(y=self.selected, animate=False)
        elif self.selected >= scroll_y + visible:
            self.scroll_to(y=self.selected - visible + 1, animate=False)

    def set_window(self, window: TimeWindow) -> None:
        """Change the window shared by all views, owned by the app."""
        app = self.app
        if isinstance(app, TimechartApp):
            app.window = window

    def ruler(self, width: int) -> Strip:
        """The time axis, labeled in seconds relative to the trace start."""
        win = self.window
        dt = win.span() / width
        tick = 14
        decimals = max(0, min(9, 1 - math.floor(math.log10(max(dt * tick, 1) / NSEC_PER_SEC))))
        chars = [" "] * width
        x = 0
        while x < width:
            label = f"|{(win.start + x * dt - self.data.first_time) / NSEC_PER_SEC:.{decimals}f}"
            if x + len(label) > width:
                break
            chars[x:x + len(label)] = list(label)
            x += max(tick, len(label) + 2)
        segments = [Segment(make_fixed_length_string("Time (s)", LABEL_WIDTH), self.ruler_style)]
        cursor_x = self.cursor_column(width)
        if 0 <= cursor_x < width:
            segments.append(Segment("".join(chars[:cursor_x]), self.ruler_style))
            segments.append(Segment("▼"))
            segments.append(Segment("".join(chars[cursor_x + 1:]), self.ruler_style))
        else:
            segments.append(Segment("".join(chars), self.ruler_style))
        return Strip(segments)

    def cursor_column(self, width: int) -> int:
        """Column of the cursor or -1 if not visible."""
        win = self.window
        if not win.start <= win.cursor < win.end:
            return -1
        return min(int((win.cursor - win.start) * width / win.span()), width - 1)

    def render_line(self, y: int) -> Strip:
        """Render the ruler or a row."""
        width = self.timeline_width()
        if y == 0:
            return self.ruler(width)
        _, scroll_y = self.scroll_offset
        idx = scroll_y + y - 1
        if idx >= len(self.rows):
            return Strip.blank(self.scrollable_content_region.width)

        colors = self.app.theme_colors if isinstance(self.app, TimechartApp) else \
            ThemeColors(self.app.theme_variables)
        win = self.window
        key = (win.start, win.end, width, colors)
        if key != self.cache_key:
            self.cache.clear()
            self.cache_key = key
        row = self.rows[idx]
        with self.data.lock:
            cells = self.cache.get(idx)
            if cells is None:
                cells = row.cells(win.start, win.span() / width, width, colors)
                self.cache[idx] = cells
            label = row.label()

        label_style = colors.selected_style if idx == self.selected else self.label_style
        segments = [Segment(make_fixed_length_string(label, LABEL_WIDTH - 1) + " ",
                            label_style)]
        cursor_x = self.cursor_column(width)
        # Merge runs of cells with the same style into one segment.
        text = ""
        style: Optional[Style] = None
        for x, (char, cell_style) in enumerate(cells):
            if x == cursor_x:
                cell_style = cell_style + self.cursor_style
            if cell_style is not style and text:
                segments.append(Segment(text, style))
                text = ""
            text += char
            style = cell_style
        if text:
            segments.append(Segment(text, style))
        return Strip(segments)

    def column_time(self, x: int) -> float:
        """Time at the center of the column containing screen offset x."""
        width = self.timeline_width()
        return self.window.start + (x - LABEL_WIDTH + 0.5) * self.window.span() / width

    def on_click(self, click: events.Click) -> None:
        """Select the clicked row and move the cursor to the clicked time."""
        if click.x >= LABEL_WIDTH:
            self.set_window(self.window.with_cursor(self.column_time(click.x)))
        if click.y > 0:
            _, scroll_y = self.scroll_offset
            self.selected = scroll_y + click.y - 1

    def action_row_up(self) -> None:
        """Select the previous row."""
        self.selected -= 1

    def action_row_down(self) -> None:
        """Select the next row."""
        self.selected += 1

    def action_page_up(self) -> None:
        """Select a row a page up."""
        self.selected -= max(self.scrollable_content_region.height - 2, 1)

    def action_page_down(self) -> None:
        """Select a row a page down."""
        self.selected += max(self.scrollable_content_region.height - 2, 1)

    def action_cursor_left(self) -> None:
        """Move the cursor one column left."""
        win = self.window
        self.set_window(win.with_cursor(win.cursor - win.span() / self.timeline_width()))

    def action_cursor_right(self) -> None:
        """Move the cursor one column right."""
        win = self.window
        self.set_window(win.with_cursor(win.cursor + win.span() / self.timeline_width()))

    def action_pan_left(self) -> None:
        """Pan a quarter of the view left."""
        self.set_window(self.window.pan(-0.25))

    def action_pan_right(self) -> None:
        """Pan a quarter of the view right."""
        self.set_window(self.window.pan(0.25))

    def action_zoom_in(self) -> None:
        """Halve the visible time range around the cursor."""
        self.set_window(self.window.zoom(0.5))

    def action_zoom_out(self) -> None:
        """Double the visible time range around the cursor."""
        self.set_window(self.window.zoom(2))

    def action_zoom_reset(self) -> None:
        """Show the whole trace."""
        self.set_window(self.window.reset())

    def action_next_change(self) -> None:
        """Move the cursor to the next change in the selected row."""
        row = self.selected_row()
        if row:
            # Skip changes within the cursor's column.
            win = self.window
            with self.data.lock:
                time = row.segments().next_change(
                    win.cursor + win.span() / self.timeline_width() / 2)
            if time is not None:
                self.set_window(win.with_cursor(time))

    def action_prev_change(self) -> None:
        """Move the cursor to the previous change in the selected row."""
        row = self.selected_row()
        if row:
            win = self.window
            with self.data.lock:
                time = row.segments().prev_change(
                    win.cursor - win.span() / self.timeline_width() / 2)
            if time is not None:
                self.set_window(win.with_cursor(time))


class TimechartApp(App):
    """A Textual application to display a timechart."""
    TITLE = "perf timechart"

    BINDINGS = [
        Binding("s", "sort", "Sort tasks", tooltip="Cycle task sort order"),
        Binding("w", "goto_waker", "Go to waker",
                tooltip="Select the task that last woke the selected task"),
        Binding(key="^q", action="quit", description="Quit", tooltip="Quit the app"),
    ]

    CSS = """
    TabbedContent, TabbedContent > ContentSwitcher, TabPane {
        height: 1fr;
    }
    TabPane {
        padding: 0;
    }
    .legend {
        height: 1;
        padding: 0 1;
    }
    #details {
        height: 8;
        border: round $primary;
        padding: 0 1;
    }
    """

    SORT_ORDERS = ["run time", "tid", "name", "first run"]

    # The visible time range and cursor, bound to each TimelineView's window.
    window: reactive[TimeWindow] = reactive(TimeWindow.whole(0, 1), init=False)
    # Index into SORT_ORDERS for the tasks view.
    sort_order: reactive[int] = reactive(0, init=False)

    def __init__(self, input_name: str, power_only: bool, tasks_only: bool,
                 filters: Sequence[str], data: Optional[TimechartData] = None) -> None:
        """Create the app, if data isn't given it is loaded from input_name.

        While the data is loading the views show what has been read so far.
        """
        super().__init__()
        self.input_name = input_name
        self.power_only = power_only
        self.tasks_only = tasks_only
        self.filters = filters
        # The data being displayed, possibly still being loaded.
        self.data = data if data else TimechartData()
        self.loaded = data is not None
        # The data being loaded in a background thread.
        self.loading: Optional[TimechartData] = None
        self.tasks: List[Task] = []
        self.io_tasks: List[Task] = []
        # Rows keyed by type and CPU or tid. They are reused as the data loads
        # so that the views can keep the selected row.
        self.row_cache: Dict[Tuple[str, int], Row] = {}
        self.io_tab_shown = True
        # Does the summary table need updating before it is next shown?
        self.summary_stale = True
        self.theme_colors = ThemeColors(self.theme_variables)

    def cached_row(self, cls: Callable[[TimechartData, Any], Row], num: int, obj: Any) -> Row:
        """Find or create the row of type cls for the CPU or task obj numbered num."""
        key = (getattr(cls, "__name__", ""), num)
        row = self.row_cache.get(key)
        if row is None:
            row = cls(self.data, obj)
            self.row_cache[key] = row
        return row

    def cpu_rows(self) -> List[Row]:
        """Rows for the CPUs tab, the data's lock must be held."""
        rows: List[Row] = []
        for cpu in sorted(self.data.cpus.values(), key=lambda c: c.cpu):
            if len(cpu.run) or len(cpu.cstate) or len(cpu.pstate):
                rows.append(self.cached_row(CpuRow, cpu.cpu, cpu))
            if len(cpu.cstate):
                rows.append(self.cached_row(CStateRow, cpu.cpu, cpu))
            if len(cpu.pstate):
                rows.append(self.cached_row(FreqRow, cpu.cpu, cpu))
        return rows

    def task_rows(self) -> List[Row]:
        """Rows for the tasks tab in the current sort order, the data's lock must be held."""
        order = self.SORT_ORDERS[self.sort_order]
        if order == "run time":
            tasks = sorted(self.tasks, key=lambda t: -t.totals[STATE_RUNNING])
        elif order == "name":
            tasks = sorted(self.tasks, key=lambda t: (t.comm, t.tid))
        elif order == "first run":
            tasks = sorted(self.tasks, key=lambda t: t.segs.starts[0])
        else:
            tasks = self.tasks
        return [self.cached_row(TaskRow, t.tid, t) for t in tasks]

    def compose(self) -> ComposeResult:
        """Composes the user interface of the application."""
        yield Header()
        with TabbedContent():
            if not self.tasks_only:
                with TabPane("CPUs", id="cpus"):
                    yield Static(id="cpus_legend", classes="legend")
                    yield TimelineView(self.data, [],
                                       id="cpus_view").data_bind(TimechartApp.window)
            if not self.power_only:
                with TabPane("Tasks", id="tasks"):
                    yield Static(id="tasks_legend", classes="legend")
                    yield TimelineView(self.data, [],
                                       id="tasks_view").data_bind(TimechartApp.window)
                # Hidden until there are tasks doing I/O.
                with TabPane("I/O", id="io"):
                    yield Static(id="io_legend", classes="legend")
                    yield TimelineView(self.data, [],
                                       id="io_view").data_bind(TimechartApp.window)
                with TabPane("Summary", id="summary"):
                    yield DataTable(id="summary_table", cursor_type="row")
        yield Static(id="details")
        yield Footer()

    def on_mount(self) -> None:
        """Show the data, loading it in the background if it wasn't given."""
        self.theme_changed_signal.subscribe(self, self.on_theme_changed)
        self.update_legends()
        self.update_views()
        view = self.active_view()
        if view:
            view.focus()
        if self.loaded:
            self.finish_loading()
        else:
            self.sub_title = f"Loading {self.input_name}"
            self.loading = self.data
            self.load_events()

    def on_theme_changed(self, _theme: Any) -> None:
        """Update colors, legends and timeline views when the theme changes."""
        self.theme_colors = ThemeColors(self.theme_variables)
        self.update_legends()
        for view in self.query(TimelineView):
            view.refresh()

    def update_legends(self) -> None:
        """Update the legend strings using the active theme's colors."""
        c = self.theme_colors
        with self.data.lock:
            cpus_legend = self.cpu_legend()
        for static in self.query("#cpus_legend").results(Static):
            static.update(cpus_legend)
        tasks_legend = (f"[{c.running}]█[/] running   "
                        f"[on {c.waiting}] [/] runnable   "
                        f"[on {c.blocked}] [/] blocked   "
                        "(bar height is the fraction of time running)")
        for static in self.query("#tasks_legend").results(Static):
            static.update(tasks_legend)
        io_legend = ("   ".join(f"[{c.io[t]}]█[/] {IOTYPE_NAMES[t]}"
                                for t in range(NUM_IOTYPES)) +
                     "   (underline is an error)")
        for static in self.query("#io_legend").results(Static):
            static.update(io_legend)

    @work(thread=True, exclusive=True)
    def load_events(self) -> None:
        """Read the perf.data file in a thread so the UI stays responsive."""
        data = self.loading or self.data
        self.loading = data

        def progress() -> None:
            try:
                # Blocks until the UI has shown the data read so far.
                self.call_from_thread(self.update_progress, data)
            except RuntimeError:
                # The app is no longer running.
                data.cancelled = True

        data.progress = progress
        data.start_progress = data.last_progress = monotonic()
        try:
            read_events(data, self.input_name)
        except LoadCancelled:
            return
        except (OSError, ValueError, RuntimeError) as e:
            self.call_from_thread(self.exit, None, 1,
                                  f"Error processing {self.input_name}: {e}")
            return
        finally:
            data.progress = None
            self.loading = None
        if data.cancelled:
            return
        if not data.has_events():
            self.call_from_thread(self.exit, None, 1,
                                  f"Error: no scheduler, power or I/O events found in "
                                  f"{self.input_name}.\n"
                                  "Record them with 'perf timechart record'.")
            return
        self.call_from_thread(self.data_loaded)

    def data_loaded(self) -> None:
        """Called on the UI thread when loading completes."""
        self.loaded = True
        self.finish_loading()

    def update_progress(self, data: TimechartData) -> None:
        """Show how much of the file has been processed and the data so far."""
        self.sub_title = (f"Loading {self.input_name}: {data.nr_samples:,} samples, "
                          f"{fmt_duration(data.last_time - data.first_time)} of trace")
        self.update_views()

    def finish_loading(self) -> None:
        """Show all of the data and a summary of it."""
        self.update_views()
        views = self.query("#cpus_view")
        if views and not views.first(TimelineView).rows:
            self.query_one(TabbedContent).hide_tab("cpus")
        task_views = self.query("#tasks_view")
        if task_views and not task_views.first(TimelineView).rows and self.io_tasks:
            self.query_one(TabbedContent).hide_tab("tasks")
        nr_tasks = len(self.tasks) or len(self.io_tasks)
        self.sub_title = (f"{self.input_name}: "
                          f"{fmt_duration(self.data.last_time - self.data.first_time)}, "
                          f"{len(self.data.cpus)} CPUs, {nr_tasks} tasks")

    def cancel_loading(self) -> None:
        """Stop a background load, the worker notices at the next progress interval."""
        loading = self.loading
        if loading is not None:
            loading.cancelled = True

    async def action_quit(self) -> None:
        """Quit, stopping any background load."""
        self.cancel_loading()
        await super().action_quit()

    def on_unmount(self) -> None:
        """Stop any background load when the app exits."""
        self.cancel_loading()

    def cpu_legend(self) -> str:
        """Legend for the CPUs tab, saying which power events are missing.

        The data's lock must be held.
        """
        c = self.theme_colors
        data = self.data
        has_pstate = any(len(cpu.pstate) for cpu in data.cpus.values())
        has_cstate = any(len(cpu.cstate) for cpu in data.cpus.values())
        parts = [f"[{c.running}]█[/] busy"]
        if has_pstate and data.max_freq > data.min_freq:
            parts.append(f"[{c.freq_low}]█[/]→[{c.freq_high}]█[/] frequency")
        elif has_pstate:
            parts.append(f"frequency constant at {data.max_freq / 1000:.0f} MHz")
        if has_cstate:
            parts.append(f"[{c.idle_light}]█[/]→[{c.idle_dark}]█[/] idle state")
        missing = []
        if not has_pstate:
            missing.append("power:cpu_frequency")
        if not has_cstate:
            missing.append("power:cpu_idle")
        if missing:
            parts.append(f"[dim](no {' or '.join(missing)} events)[/dim]")
        return "   ".join(parts)

    def update_views(self) -> None:
        """Update the views from the data, which may still be loading."""
        data = self.data
        with data.lock:
            first, last = data.first_time, data.last_time
            self.tasks = data.sched_tasks(self.filters)
            self.io_tasks = data.io_tasks(self.filters)
            all_rows = {
                "cpus_view": self.cpu_rows(),
                "tasks_view": self.task_rows(),
                "io_view": [self.cached_row(IoRow, t.tid, t) for t in self.io_tasks],
            }
            legend = self.cpu_legend()
        for view_id, rows in all_rows.items():
            for view in self.query(f"#{view_id}").results(TimelineView):
                view.set_rows(rows)
        for static in self.query("#cpus_legend").results(Static):
            static.update(legend)
        if self.query("#io") and bool(self.io_tasks) != self.io_tab_shown:
            self.io_tab_shown = bool(self.io_tasks)
            tabbed = self.query_one(TabbedContent)
            if self.io_tab_shown:
                tabbed.show_tab("io")
            else:
                tabbed.hide_tab("io")
        if first or last:
            self.window = self.window.extend(first, last)
        self.summary_stale = True
        tabs = self.query(TabbedContent)
        if tabs and tabs.first(TabbedContent).active == "summary":
            self.update_summary()
        self.update_details()

    def update_summary(self) -> None:
        """Fill in the summary table keeping the selected task."""
        tables = self.query("#summary_table")
        if not tables:
            return
        self.summary_stale = False
        table = tables.first(DataTable)
        if not table.columns:
            table.add_columns("Task", "TID", "Running", "Waiting", "Blocked",
                              "Switches", "Wakeups", "I/O bytes")
        selected = None
        if table.row_count:
            try:
                selected = table.coordinate_to_cell_key(table.cursor_coordinate).row_key
            except CellDoesNotExist:
                pass
        summary_tasks = self.tasks if self.tasks else self.io_tasks
        with self.data.lock:
            rows = [(task.comm, task.tid,
                     fmt_duration(task.totals[STATE_RUNNING]),
                     fmt_duration(task.totals[STATE_WAITING]),
                     fmt_duration(task.totals[STATE_BLOCKED]),
                     task.switches, len(task.wakeups), task.io_bytes)
                    for task in sorted(summary_tasks,
                                       key=lambda t: (-t.totals[STATE_RUNNING], -t.io_bytes))]
        table.clear()
        for row in rows:
            table.add_row(*row, key=str(row[1]))
        if selected is not None:
            try:
                table.move_cursor(row=table.get_row_index(selected), animate=False)
            except RowDoesNotExist:
                pass

    def active_view(self) -> Optional[TimelineView]:
        """The timeline view in the active tab, if any."""
        tabs = self.query(TabbedContent)
        if not tabs:
            return None
        active = tabs.first(TabbedContent).active
        views = self.query(f"#{active}_view") if active else None
        return views.first(TimelineView) if views else None

    def watch_window(self) -> None:
        """The bound views repaint themselves, update the details."""
        self.update_details()

    @on(TimelineView.SelectionChanged)
    def on_selection_changed(self) -> None:
        """Describe the newly selected row."""
        self.update_details()

    def update_details(self) -> None:
        """Describe the selected row at the cursor."""
        details = self.query("#details")
        if not details:
            return
        win = self.window
        text = (f"Cursor [b]{self.data.fmt_time(win.cursor)}[/b]   "
                f"View {self.data.fmt_time(win.start)} - {self.data.fmt_time(win.end)} "
                f"({fmt_duration(win.span())})\n")
        view = self.active_view()
        row = view.selected_row() if view else None
        if row:
            with self.data.lock:
                text += row.describe(win.cursor, win.start, win.end)
        elif view is None:
            text += "Select a task and press enter to show it in the Tasks timeline"
        details.first(Static).update(text)

    @on(TabbedContent.TabActivated)
    def on_tab_activated(self) -> None:
        """Focus the view in the newly active tab."""
        view = self.active_view()
        if view:
            view.focus()
        else:
            if self.summary_stale:
                self.update_summary()
            tables = self.query("#summary_table")
            if tables:
                tables.first(DataTable).focus()
        self.update_details()

    @on(DataTable.RowSelected)
    def on_row_selected(self, event: DataTable.RowSelected) -> None:
        """Show the selected summary task in the tasks timeline."""
        if event.row_key.value is not None:
            self.goto_task(int(event.row_key.value), None)

    def goto_task(self, tid: int, time: Optional[float]) -> None:
        """Select the task in the tasks or I/O view, optionally moving the cursor."""
        for tab_id, view_id, row_type in (("tasks", "#tasks_view", TaskRow),
                                          ("io", "#io_view", IoRow)):
            views = self.query(view_id)
            if not views:
                continue
            view = views.first(TimelineView)
            for idx, row in enumerate(view.rows):
                if isinstance(row, row_type) and row.task.tid == tid:
                    self.query_one(TabbedContent).active = tab_id
                    if time is not None:
                        self.window = self.window.with_cursor(time)
                    view.selected = idx
                    view.focus()
                    return
        self.notify(f"Task {tid} isn't shown in the tasks view", severity="warning")

    def action_sort(self) -> None:
        """Cycle the sort order of the tasks view."""
        self.sort_order = (self.sort_order + 1) % len(self.SORT_ORDERS)

    def watch_sort_order(self) -> None:
        """Re-sort the tasks view."""
        views = self.query("#tasks_view")
        if not views:
            return
        with self.data.lock:
            rows = self.task_rows()
        views.first(TimelineView).set_rows(rows)
        self.notify(f"Tasks sorted by {self.SORT_ORDERS[self.sort_order]}")
        self.update_details()

    def action_goto_waker(self) -> None:
        """Select the task that woke the selected task before the cursor."""
        view = self.active_view()
        row = view.selected_row() if view else None
        if not isinstance(row, (TaskRow, IoRow)):
            return
        with self.data.lock:
            wakeups = row.task.wakeups
            idx = bisect.bisect_right(wakeups, (self.window.cursor, sys.maxsize)) - 1
            wake_time, waker = wakeups[idx] if idx >= 0 else (0, 0)
        if idx < 0:
            self.notify("No wakeup before the cursor", severity="warning")
            return
        if waker <= 0:
            self.notify(f"Woken by {self.data.task_name(waker)} at "
                        f"{self.data.fmt_time(wake_time)}")
            return
        self.goto_task(waker, wake_time)


def main() -> None:
    """Parse arguments, read the perf.data file and run the app."""
    parser = argparse.ArgumentParser(
        description="Interactive timechart of CPU, task and I/O activity.")
    parser.add_argument("-i", "--input", default="perf.data", help="input perf.data file")
    parser.add_argument("-P", "--power-only", action="store_true",
                        help="only show CPU power information")
    parser.add_argument("-T", "--tasks-only", action="store_true",
                        help="only show task information")
    parser.add_argument("-p", "--process", action="append", default=[],
                        help="only show processes with the given name or PID, may be repeated")
    parser.add_argument("--dump", action="store_true",
                        help="print a text summary rather than running the interactive UI")
    args = parser.parse_args()

    if args.power_only and args.tasks_only:
        print("Error: -P and -T options cannot be used at the same time.", file=sys.stderr)
        sys.exit(1)

    if args.input == "-":
        if not args.dump:
            # The interactive UI reads the keyboard from stdin.
            print("Error: reading perf.data from stdin requires --dump.", file=sys.stderr)
            sys.exit(1)
    elif not os.path.exists(args.input):
        print(f"Error: {args.input} not found. (try 'perf timechart record' first)",
              file=sys.stderr)
        sys.exit(1)

    if args.dump:
        data = load_data(args.input)
        data.dump(args.power_only, args.tasks_only, args.process)
        return

    # The app starts immediately and loads the data in the background.
    app = TimechartApp(args.input, args.power_only, args.tasks_only, args.process)
    app.run()
    sys.exit(app.return_code or 0)


def read_events(data: TimechartData, input_name: str) -> None:
    """Process the events in input_name into data, raising on errors."""
    if data.cancelled:
        raise LoadCancelled()
    try:
        data.session = perf.session(perf.data(input_name), sample=data.process_event)
        data.session.process_events()
        data.finish()
    finally:
        # Break the reference cycle between the session and the callback.
        data.session = None


def load_data(input_name: str) -> TimechartData:
    """Read the perf.data file exiting on errors or if there's nothing to show."""
    if input_name != "-" and not os.path.exists(input_name):
        print(f"Error: {input_name} not found. (try 'perf timechart record' first)",
              file=sys.stderr)
        sys.exit(1)

    data = TimechartData()
    try:
        read_events(data, input_name)
    except (OSError, ValueError, RuntimeError) as e:
        print(f"Error processing {input_name}: {e}", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        data.finish()

    if not data.has_events():
        print(f"Error: no scheduler, power or I/O events found in {input_name}.\n"
              "Record them with 'perf timechart record'.", file=sys.stderr)
        sys.exit(1)
    return data


if __name__ == "__main__":
    main()