simpleperf report数据chrome trace json格式

simpleperf2chrome.py

import re
import json
import sys
from datetime import datetime


def parse_simpleperf_line(line):
    """解析simpleperf报告中的单行数据"""
    # 解析样本行
    sample_pattern = re.compile(
        r'^(?P<thread_comm>\S+)\t(?P<pid>\d+)/(?P<tid>\d+)\s+\[(?P<cpu>\d+)\]\s+'
        r'(?P<sec>\d+)\.(?P<usec>\d+):\s+(?P<period>\d+)\s+(?P<event_name>\S+):$'
    )
    match = sample_pattern.match(line)
    if match:
        return {
            'type': 'sample',
            'thread_comm': match.group('thread_comm'),
            'pid': int(match.group('pid')),
            'tid': int(match.group('tid')),
            'cpu': int(match.group('cpu')),
            'timestamp': float(f"{match.group('sec')}.{match.group('usec')}"),
            'period': int(match.group('period')),
            'event_name': match.group('event_name')
        }

    # 解析调用链条目
    callchain_pattern = re.compile(
        r'^\t(?P<ip>[0-9a-fA-Fx]+)\s+(?P<symbol_name>.*?)\s+\((?P<dso_name>.*?)\)$'
    )
    match = callchain_pattern.match(line)
    if match:
        return {
            'type': 'callchain_entry',
            'ip': match.group('ip'),
            'symbol_name': match.group('symbol_name').strip(),
            'dso_name': match.group('dso_name').strip()
        }

    return None


def parse_simpleperf_report(input_data):
    """解析整个simpleperf报告"""
    samples = []
    current_sample = None
    current_callchain = []

    for line in input_data.split('\n'):
        line = line.strip()
        if not line:
            # 空行表示一个样本的结束
            if current_sample:
                current_sample['callchain'] = current_callchain
                samples.append(current_sample)
                current_sample = None
                current_callchain = []
            continue

        parsed = parse_simpleperf_line(line)
        if parsed:
            if parsed['type'] == 'sample':
                # 新样本开始，如果有当前样本则保存
                if current_sample:
                    current_sample['callchain'] = current_callchain
                    samples.append(current_sample)
                    current_callchain = []
                current_sample = parsed
            elif parsed['type'] == 'callchain_entry' and current_sample:
                current_callchain.append(parsed)

    # 处理最后一个样本
    if current_sample:
        current_sample['callchain'] = current_callchain
        samples.append(current_sample)

    return samples


def convert_to_chrome_trace(samples):
    """将解析后的simpleperf数据转换为Chrome Trace格式"""
    trace_events = []
    metadata = {
        "name": "simpleperf to Chrome Trace conversion",
        "startTime": int(samples[0]['timestamp'] * 1e6) if samples else 0,
        "endTime": int(samples[-1]['timestamp'] * 1e6) if samples else 0,
        "displayTimeUnit": "ns"
    }

    # 添加进程和线程元数据
    processes = {}
    for sample in samples:
        pid = sample['pid']
        tid = sample['tid']
        comm = sample['thread_comm']

        if pid not in processes:
            processes[pid] = {
                "name": comm,
                "threads": {}
            }

        if tid not in processes[pid]['threads']:
            processes[pid]['threads'][tid] = {
                "name": comm
            }

    # 添加进程元数据事件
    for pid, proc_info in processes.items():
        trace_events.append({
            "name": "process_name",
            "ph": "M",
            "pid": pid,
            "tid": 0,
            "args": {
                "name": proc_info["name"]
            }
        })

        # 添加线程元数据事件
        for tid, thread_info in proc_info['threads'].items():
            trace_events.append({
                "name": "thread_name",
                "ph": "M",
                "pid": pid,
                "tid": tid,
                "args": {
                    "name": thread_info["name"]
                }
            })

    # 添加样本事件
    for i, sample in enumerate(samples):
        # 计算事件持续时间（使用下一个样本的时间作为当前样本的结束时间）
        duration = 0
        if i < len(samples) - 1:
            duration = (samples[i + 1]['timestamp'] - sample['timestamp']) * 1e6

        # 创建主事件
        event = {
            "name": sample['event_name'],
            "ph": "X",  # 完整事件
            "ts": sample['timestamp'] * 1e6,  # 转换为微秒
            "dur": duration,
            "pid": sample['pid'],
            "tid": sample['tid'],
            "cpu": sample['cpu'],
            "args": {
                "period": sample['period'],
                "callchain": [
                    f"{entry['symbol_name']} ({entry['dso_name']})"
                    for entry in sample['callchain']
                ]
            }
        }
        trace_events.append(event)

        # 为调用链中的每个函数添加事件
        # 我们使用嵌套的X事件来表示调用栈
        prev_ts = sample['timestamp'] * 1e6
        for j, entry in enumerate(sample['callchain']):
            # 分配一小部分时间给每个调用链条目
            entry_duration = duration / (len(sample['callchain']) + 1) if duration > 0 else 1

            call_event = {
                "name": entry['symbol_name'],
                "ph": "X",
                "ts": prev_ts + entry_duration,
                "dur": entry_duration,
                "pid": sample['pid'],
                "tid": sample['tid'],
                "cpu": sample['cpu'],
                "args": {
                    "ip": entry['ip'],
                    "dso_name": entry['dso_name']
                }
            }
            trace_events.append(call_event)
            prev_ts = call_event["ts"]

    return {
        "traceEvents": trace_events,
        "metadata": metadata
    }


def main():
    """主函数：读取输入，转换格式，输出结果"""
    if len(sys.argv) < 3:
        print("用法: python simpleperf_to_chrome_trace.py <输入文件> <输出文件>")
        print("   或: python simpleperf_to_chrome_trace.py - <输出文件>  # 从标准输入读取")
        sys.exit(1)

    input_path = sys.argv[1]
    output_path = sys.argv[2]

    # 读取输入数据
    try:
        if input_path == '-':
            print("从标准输入读取数据...")
            input_data = sys.stdin.read()
        else:
            with open(input_path, 'r') as f:
                input_data = f.read()
    except Exception as e:
        print(f"读取输入文件失败: {e}")
        sys.exit(1)

    # 解析simpleperf报告
    print("解析simpleperf报告...")
    samples = parse_simpleperf_report(input_data)
    print(f"解析完成，共发现 {len(samples)} 个样本")

    if not samples:
        print("没有找到有效的样本数据")
        sys.exit(1)

    # 转换为Chrome Trace格式
    print("转换为Chrome Trace格式...")
    chrome_trace = convert_to_chrome_trace(samples)

    # 保存结果
    try:
        with open(output_path, 'w') as f:
            json.dump(chrome_trace, f, indent=2)
        print(f"转换完成，结果已保存到 {output_path}")
        print("可以在Chrome浏览器中打开 chrome://tracing 并加载此文件查看结果")
    except Exception as e:
        print(f"保存输出文件失败: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()