strace timeline

strace_timeline.png

strace_timeline.py

# python strace_timeline.py -e execve,unlink,symlink,openat,lstat python ./x.py > timeline.json
# https://ui.perfetto.dev

import sys
import re
import subprocess
import tempfile
import json

with tempfile.NamedTemporaryFile() as f:
    subprocess.run(["/usr/bin/strace",
                    "-o", f.name,
                    "-f",
                    "-Tttt",
                    *sys.argv[1:]])

    data = f.read().decode().strip()

events = []

thread_states = {}

for line in data.split("\n"):
    match = re.match(r"^(\d+) +(\d+\.\d+) +(.+)$", line)
    assert match

    pid = int(match.group(1))
    time = int(float(match.group(2)) * 1000000)
    syscall_info = match.group(3)

    syscall_name = re.search(r"\w+", syscall_info).group(0)
    args = {}
    resumed = False
    if "unfinished" in syscall_info:
        thread_states[pid] = (time, syscall_info)
        continue
    elif "resumed" in syscall_info:
        resumed = True
        resume_time = time
        time, unfinished_info = thread_states[pid]

    match = re.match(r"^(.+) +<(\d+\.\d+)>$", syscall_info)
    if match is not None:
        syscall_info = match.group(1)
        duration = int(float(match.group(2)) * 1000000)
    else:
        duration = 0

    if not resumed:
        events.append({
            "pid": pid,
            "tid": pid,
            "ts": time,
            "dur": duration,
            "ph": "X" if duration else "I",
            "name": syscall_name,
            "args": {
                "info": syscall_info
            },
        })
    else:
        events.append({
            "pid": pid,
            "tid": pid,
            "ts": time,
            "dur": duration,
            "ph": "X",
            "name": syscall_name,
            "args": {
                "unfinished": unfinished_info,
                "resumed": syscall_info,
                "resume_after": resume_time - time,
            },
        })

print(json.dumps({"traceEvents": events}))