并发选择指南与 GIL

import sys
import time

# 查看 GIL 切换间隔
print(f"GIL 切换间隔: {sys.getswitchinterval()}s")  # 0.005 (5ms, Python 3.12+)

# 设置 GIL 切换间隔（慎用）
sys.setswitchinterval(0.01)  # 设置为 10ms

# C 扩展可以主动释放 GIL
# 在 C 代码中：
# Py_BEGIN_ALLOW_THREADS  → 释放 GIL
# ... 执行不涉及 Python 对象的 C 代码 ...
# Py_END_ALLOW_THREADS    → 重新获取 GIL

"""
演示：纯 Python CPU 密集型任务无法利用多核
"""
import threading
import time
import os

def cpu_bound_task(n: int) -> int:
    """CPU 密集型：纯 Python 计算"""
    total = 0
    for i in range(n):
        total += i * i
    return total

def run_single_thread():
    """单线程执行"""
    start = time.perf_counter()
    result1 = cpu_bound_task(5_000_000)
    result2 = cpu_bound_task(5_000_000)
    elapsed = time.perf_counter() - start
    print(f"单线程耗时: {elapsed:.2f}s")
    return elapsed

def run_multi_thread():
    """多线程执行"""
    results = [0, 0]

    def worker(idx, n):
        results[idx] = cpu_bound_task(n)

    start = time.perf_counter()
    t1 = threading.Thread(target=worker, args=(0, 5_000_000))
    t2 = threading.Thread(target=worker, args=(1, 5_000_000))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    elapsed = time.perf_counter() - start
    print(f"多线程耗时: {elapsed:.2f}s (由于 GIL，可能比单线程更慢)")
    return elapsed

if __name__ == "__main__":
    print(f"CPU 核心数: {os.cpu_count()}")
    t_single = run_single_thread()
    t_multi = run_multi_thread()
    print(f"多线程/单线程比率: {t_multi / t_single:.2f}x")
    # 预期结果：多线程不会更快，甚至可能更慢（GIL 切换开销）

"""
演示：IO 密集型任务，多线程显著加速
"""
import threading
import time
from urllib.request import urlopen

URLS = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/1",
]

def fetch_url(url: str) -> str:
    """IO 密集型：网络请求"""
    with urlopen(url, timeout=10) as resp:
        return resp.read()

def run_sync():
    """同步顺序执行"""
    start = time.perf_counter()
    for url in URLS:
        fetch_url(url)
    elapsed = time.perf_counter() - start
    print(f"同步执行耗时: {elapsed:.2f}s")
    return elapsed

def run_threaded():
    """多线程并发执行"""
    start = time.perf_counter()
    threads = []
    for url in URLS:
        t = threading.Thread(target=fetch_url, args=(url,))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    elapsed = time.perf_counter() - start
    print(f"多线程执行耗时: {elapsed:.2f}s")
    return elapsed

if __name__ == "__main__":
    t_sync = run_sync()
    t_thread = run_threaded()
    print(f"加速比: {t_sync / t_thread:.2f}x")
    # 预期结果：多线程接近 4x 加速（4 个 IO 任务并行）

"""
演示：CPU 密集型任务，多进程显著加速，多线程无效
"""
import threading
import multiprocessing
import time

def cpu_heavy(n: int) -> int:
    """CPU 密集型任务"""
    total = 0
    for i in range(n):
        total += i * i
    return total

N = 5_000_000
WORKERS = 4

def benchmark_sync():
    start = time.perf_counter()
    for _ in range(WORKERS):
        cpu_heavy(N)
    return time.perf_counter() - start

def benchmark_threading():
    results = [0] * WORKERS

    def worker(idx):
        results[idx] = cpu_heavy(N)

    start = time.perf_counter()
    threads = [threading.Thread(target=worker, args=(i,)) for i in range(WORKERS)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return time.perf_counter() - start

def benchmark_multiprocessing():
    start = time.perf_counter()
    with multiprocessing.Pool(WORKERS) as pool:
        pool.map(cpu_heavy, [N] * WORKERS)
    return time.perf_counter() - start

if __name__ == "__main__":
    t_sync = benchmark_sync()
    t_thread = benchmark_threading()
    t_proc = benchmark_multiprocessing()

    print(f"同步:     {t_sync:.2f}s (基准)")
    print(f"多线程:   {t_thread:.2f}s ({t_thread / t_sync:.2f}x)")
    print(f"多进程:   {t_proc:.2f}s ({t_proc / t_sync:.2f}x)")
    # 预期结果：
    # 多线程 ≈ 1.0x~1.2x（GIL 限制，甚至更慢）
    # 多进程 ≈ 0.3x~0.5x（真正并行，接近核心数倍加速）

"""
演示：NumPy 等 C 扩展在执行计算时释放 GIL
"""
import numpy as np
import threading
import time

def numpy_heavy():
    """NumPy 矩阵运算，内部释放 GIL"""
    a = np.random.rand(2000, 2000)
    b = np.random.rand(2000, 2000)
    for _ in range(10):
        _ = a @ b  # 矩阵乘法，释放 GIL

def benchmark_numpy_threading():
    start = time.perf_counter()
    threads = [threading.Thread(target=numpy_heavy) for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return time.perf_counter() - start

def benchmark_numpy_sync():
    start = time.perf_counter()
    for _ in range(4):
        numpy_heavy()
    return time.perf_counter() - start

if __name__ == "__main__":
    t_sync = benchmark_numpy_sync()
    t_thread = benchmark_numpy_threading()
    print(f"NumPy 同步:   {t_sync:.2f}s")
    print(f"NumPy 多线程: {t_thread:.2f}s ({t_thread / t_sync:.2f}x)")
    # 预期结果：多线程有显著加速（NumPy 释放了 GIL）

"""
Python 3.13+ free-threaded 模式演示
构建方式: ./configure --disable-gil && make
运行: python3.13t script.py
"""
import sys
import threading
import time

# 检查是否运行在 free-threaded 模式
if sys.version_info >= (3, 13):
    # Python 3.13+ 提供 GIL 状态查询
    if hasattr(sys, '_is_gil_enabled'):
        print(f"GIL 启用: {sys._is_gil_enabled()}")
    else:
        print("当前 Python 不支持 GIL 状态查询")

def cpu_heavy(n: int) -> int:
    total = 0
    for i in range(n):
        total += i * i
    return total

def benchmark_free_threaded():
    """在 free-threaded 模式下，多线程可真正并行"""
    N = 5_000_000
    WORKERS = 4
    results = [0] * WORKERS

    def worker(idx):
        results[idx] = cpu_heavy(N)

    start = time.perf_counter()
    threads = [threading.Thread(target=worker, args=(i,)) for i in range(WORKERS)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    elapsed = time.perf_counter() - start
    print(f"Free-threaded 多线程耗时: {elapsed:.2f}s")
    return elapsed

if __name__ == "__main__":
    benchmark_free_threaded()

"""
并发模型性能对比基准测试
测试场景：IO 密集型、CPU 密集型、混合型
"""
import asyncio
import time
import threading
import multiprocessing
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from functools import partial

# ============================================================
# 任务定义
# ============================================================

def io_task(duration: float = 0.1) -> str:
    """模拟 IO 任务（time.sleep 模拟阻塞）"""
    time.sleep(duration)
    return f"io_done_{duration}"

async def async_io_task(duration: float = 0.1) -> str:
    """模拟异步 IO 任务"""
    await asyncio.sleep(duration)
    return f"async_io_done_{duration}"

def cpu_task(n: int = 500_000) -> int:
    """CPU 密集型任务"""
    total = 0
    for i in range(n):
        total += i * i
    return total

def mixed_task(n: int = 100_000, duration: float = 0.05) -> int:
    """混合型任务：先 IO 再 CPU"""
    time.sleep(duration)  # IO 部分
    total = 0
    for i in range(n):    # CPU 部分
        total += i * i
    return total

# ============================================================
# 同步基准
# ============================================================

def benchmark_sync_io(task_count: int = 8, duration: float = 0.1) -> float:
    start = time.perf_counter()
    for _ in range(task_count):
        io_task(duration)
    return time.perf_counter() - start

def benchmark_sync_cpu(task_count: int = 4, n: int = 500_000) -> float:
    start = time.perf_counter()
    for _ in range(task_count):
        cpu_task(n)
    return time.perf_counter() - start

# ============================================================
# 多线程
# ============================================================

def benchmark_threading_io(task_count: int = 8, duration: float = 0.1) -> float:
    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=task_count) as executor:
        list(executor.map(partial(io_task, duration), [duration] * task_count))
    return time.perf_counter() - start

def benchmark_threading_cpu(task_count: int = 4, n: int = 500_000) -> float:
    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=task_count) as executor:
        list(executor.map(cpu_task, [n] * task_count))
    return time.perf_counter() - start

# ============================================================
# 多进程
# ============================================================

def benchmark_multiprocessing_cpu(task_count: int = 4, n: int = 500_000) -> float:
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=task_count) as executor:
        list(executor.map(cpu_task, [n] * task_count))
    return time.perf_counter() - start

def benchmark_multiprocessing_io(task_count: int = 8, duration: float = 0.1) -> float:
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=task_count) as executor:
        list(executor.map(partial(io_task, duration), [duration] * task_count))
    return time.perf_counter() - start

# ============================================================
# asyncio
# ============================================================

async def _async_io_benchmark(task_count: int, duration: float) -> float:
    start = time.perf_counter()
    tasks = [async_io_task(duration) for _ in range(task_count)]
    await asyncio.gather(*tasks)
    return time.perf_counter() - start

def benchmark_asyncio_io(task_count: int = 8, duration: float = 0.1) -> float:
    return asyncio.run(_async_io_benchmark(task_count, duration))

# ============================================================
# 运行全部基准
# ============================================================

def run_all_benchmarks():
    print("=" * 60)
    print("并发模型性能对比基准测试")
    print("=" * 60)

    # IO 密集型
    print("\n--- IO 密集型 (8 个任务, 每个 0.1s) ---")
    t_sync_io = benchmark_sync_io()
    t_thread_io = benchmark_threading_io()
    t_process_io = benchmark_multiprocessing_io()
    t_async_io = benchmark_asyncio_io()

    print(f"同步:       {t_sync_io:.3f}s (基准)")
    print(f"多线程:     {t_thread_io:.3f}s ({t_sync_io / t_thread_io:.1f}x 加速)")
    print(f"多进程:     {t_process_io:.3f}s ({t_sync_io / t_process_io:.1f}x 加速)")
    print(f"asyncio:    {t_async_io:.3f}s ({t_sync_io / t_async_io:.1f}x 加速)")

    # CPU 密集型
    print("\n--- CPU 密集型 (4 个任务, 各 500K 迭代) ---")
    t_sync_cpu = benchmark_sync_cpu()
    t_thread_cpu = benchmark_threading_cpu()
    t_process_cpu = benchmark_multiprocessing_cpu()

    print(f"同步:       {t_sync_cpu:.3f}s (基准)")
    print(f"多线程:     {t_thread_cpu:.3f}s ({t_sync_cpu / t_thread_cpu:.2f}x)")
    print(f"多进程:     {t_process_cpu:.3f}s ({t_sync_cpu / t_process_cpu:.1f}x 加速)")

if __name__ == "__main__":
    run_all_benchmarks()

"""
混合策略：进程池处理 CPU 密集计算，线程池处理 IO 操作
场景：批量下载图片 + 图片处理
"""
import time
import os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

def download_image(url: str) -> bytes:
    """IO 密集型：下载图片（模拟）"""
    time.sleep(0.1)  # 模拟网络延迟
    return f"data_from_{url}".encode()

def process_image(data: bytes) -> dict:
    """CPU 密集型：图片处理（模拟）"""
    total = 0
    for i in range(200_000):
        total += i * i
    return {"size": len(data), "hash": hash(data), "computed": total}

def pipeline_thread_then_process(urls: list[str]) -> list[dict]:
    """
    混合策略一：先线程池下载，再进程池处理
    适合 IO 和 CPU 阶段明显分离的任务
    """
    # 阶段1：线程池并发下载
    downloaded = []
    with ThreadPoolExecutor(max_workers=8) as thread_pool:
        future_to_url = {
            thread_pool.submit(download_image, url): url
            for url in urls
        }
        for future in as_completed(future_to_url):
            downloaded.append(future.result())

    # 阶段2：进程池并发处理
    results = []
    cpu_workers = min(os.cpu_count() or 4, len(downloaded))
    with ProcessPoolExecutor(max_workers=cpu_workers) as process_pool:
        for result in process_pool.map(process_image, downloaded):
            results.append(result)

    return results

if __name__ == "__main__":
    urls = [f"https://example.com/img{i}.jpg" for i in range(16)]
    start = time.perf_counter()
    results = pipeline_thread_then_process(urls)
    elapsed = time.perf_counter() - start
    print(f"混合策略耗时: {elapsed:.2f}s, 处理 {len(results)} 张图片")

"""
混合策略二：asyncio 主循环 + ProcessPoolExecutor 处理 CPU 任务
场景：异步 Web 服务中需要执行 CPU 密集计算
"""
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor

def cpu_intensive_analysis(data: dict) -> dict:
    """CPU 密集型：数据分析"""
    total = 0
    for i in range(1_000_000):
        total += i * i
    return {"input_id": data.get("id"), "result": total}

async def fetch_data(session_id: int) -> dict:
    """IO 密集型：异步获取数据"""
    await asyncio.sleep(0.05)  # 模拟异步数据库查询
    return {"id": session_id, "raw": f"session_{session_id}"}

async def process_pipeline():
    """
    混合策略二：asyncio 主循环 + 进程池
    异步获取数据，CPU 计算委托给进程池
    """
    loop = asyncio.get_running_loop()
    cpu_workers = min(4, os.cpu_count() or 4)
    process_pool = ProcessPoolExecutor(max_workers=cpu_workers)

    try:
        # 异步并发获取数据
        fetch_tasks = [fetch_data(i) for i in range(8)]
        data_list = await asyncio.gather(*fetch_tasks)

        # CPU 密集计算委托给进程池（不阻塞事件循环）
        analysis_tasks = [
            loop.run_in_executor(process_pool, cpu_intensive_analysis, data)
            for data in data_list
        ]
        results = await asyncio.gather(*analysis_tasks)

        return results
    finally:
        process_pool.shutdown(wait=False)

import os

if __name__ == "__main__":
    start = time.perf_counter()
    results = asyncio.run(process_pipeline())
    elapsed = time.perf_counter() - start
    print(f"asyncio + 进程池耗时: {elapsed:.2f}s, 结果数: {len(results)}")

"""
智能并发选择器：根据任务类型自动选择最优并发模型
"""
import asyncio
import time
import os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Any

class TaskType(Enum):
    IO_BOUND = "io_bound"
    CPU_BOUND = "cpu_bound"
    MIXED = "mixed"

@dataclass
class BenchmarkResult:
    task_type: TaskType
    task_count: int
    sync_time: float
    concurrent_time: float
    speedup: float
    model_used: str

class SmartConcurrencySelector:
    """智能并发选择器"""

    def __init__(self, cpu_workers: int | None = None):
        self.cpu_workers = cpu_workers or min(os.cpu_count() or 4, 8)
        self.io_workers = min(32, (os.cpu_count() or 4) * 4)

    def detect_task_type(
        self,
        func: Callable,
        sample_args: tuple = (),
        io_threshold: float = 0.5,
    ) -> TaskType:
        """
        自动检测任务类型：
        - 运行一次函数，测量 CPU 时间和墙钟时间
        - CPU时间 / 墙钟时间 < io_threshold → IO密集型
        - CPU时间 / 墙钟时间 > (1 - io_threshold) → CPU密集型
        - 否则 → 混合型
        """
        import resource

        start_cpu = time.process_time()
        start_wall = time.perf_counter()
        func(*sample_args)
        cpu_time = time.process_time() - start_cpu
        wall_time = time.perf_counter() - start_wall

        if wall_time == 0:
            return TaskType.CPU_BOUND

        cpu_ratio = cpu_time / wall_time

        if cpu_ratio < io_threshold:
            return TaskType.IO_BOUND
        elif cpu_ratio > (1 - io_threshold):
            return TaskType.CPU_BOUND
        else:
            return TaskType.MIXED

    def select_model(self, task_type: TaskType, task_count: int) -> str:
        """根据任务类型和数量选择并发模型"""
        if task_type == TaskType.IO_BOUND:
            if task_count <= 50:
                return "threading"
            else:
                return "asyncio"
        elif task_type == TaskType.CPU_BOUND:
            return "multiprocessing"
        else:  # MIXED
            if task_count <= 20:
                return "multiprocessing"
            else:
                return "asyncio+process_pool"

    def run(
        self,
        func: Callable,
        args_list: list[tuple],
        model: str | None = None,
    ) -> list[Any]:
        """
        执行任务，自动选择并发模型或使用指定模型
        """
        task_count = len(args_list)

        # 自动检测任务类型
        if args_list:
            task_type = self.detect_task_type(func, args_list[0])
        else:
            task_type = TaskType.IO_BOUND

        # 选择模型
        if model is None:
            model = self.select_model(task_type, task_count)

        print(f"任务类型: {task_type.value}, 任务数: {task_count}, 选择模型: {model}")

        # 执行
        if model == "threading":
            return self._run_threading(func, args_list)
        elif model == "multiprocessing":
            return self._run_multiprocessing(func, args_list)
        elif model == "asyncio":
            return asyncio.run(self._run_asyncio(func, args_list))
        elif model == "asyncio+process_pool":
            return asyncio.run(self._run_asyncio_with_process_pool(func, args_list))
        else:
            raise ValueError(f"未知模型: {model}")

    def _run_threading(self, func, args_list):
        with ThreadPoolExecutor(max_workers=self.io_workers) as pool:
            futures = [pool.submit(func, *args) for args in args_list]
            return [f.result() for f in futures]

    def _run_multiprocessing(self, func, args_list):
        with ProcessPoolExecutor(max_workers=self.cpu_workers) as pool:
            futures = [pool.submit(func, *args) for args in args_list]
            return [f.result() for f in futures]

    async def _run_asyncio(self, func, args_list):
        """对同步函数使用线程池包装为异步"""
        loop = asyncio.get_running_loop()
        with ThreadPoolExecutor(max_workers=self.io_workers) as pool:
            tasks = [
                loop.run_in_executor(pool, partial(func, *args))
                for args in args_list
            ]
            return await asyncio.gather(*tasks)

    async def _run_asyncio_with_process_pool(self, func, args_list):
        """asyncio + 进程池混合"""
        loop = asyncio.get_running_loop()
        with ProcessPoolExecutor(max_workers=self.cpu_workers) as pool:
            tasks = [
                loop.run_in_executor(pool, partial(func, *args))
                for args in args_list
            ]
            return await asyncio.gather(*tasks)

from functools import partial

# 使用示例
if __name__ == "__main__":
    selector = SmartConcurrencySelector()

    # IO 密集型任务
    def io_task(url: str):
        time.sleep(0.1)
        return f"response_from_{url}"

    urls = [f"https://api.example.com/data/{i}" for i in range(16)]
    results = selector.run(io_task, [(url,) for url in urls])

    # CPU 密集型任务
    def cpu_task(n: int):
        total = 0
        for i in range(n):
            total += i * i
        return total

    nums = [500_000] * 8
    results = selector.run(cpu_task, [(n,) for n in nums])

"""
CPU+IO 混合型任务的最佳策略
场景：数据分析平台 — 从数据库读取数据 + 执行统计计算 + 写回结果
"""
import asyncio
import time
import os
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass

@dataclass
class DataChunk:
    chunk_id: int
    size: int

@dataclass
class AnalysisResult:
    chunk_id: int
    mean: float
    variance: float
    compute_time: float

# ============================================================
# IO 操作（模拟异步数据库）
# ============================================================

async def read_chunk_from_db(chunk_id: int) -> DataChunk:
    """异步读取数据块"""
    await asyncio.sleep(0.05)  # 模拟数据库查询
    return DataChunk(chunk_id=chunk_id, size=100_000)

async def write_result_to_db(result: AnalysisResult) -> None:
    """异步写回结果"""
    await asyncio.sleep(0.02)  # 模拟数据库写入
    # 实际中: await db.execute("INSERT INTO results ...", result)

# ============================================================
# CPU 操作（在进程池中执行）
# ============================================================

def analyze_chunk(chunk: DataChunk) -> AnalysisResult:
    """CPU 密集型：统计分析"""
    start = time.perf_counter()
    total = 0
    total_sq = 0
    for i in range(chunk.size):
        val = i * 0.001
        total += val
        total_sq += val * val
    n = chunk.size
    mean = total / n
    variance = total_sq / n - mean * mean
    compute_time = time.perf_counter() - start
    return AnalysisResult(
        chunk_id=chunk.chunk_id,
        mean=mean,
        variance=variance,
        compute_time=compute_time,
    )

# ============================================================
# 混合策略实现
# ============================================================

async def process_one_chunk(
    chunk_id: int,
    process_pool: ProcessPoolExecutor,
) -> AnalysisResult:
    """
    单个数据块的处理流水线：
    异步读取 → 进程池计算 → 异步写回
    """
    # 1. 异步读取
    chunk = await read_chunk_from_db(chunk_id)

    # 2. CPU 计算委托给进程池（不阻塞事件循环）
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(process_pool, analyze_chunk, chunk)

    # 3. 异步写回
    await write_result_to_db(result)

    return result

async def run_mixed_pipeline(chunk_count: int = 12) -> list[AnalysisResult]:
    """
    混合策略主入口：asyncio + ProcessPoolExecutor
    - asyncio 处理所有 IO 操作（读/写数据库）
    - ProcessPoolExecutor 处理 CPU 计算
    - 流水线并行：多个 chunk 可同时处于不同阶段
    """
    cpu_workers = min(os.cpu_count() or 4, chunk_count)
    process_pool = ProcessPoolExecutor(max_workers=cpu_workers)

    try:
        # 信号量控制并发数，避免数据库过载
        db_semaphore = asyncio.Semaphore(10)

        async def limited_process(chunk_id: int) -> AnalysisResult:
            async with db_semaphore:
                return await process_one_chunk(chunk_id, process_pool)

        # 并发处理所有数据块
        tasks = [limited_process(i) for i in range(chunk_count)]
        results = await asyncio.gather(*tasks)
        return results
    finally:
        process_pool.shutdown(wait=False)

# ============================================================
# 对比：纯同步方案
# ============================================================

def run_sync_pipeline(chunk_count: int = 12) -> list[AnalysisResult]:
    """纯同步方案作为对比"""
    results = []
    for i in range(chunk_count):
        # 同步 IO 读取
        time.sleep(0.05)
        chunk = DataChunk(chunk_id=i, size=100_000)
        # CPU 计算
        result = analyze_chunk(chunk)
        # 同步 IO 写回
        time.sleep(0.02)
        results.append(result)
    return results

if __name__ == "__main__":
    CHUNK_COUNT = 12

    # 同步方案
    start = time.perf_counter()
    sync_results = run_sync_pipeline(CHUNK_COUNT)
    sync_time = time.perf_counter() - start
    print(f"同步方案:   {sync_time:.2f}s, 处理 {len(sync_results)} 块")

    # 混合策略
    start = time.perf_counter()
    mixed_results = asyncio.run(run_mixed_pipeline(CHUNK_COUNT))
    mixed_time = time.perf_counter() - start
    print(f"混合策略:   {mixed_time:.2f}s, 处理 {len(mixed_results)} 块")

    print(f"加速比:     {sync_time / mixed_time:.1f}x")
    # 预期：混合策略 2-4x 加速

"""
常见陷阱：在 async 函数中使用同步阻塞调用
"""
import asyncio
import time

# ❌ 错误：阻塞事件循环
async def bad_fetch():
    """这会阻塞整个事件循环！"""
    time.sleep(2)  # 同步阻塞，所有协程卡住
    return "result"

# ✅ 正确：使用异步等价物
async def good_fetch():
    """正确使用异步等待"""
    await asyncio.sleep(2)  # 异步等待，其他协程可运行
    return "result"

# ✅ 正确：用 run_in_executor 包装不可替代的同步调用
async def good_sync_wrapper():
    """将同步阻塞调用委托给线程池"""
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, time.sleep, 2)
    return result

# 演示对比
async def demo():
    # 错误方式：所有请求串行执行
    start = time.perf_counter()
    await asyncio.gather(bad_fetch(), bad_fetch(), bad_fetch())
    print(f"阻塞方式: {time.perf_counter() - start:.2f}s")  # ~6s

    # 正确方式：所有请求并发执行
    start = time.perf_counter()
    await asyncio.gather(good_fetch(), good_fetch(), good_fetch())
    print(f"异步方式: {time.perf_counter() - start:.2f}s")  # ~2s

"""
常见陷阱：multiprocessing fork 后的死锁
"""
import multiprocessing as mp
import threading
import time

# ❌ 危险：fork 后线程状态不一致
lock = threading.Lock()

def worker():
    """子进程中 lock 的状态可能不一致"""
    with lock:  # 可能永远阻塞（父进程中的锁状态被 fork 复制）
        return "done"

# ✅ 安全：使用 spawn 启动方式
def safe_worker():
    """spawn 方式启动全新进程，避免继承状态"""
    return "done"

if __name__ == "__main__":
    # 安全的启动方式
    mp.set_start_method("spawn", force=True)

    with mp.Pool(4) as pool:
        results = pool.map(safe_worker, range(4))
        print(results)

"""
并发编程最佳实践清单
"""

# 1. 先测量，再优化
import time

def measure(func, *args):
    start = time.perf_counter()
    result = func(*args)
    elapsed = time.perf_counter() - start
    print(f"{func.__name__}: {elapsed:.3f}s")
    return result

# 2. 用池化替代手动创建
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# ❌ 避免
# threads = [threading.Thread(target=work) for _ in range(100)]
# ✅ 推荐
with ThreadPoolExecutor(max_workers=8) as pool:
    pool.map(work, items)

# 3. 优先使用高级接口
from concurrent.futures import as_completed

# ✅ as_completed 按完成顺序获取结果
with ThreadPoolExecutor() as pool:
    futures = {pool.submit(work, item): item for item in items}
    for future in as_completed(futures):
        result = future.result()

# 4. 注意异常处理
def safe_work(item):
    try:
        return work(item)
    except Exception as e:
        print(f"处理 {item} 失败: {e}")
        return None

# 5. 合理使用上下文管理器
# ✅ 确保资源释放
with ProcessPoolExecutor() as pool:
    results = list(pool.map(cpu_work, data))

# 6. 避免在子进程中使用全局状态
# 使用初始化函数
def init_worker():
    global local_cache
    local_cache = {}  # 每个进程独立初始化

with ProcessPoolExecutor(initializer=init_worker) as pool:
    pool.map(work_with_cache, items)