===== Python 火焰图分析 =====
===== 使用 py-spy 生成火焰图定位 CPU 热点 =====

"""
火焰图 (Flame Graph) 是一种可视化 CPU 时间分布的工具。
py-spy 是一个采样 profiler，无需修改代码即可生成火焰图。

安装: pip install py-spy
"""

import time
import random
import math


# ===== 1. 模拟 CPU 密集型任务 =====

def calculate_primes(limit: int) -> list:
"""
计算素数，典型的 CPU 密集操作。
这是火焰图中应当出现的 "热点" 函数。
"""
primes = []
for num in range(2, limit):
is_prime = True
for divisor in range(2, int(num ** 0.5) + 1):
if num % divisor == 0:
is_prime = False
break
if is_prime:
primes.append(num)
return primes


# ===== 2. 模拟 IO 密集型与 CPU 混合任务 =====

def process_data_chunk(chunk_id: int, size: int) -> float:
"""
处理数据块，包含 CPU 计算和模拟 IO 等待。
"""
time.sleep(0.005) # 模拟 IO 等待
result = 0.0
for i in range(size):
result += math.sin(i) * math.cos(i)
result += math.sqrt(i + 1)
return result


def parallel_simulation(workers: int):
"""
模拟多任务并行处理，每个任务独立计算。
"""
results = []
for wid in range(workers):
res = process_data_chunk(wid, 5000)
results.append(res)
return sum(results)


# ===== 3. 递归函数调用分析 =====

def recursive_compute(n: int, depth: int = 0) -> int:
"""
递归计算函数，火焰图会显示深层调用栈。
"""
if depth > 10:
return n
# 每一层做一些计算
total = 0
for i in range(n):
total += i * i
# 递归调用
return total + recursive_compute(n // 2, depth + 1)


# ===== 4. 复杂数据处理管道 =====

def transform_step(data: list) -> list:
"""管道第一步：数学变换"""
return [x ** 2 - 3 * x + 1 for x in data]


def filter_step(data: list) -> list:
"""管道第二步：条件过滤"""
return [x for x in data if x > 100]


def sort_step(data: list) -> list:
"""管道第三步：排序"""
return sorted(data, reverse=True)


def pipeline(data: list) -> list:
"""
完整数据处理管道。
火焰图中可以看到每一步的耗时占比。
"""
step1 = transform_step(data)
step2 = filter_step(step1)
step3 = sort_step(step2)
return step3


# ===== 5. 长时间运行的主逻辑 =====

def heavy_workload():
"""
主工作负载函数，火焰图的核心分析目标。
它组合调用多个子函数，形成调用栈层次。
"""
print("[WORK] 开始繁重计算任务...")

# 任务 1：素数计算（CPU 密集）
primes = calculate_primes(5000)
print(f"[WORK] 找到 {len(primes)} 个素数")

# 任务 2：并行模拟（混合型）
sim_result = parallel_simulation(10)
print(f"[WORK] 模拟结果: {sim_result:.2f}")

# 任务 3：递归计算
rec_result = recursive_compute(100)
print(f"[WORK] 递归结果: {rec_result}")

# 任务 4：数据处理管道
raw_data = [random.randint(1, 1000) for _ in range(10000)]
processed = pipeline(raw_data)
print(f"[WORK] 管道处理了 {len(processed)} 个元素")


def background_overhead():
"""
后台额外开销函数，模拟不定时触发的小任务。
这类函数在火焰图中呈现为较窄的柱子。
"""
for _ in range(50):
_ = [math.sqrt(random.random()) for __ in range(200)]
time.sleep(0.001)


# ===== 6. py-spy 使用方式 =====

"""
py-spy 的常用命令:

# 1. 实时采样并生成火焰图 SVG
py-spy record -o flamegraph.svg --duration 30 -- python 本文件.py

# 2. 实时查看调用栈（top 风格）
py-spy top -- python 本文件.py

# 3. 附加到正在运行的进程
py-spy record -o flame.svg --pid 12345

# 4. 指定采样频率（默认 100，即每秒 100 次）
py-spy record -o flame.svg --rate 50 -- python 本文件.py

# 5. 只分析特定线程
py-spy record -o flame.svg --subprocesses -- python 本文件.py

火焰图解读:
- X 轴: 采样总数，不代表时间流逝
- Y 轴: 调用栈深度，顶部是实际执行函数
- 宽度: 函数占用 CPU 时间的比例
- 颜色: 通常随机分配，无特殊含义
- 从下往上看: 程序的执行路径

关键分析方法:
1. 找最宽的 "平顶": 这是最耗 CPU 的热点函数
2. 看调用路径: 追溯宽条是如何被调用的
3. 山峰的形状:
- 平坦宽阔: 循环或重复调用
- 高耸细窄: 深递归或深层调用链
- 锯齿状: 频繁的上下文切换
4. 关注那些出乎意料的宽条，它们可能是性能问题
"""


# ===== 7. 使用 py-spy 记录采样到文件 =====

def save_sampling_data():
"""
在被测脚本内部输出提示信息。
实际采样由 py-spy record 命令在外部完成。
"""
print("[PY-SPY] py-spy 正在采样中，请等待完成...")
print("[PY-SPY] 采样结束后将生成 SVG 火焰图")
print("[PY-SPY] 可以用浏览器直接打开 SVG 文件查看")


# ===== 主入口 =====

if __name__ == '__main__':
import sys

print("=" * 50)
print("py-spy 火焰图分析示例")
print("使用方法: py-spy record -o flame.svg -d 20 -- python 本文件")
print("=" * 50)

save_sampling_data()

# 运行主工作负载
heavy_workload()

# 一些后台开销
background_overhead()

print("[DONE] 所有任务完成，请查看生成的火焰图 SVG 文件")