Python 性能优化：使用 ZeroMQ 突破 GIL 限制

#!/usr/bin/env python
# 基本的 ZeroMQ REQ-REP 模式示例

import zmq
import time
import sys
import threading

def server():
    # 创建上下文和 socket
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.bind("tcp://*:5555")
    
    print("服务器已启动，等待请求...")
    
    # 服务循环
    while True:
        # 等待客户端请求
        message = socket.recv()
        print(f"收到请求: {message.decode()}")
        
        # 模拟工作
        time.sleep(1)
        
        # 发送回复
        socket.send(b"Request processed")

def client():
    # 创建上下文和 socket
    context = zmq.Context()
    socket = context.socket(zmq.REQ)
    socket.connect("tcp://localhost:5555")
    
    # 发送请求
    for i in range(5):
        print(f"发送请求 {i}...")
        socket.send(f"请求 #{i}".encode())
        
        # 获取回复
        message = socket.recv()
        print(f"收到回复: {message.decode()}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python zmq_basic.py [server|client]")
        sys.exit(1)
    
    if sys.argv[1] == "server":
        server()
    elif sys.argv[1] == "client":
        client()
    elif sys.argv[1] == "both":
        # 在单独的线程中启动服务器
        server_thread = threading.Thread(target=server)
        server_thread.daemon = True
        server_thread.start()
        
        # 给服务器一点时间启动
        time.sleep(1)
        
        # 运行客户端
        client()
    else:
        print("用法: python zmq_basic.py [server|client|both]")

#!/usr/bin/env python
# filename: cpu_bound_demo.py
# CPU密集型任务示例 - 受GIL限制的性能问题

import numpy as np
import time
import matplotlib.pyplot as plt
from matplotlib.figure import Figure
from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas

def generate_data(size=1000):
    """生成随机数据矩阵"""
    return np.random.random((size, size))

def process_data(data, kernel_size=5):
    """对数据应用简单的模糊滤镜 (CPU密集型操作)"""
    result = np.zeros_like(data)
    rows, cols = data.shape
    
    # 简单的滑动窗口平均 (模拟模糊操作)
    for i in range(kernel_size//2, rows - kernel_size//2):
        for j in range(kernel_size//2, cols - kernel_size//2):
            # 提取窗口
            window = data[i-kernel_size//2:i+kernel_size//2+1, 
                          j-kernel_size//2:j+kernel_size//2+1]
            # 计算平均值
            result[i, j] = np.mean(window)
    
    return result

def visualize_results(original, processed, execution_time, title):
    """可视化原始数据和处理后的数据"""
    fig = Figure(figsize=(10, 5))
    canvas = FigureCanvas(fig)
    
    ax1 = fig.add_subplot(121)
    ax1.imshow(original, cmap='viridis')
    ax1.set_title('Original Data')
    ax1.axis('off')
    
    ax2 = fig.add_subplot(122)
    ax2.imshow(processed, cmap='viridis')
    ax2.set_title(f'Processed Data\nExecution Time: {execution_time:.2f} seconds')
    ax2.axis('off')
    
    fig.suptitle(title)
    fig.tight_layout()
    
    # 保存图像
    fig.savefig(f"{title.replace(' ', '_').lower()}.png")
    print(f"结果已保存为 {title.replace(' ', '_').lower()}.png")

def run_single_process(data_size=500, kernel_size=5):
    """在单个进程中运行数据处理"""
    print(f"生成 {data_size}x{data_size} 的数据矩阵...")
    data = generate_data(data_size)
    
    print("开始处理数据...")
    start_time = time.time()
    result = process_data(data, kernel_size)
    end_time = time.time()
    
    execution_time = end_time - start_time
    print(f"处理完成，耗时: {execution_time:.2f} 秒")
    
    # 可视化结果
    visualize_results(data, result, execution_time, "Single Process")
    
    return execution_time

if __name__ == "__main__":
    # 运行单进程版本
    execution_time = run_single_process(data_size=500, kernel_size=5)
    print(f"单进程执行时间: {execution_time:.2f} 秒")

#!/usr/bin/env python
# filename: zmq_distributed_demo.py
# 使用不同并行处理方法对比CPU密集型任务的性能

import numpy as np
import time
import zmq
import pickle
import multiprocessing
from multiprocessing import Pool, Process, Queue, cpu_count
import os
import matplotlib.pyplot as plt
from matplotlib.figure import Figure
from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas
import hashlib

# 导入单进程版本中的函数
from cpu_bound_demo import generate_data, process_data, visualize_results

def worker(worker_id):
    """工作进程 - 接收数据块并处理"""
    context = zmq.Context()
    
    # 设置接收任务的PULL socket
    receiver = context.socket(zmq.PULL)
    receiver.connect("tcp://localhost:5557")
    
    # 设置发送结果的PUSH socket
    sender = context.socket(zmq.PUSH)
    sender.connect("tcp://localhost:5558")
    
    print(f"工作进程 {worker_id} 已启动")
    
    # 处理循环
    while True:
        try:
            # 接收任务
            task = receiver.recv_pyobj()
            
            # 检查是否是终止信号
            if task == "DONE":
                print(f"工作进程 {worker_id} 收到终止信号")
                break
            
            # 解包任务数据
            chunk_id, data_chunk, kernel_size = task
            
            # 处理数据
            result_chunk = process_data(data_chunk, kernel_size)
            
            # 发送结果
            sender.send_pyobj((chunk_id, result_chunk))
            
        except Exception as e:
            print(f"工作进程 {worker_id} 错误: {e}")
    
    # 关闭连接
    receiver.close()
    sender.close()
    context.term()
    print(f"工作进程 {worker_id} 已终止")

def split_data(data, num_chunks, kernel_size=5, overlap=True):
    """将数据分割成多个块，可选添加重叠区域以解决边界问题"""
    rows, cols = data.shape
    chunk_size = rows // num_chunks
    chunks = []
    chunk_bounds = []  # 记录每个块的有效边界（不包括重叠区域）
    
    # 计算重叠大小，至少需要kernel_size//2的重叠
    overlap_size = kernel_size // 2 if overlap else 0
    
    for i in range(num_chunks):
        # 计算块的起始和结束行，包括重叠区域
        start_row = max(0, i * chunk_size - overlap_size)
        end_row = min(rows, (i + 1) * chunk_size + overlap_size if i < num_chunks - 1 else rows)
        
        # 计算有效边界（不包括重叠区域）
        valid_start = i * chunk_size
        valid_end = min(rows, (i + 1) * chunk_size)
        
        # 存储块及其有效边界
        chunks.append(data[start_row:end_row, :].copy())  # 使用copy()确保数据独立
        chunk_bounds.append((valid_start - start_row, valid_end - start_row))
    
    return chunks, chunk_bounds

def run_distributed_process(data_size=500, kernel_size=5, num_workers=None, input_data=None):
    """使用ZeroMQ分布式运行数据处理"""
    # 如果没有指定工作进程数，使用CPU核心数
    if num_workers is None:
        num_workers = multiprocessing.cpu_count()
    
    if input_data is None:
        print(f"生成 {data_size}x{data_size} 的数据矩阵...")
        data = generate_data(data_size)
    else:
        print(f"使用提供的输入数据...")
        data = input_data
    
    # 分割数据，并添加重叠区域
    print(f"将数据分割成 {num_workers} 块，并添加重叠区域...")
    data_chunks, chunk_bounds = split_data(data, num_workers, kernel_size)
    
    # 创建ZeroMQ上下文
    context = zmq.Context()
    
    # 设置任务分发的PUSH socket
    task_sender = context.socket(zmq.PUSH)
    task_sender.bind("tcp://*:5557")
    
    # 设置结果收集的PULL socket
    result_receiver = context.socket(zmq.PULL)
    result_receiver.bind("tcp://*:5558")
    
    # 启动工作进程
    processes = []
    for i in range(num_workers):
        p = multiprocessing.Process(target=worker, args=(i,))
        p.daemon = True
        p.start()
        processes.append(p)
    
    # 给工作进程一点时间启动
    time.sleep(0.5)
    
    # 开始计时
    print("开始分布式处理数据...")
    start_time = time.time()
    
    # 发送任务到工作进程
    for i, chunk in enumerate(data_chunks):
        task_sender.send_pyobj((i, chunk, kernel_size))
    
    # 收集结果
    results = [None] * len(data_chunks)
    for _ in range(len(data_chunks)):
        chunk_id, result_chunk = result_receiver.recv_pyobj()
        results[chunk_id] = result_chunk
    
    # 合并结果，只保留每个块的有效区域
    final_results = []
    for i, result_chunk in enumerate(results):
        valid_start, valid_end = chunk_bounds[i]
        final_results.append(result_chunk[valid_start:valid_end, :])
    
    # 合并有效区域
    result = np.vstack(final_results)
    
    # 结束计时
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"处理完成，耗时: {execution_time:.2f} 秒")
    
    # 发送终止信号给工作进程
    for _ in range(num_workers):
        task_sender.send_pyobj("DONE")
    
    # 等待工作进程终止
    for p in processes:
        p.join(timeout=1)
    
    # 关闭ZeroMQ连接
    task_sender.close()
    result_receiver.close()
    context.term()
    
    # 可视化结果
    visualize_results(data, result, execution_time, "ZeroMQ Distributed")
    
    return execution_time, result

# 新增的直接使用多进程的实现
def mp_worker(data_chunk, kernel_size, result_queue, chunk_id):
    """多进程工作函数 - 处理数据块并将结果放入队列"""
    try:
        # 处理数据
        result_chunk = process_data(data_chunk, kernel_size)
        # 将结果放入队列
        result_queue.put((chunk_id, result_chunk))
    except Exception as e:
        print(f"多进程工作函数错误: {e}")

def run_multiprocessing(data_size=500, kernel_size=5, num_workers=None, input_data=None):
    """使用Python原生多进程运行数据处理"""
    # 如果没有指定工作进程数，使用CPU核心数
    if num_workers is None:
        num_workers = cpu_count()
    
    if input_data is None:
        print(f"生成 {data_size}x{data_size} 的数据矩阵...")
        data = generate_data(data_size)
    else:
        print(f"使用提供的输入数据...")
        data = input_data
    
    # 分割数据，并添加重叠区域
    print(f"将数据分割成 {num_workers} 块，并添加重叠区域...")
    data_chunks, chunk_bounds = split_data(data, num_workers, kernel_size)
    
    # 创建结果队列
    result_queue = Queue()
    
    # 创建进程
    processes = []
    
    # 开始计时
    print("开始多进程处理数据...")
    start_time = time.time()
    
    # 启动工作进程
    for i, chunk in enumerate(data_chunks):
        p = Process(target=mp_worker, args=(chunk, kernel_size, result_queue, i))
        processes.append(p)
        p.start()
    
    # 收集结果
    results = [None] * len(data_chunks)
    for _ in range(len(data_chunks)):
        chunk_id, result_chunk = result_queue.get()
        results[chunk_id] = result_chunk
    
    # 等待所有进程完成
    for p in processes:
        p.join()
    
    # 合并结果，只保留每个块的有效区域
    final_results = []
    for i, result_chunk in enumerate(results):
        valid_start, valid_end = chunk_bounds[i]
        final_results.append(result_chunk[valid_start:valid_end, :])
    
    # 合并有效区域
    result = np.vstack(final_results)
    
    # 结束计时
    end_time = time.time()
    execution_time = end_time - start_time
    print(f"处理完成，耗时: {execution_time:.2f} 秒")
    
    # 可视化结果
    visualize_results(data, result, execution_time, "Multiprocessing")
    
    return execution_time, result

# 用于验证结果一致性的函数
def calculate_result_hash(result):
    """计算结果数组的哈希值以验证一致性"""
    # 将numpy数组转换为字节序列
    # 先四舍五入到固定小数位数，避免浮点数误差引起的不一致
    rounded_result = np.round(result, 6)
    result_bytes = rounded_result.tobytes()
    # 计算SHA-256哈希
    return hashlib.sha256(result_bytes).hexdigest()

def compare_performance():
    """比较三种实现的性能并验证结果一致性"""
    print("\n===== 性能对比 =====")
    data_size = 2000
    kernel_size = 5
    
    # 设置随机种子以确保可重现性
    np.random.seed(42)
    
    # 运行单进程版本
    print("\n运行单进程版本...")
    single_data = generate_data(data_size)  # 保证所有实现使用相同的输入数据
    
    # 为了确保结果一致性，我们将使用与分布式实现相同的处理方式
    # 将数据分割，处理，然后再合并
    num_workers = cpu_count()
    data_chunks, chunk_bounds = split_data(single_data, num_workers, kernel_size)
    
    single_start = time.time()
    
    # 处理每个块
    result_chunks = []
    for chunk in data_chunks:
        result_chunk = process_data(chunk, kernel_size)
        result_chunks.append(result_chunk)
    
    # 合并结果，只保留每个块的有效区域
    final_results = []
    for i, result_chunk in enumerate(result_chunks):
        valid_start, valid_end = chunk_bounds[i]
        final_results.append(result_chunk[valid_start:valid_end, :])
    
    # 合并有效区域
    single_result = np.vstack(final_results)
    
    single_time = time.time() - single_start
    print(f"处理完成，耗时: {single_time:.2f} 秒")
    visualize_results(single_data, single_result, single_time, "Single Process")
    
    # 运行分布式版本
    print("\n运行ZeroMQ分布式版本...")
    zmq_time, zmq_result = run_distributed_process(data_size=data_size, kernel_size=kernel_size, input_data=single_data)
    
    # 运行原生多进程版本
    print("\n运行原生多进程版本...")
    mp_time, mp_result = run_multiprocessing(data_size=data_size, kernel_size=kernel_size, input_data=single_data)
    
    # 验证结果一致性
    single_hash = calculate_result_hash(single_result)
    zmq_hash = calculate_result_hash(zmq_result)
    mp_hash = calculate_result_hash(mp_result)
    
    print(f"\n哈希值检查:")
    print(f"  单进程结果哈希: {single_hash[:10]}...")
    print(f"  ZeroMQ结果哈希: {zmq_hash[:10]}...")
    print(f"  多进程结果哈希: {mp_hash[:10]}...")
    
    # 检查结果是否相同
    if single_hash == zmq_hash and single_hash == mp_hash:
        print("  结果一致性检查: 通过 \u2705")
    else:
        print("  结果一致性检查: 失败 \u274c")
        if single_hash != zmq_hash:
            print("    - 单进程与ZeroMQ结果不一致")
        if single_hash != mp_hash:
            print("    - 单进程与多进程结果不一致")
        if zmq_hash != mp_hash:
            print("    - ZeroMQ与多进程结果不一致")
    
    # 检查结果是否相同
    zmq_match = (single_hash == zmq_hash)
    mp_match = (single_hash == mp_hash)
    results_match = zmq_match and mp_match
    
    # 计算加速比
    zmq_speedup = single_time / zmq_time
    mp_speedup = single_time / mp_time
    
    print("\n===== 结果对比 =====")
    print(f"单进程执行时间: {single_time:.2f} 秒")
    print(f"ZeroMQ分布式执行时间: {zmq_time:.2f} 秒 (加速比: {zmq_speedup:.2f}x)")
    print(f"原生多进程执行时间: {mp_time:.2f} 秒 (加速比: {mp_speedup:.2f}x)")
    print(f"结果一致性检查: {'通过' if results_match else '失败'}")
    
    # 绘制性能对比图
    fig = Figure(figsize=(10, 6))
    canvas = FigureCanvas(fig)
    ax = fig.add_subplot(111)
    
    methods = ['Single Process', 'ZeroMQ Distributed', 'Python Multiprocessing']
    times = [single_time, zmq_time, mp_time]
    colors = ['blue', 'green', 'orange']
    
    ax.bar(methods, times, color=colors)
    ax.set_ylabel('Execution Time (seconds)')
    ax.set_title('Performance Comparison')
    
    # 添加数值标签
    for i, v in enumerate(times):
        ax.text(i, v + 0.1, f"{v:.2f}s", ha='center')
    
    # 添加加速比
    ax.text(1, times[1] * 0.5, f"Speedup: {zmq_speedup:.2f}x", 
            ha='center', fontsize=10, bbox=dict(facecolor='white', alpha=0.8))
    ax.text(2, times[2] * 0.5, f"Speedup: {mp_speedup:.2f}x", 
            ha='center', fontsize=10, bbox=dict(facecolor='white', alpha=0.8))
    
    fig.tight_layout()
    fig.savefig("performance_comparison_all.png")
    print("性能对比图已保存为 performance_comparison_all.png")

if __name__ == "__main__":
    # 比较性能
    compare_performance()

#!/usr/bin/env python
# 模拟vLLM架构的简化版本，使用ZeroMQ分离GPU和CPU工作负载

import numpy as np
import time
import zmq
import multiprocessing
import threading
import queue
import json
import argparse
from tqdm import tqdm
import matplotlib.pyplot as plt
from matplotlib.figure import Figure
from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas

# 模拟GPU计算的函数
def simulate_gpu_computation(input_data, computation_time=0.1):
    """模拟GPU上的矩阵乘法计算"""
    # 实际上是在CPU上运行，但我们用sleep来模拟GPU计算时间
    time.sleep(computation_time)
    # 模拟矩阵乘法
    result = np.dot(input_data, input_data.T)
    return result

# 模拟CPU处理的函数
def simulate_cpu_preprocessing(request_id, size=100, processing_time=0.05):
    """模拟CPU上的预处理操作"""
    # 模拟预处理耗时
    time.sleep(processing_time)
    # 生成随机输入数据
    input_data = np.random.random((size, size))
    return input_data

def simulate_cpu_postprocessing(request_id, result, processing_time=0.05):
    """模拟CPU上的后处理操作"""
    # 模拟后处理耗时
    time.sleep(processing_time)
    # 简单处理结果
    processed_result = np.mean(result)
    return processed_result

# 传统方式：单进程中顺序执行CPU和GPU操作
def traditional_approach(num_requests=10, matrix_size=100, 
                         preprocess_time=0.05, gpu_time=0.1, postprocess_time=0.05):
    """传统方式：在单一进程中顺序执行CPU和GPU操作"""
    print("\n运行传统方式（单进程顺序执行）...")
    
    start_time = time.time()
    gpu_active_time = 0
    
    for i in tqdm(range(num_requests), desc="处理请求"):
        request_id = f"req_{i}"
        
        # CPU预处理
        preprocess_start = time.time()
        input_data = simulate_cpu_preprocessing(request_id, matrix_size, preprocess_time)
        preprocess_end = time.time()
        
        # GPU计算
        gpu_start = time.time()
        result = simulate_gpu_computation(input_data, gpu_time)
        gpu_end = time.time()
        gpu_active_time += (gpu_end - gpu_start)
        
        # CPU后处理
        postprocess_start = time.time()
        final_result = simulate_cpu_postprocessing(request_id, result, postprocess_time)
        postprocess_end = time.time()
    
    end_time = time.time()
    total_time = end_time - start_time
    gpu_utilization = gpu_active_time / total_time * 100
    
    print(f"传统方式完成时间: {total_time:.2f} 秒")
    print(f"GPU活跃时间: {gpu_active_time:.2f} 秒")
    print(f"GPU利用率: {gpu_utilization:.2f}%")
    
    return total_time, gpu_utilization

# GPU进程：接收输入数据，执行GPU计算，返回结果
def gpu_worker(port_recv=5555, port_send=5556):
    """GPU工作进程，接收输入数据，执行GPU计算，发送结果"""
    context = zmq.Context()
    
    # 设置接收输入数据的PULL socket
    receiver = context.socket(zmq.PULL)
    receiver.bind(f"tcp://*:{port_recv}")
    
    # 设置发送结果的PUSH socket
    sender = context.socket(zmq.PUSH)
    sender.bind(f"tcp://*:{port_send}")
    
    print("GPU工作进程已启动")
    gpu_active_time = 0
    processed_count = 0
    start_time = time.time()
    
    # 记录每次GPU活动的开始和结束时间
    gpu_activity_periods = []
    
    try:
        while True:
            # 接收任务
            message = receiver.recv_pyobj()
            
            # 检查是否是终止信号
            if message == "DONE":
                print("GPU工作进程收到终止信号")
                # 发送GPU利用率信息
                total_time = time.time() - start_time
                gpu_utilization = gpu_active_time / total_time * 100 if total_time > 0 else 0
                sender.send_pyobj({
                    "type": "STATS",
                    "gpu_active_time": gpu_active_time,
                    "total_time": total_time,
                    "gpu_utilization": gpu_utilization,
                    "processed_count": processed_count,
                    "gpu_activity_periods": gpu_activity_periods
                })
                break
            
            # 解包任务数据
            request_id, input_data, gpu_time = message
            
            # 执行GPU计算
            gpu_start = time.time()
            result = simulate_gpu_computation(input_data, gpu_time)
            gpu_end = time.time()
            
            # 记录GPU活动时间段
            gpu_activity_periods.append((gpu_start, gpu_end))
            
            # 更新GPU活跃时间
            gpu_active_time += (gpu_end - gpu_start)
            processed_count += 1
            
            # 发送结果
            sender.send_pyobj((request_id, result))
    
    except Exception as e:
        print(f"GPU工作进程错误: {e}")
    
    finally:
        # 关闭连接
        receiver.close()
        sender.close()
        context.term()
        print("GPU工作进程已终止")

# CPU进程：生成请求，预处理，发送到GPU，接收结果，后处理
def zmq_approach(num_requests=10, matrix_size=100, 
                preprocess_time=0.05, gpu_time=0.1, postprocess_time=0.05):
    """使用ZeroMQ分离CPU和GPU操作"""
    print("\n运行ZeroMQ方式（分离CPU和GPU操作）...")
    
    # 启动GPU工作进程
    gpu_process = multiprocessing.Process(target=gpu_worker)
    gpu_process.daemon = True
    gpu_process.start()
    
    # 给GPU进程一点时间启动
    time.sleep(0.5)
    
    # 创建ZeroMQ上下文
    context = zmq.Context()
    
    # 设置发送输入数据的PUSH socket
    sender = context.socket(zmq.PUSH)
    sender.connect("tcp://localhost:5555")
    
    # 设置接收结果的PULL socket
    receiver = context.socket(zmq.PULL)
    receiver.connect("tcp://localhost:5556")
    
    # 开始计时
    start_time = time.time()
    
    # 创建结果字典
    results = {}
    
    # 记录CPU活动时间段
    cpu_activity_periods = []
    
    # 启动预处理和发送线程
    def preprocess_and_send():
        for i in range(num_requests):
            request_id = f"req_{i}"
            
            # CPU预处理开始
            cpu_start = time.time()
            
            # CPU预处理
            input_data = simulate_cpu_preprocessing(request_id, matrix_size, preprocess_time)
            
            # CPU预处理结束
            cpu_end = time.time()
            
            # 记录CPU预处理时间段
            cpu_activity_periods.append(("preprocess", cpu_start, cpu_end))
            
            # 发送到GPU进程
            sender.send_pyobj((request_id, input_data, gpu_time))
            
            # 简单的进度显示
            if (i + 1) % 10 == 0 or i == num_requests - 1:
                print(f"已发送 {i + 1}/{num_requests} 个请求")
    
    send_thread = threading.Thread(target=preprocess_and_send)
    send_thread.daemon = True
    send_thread.start()
    
    # 接收结果和后处理
    for _ in tqdm(range(num_requests), desc="接收和处理结果"):
        # 接收结果
        message = receiver.recv_pyobj()
        
        # 处理结果
        if isinstance(message, dict) and message.get("type") == "STATS":
            # 这是GPU进程发送的统计信息
            gpu_stats = message
            continue
        
        request_id, result = message
        
        # CPU后处理开始
        cpu_start = time.time()
        
        # CPU后处理
        final_result = simulate_cpu_postprocessing(request_id, result, postprocess_time)
        
        # CPU后处理结束
        cpu_end = time.time()
        
        # 记录CPU后处理时间段
        cpu_activity_periods.append(("postprocess", cpu_start, cpu_end))
        
        # 存储结果
        results[request_id] = final_result
    
    # 发送终止信号给GPU进程
    sender.send_pyobj("DONE")
    
    # 接收GPU统计信息
    gpu_stats = receiver.recv_pyobj()
    
    # 等待GPU进程终止
    gpu_process.join(timeout=1)
    
    # 结束计时
    end_time = time.time()
    total_time = end_time - start_time
    
    # 计算CPU-GPU重叠时间
    gpu_periods = gpu_stats['gpu_activity_periods']
    overlap_time = calculate_overlap(cpu_activity_periods, gpu_periods)
    overlap_percentage = (overlap_time / total_time) * 100
    
    # 关闭ZeroMQ连接
    sender.close()
    receiver.close()
    context.term()
    
    print(f"ZeroMQ方式完成时间: {total_time:.2f} 秒")
    print(f"GPU活跃时间: {gpu_stats['gpu_active_time']:.2f} 秒")
    print(f"CPU活跃时间: {sum([end-start for _, start, end in cpu_activity_periods]):.2f} 秒")
    print(f"CPU-GPU重叠时间: {overlap_time:.2f} 秒 ({overlap_percentage:.2f}%)")
    print(f"GPU利用率: {gpu_stats['gpu_utilization']:.2f}%")
    
    return total_time, gpu_stats['gpu_utilization'], overlap_percentage

def calculate_overlap(cpu_periods, gpu_periods):
    """计算CPU和GPU活动时间的重叠部分"""
    # 将CPU预处理和后处理时间段合并为单一列表
    cpu_time_ranges = [(start, end) for _, start, end in cpu_periods]
    
    # 初始化重叠时间
    total_overlap = 0.0
    
    # 对每个GPU时间段，计算与所有CPU时间段的重叠
    for gpu_start, gpu_end in gpu_periods:
        for cpu_start, cpu_end in cpu_time_ranges:
            # 计算重叠部分
            overlap_start = max(gpu_start, cpu_start)
            overlap_end = min(gpu_end, cpu_end)
            
            # 如果有重叠，累加重叠时间
            if overlap_end > overlap_start:
                total_overlap += (overlap_end - overlap_start)
    
    return total_overlap

def compare_performance(num_requests=50):
    """比较传统方式和ZeroMQ方式的性能"""
    # 设置参数
    matrix_size = 100
    preprocess_time = 0.05  # CPU预处理时间
    gpu_time = 0.1         # GPU计算时间
    postprocess_time = 0.05  # CPU后处理时间
    
    # 运行传统方式
    trad_time, trad_gpu_util = traditional_approach(
        num_requests, matrix_size, preprocess_time, gpu_time, postprocess_time
    )
    
    # 运行ZeroMQ方式
    zmq_time, zmq_gpu_util, overlap_percentage = zmq_approach(
        num_requests, matrix_size, preprocess_time, gpu_time, postprocess_time
    )
    
    # 计算加速比
    speedup = trad_time / zmq_time
    
    print("\n===== 性能对比 =====")
    print(f"传统方式执行时间: {trad_time:.2f} 秒, GPU利用率: {trad_gpu_util:.2f}%")
    print(f"ZeroMQ方式执行时间: {zmq_time:.2f} 秒, GPU利用率: {zmq_gpu_util:.2f}%")
    print(f"CPU-GPU重叠比例: {overlap_percentage:.2f}%")
    print(f"加速比: {speedup:.2f}x")
    print(f"GPU利用率提升: {zmq_gpu_util - trad_gpu_util:.2f}%")
    
    # 绘制性能对比图
    fig = Figure(figsize=(15, 5))
    canvas = FigureCanvas(fig)
    
    # 执行时间对比
    ax1 = fig.add_subplot(131)
    methods = ['Traditional', 'ZeroMQ']
    times = [trad_time, zmq_time]
    
    ax1.bar(methods, times, color=['blue', 'green'])
    ax1.set_ylabel('Execution Time (seconds)')
    ax1.set_title('Execution Time Comparison')
    
    # 添加数值标签
    for i, v in enumerate(times):
        ax1.text(i, v + 0.1, f"{v:.2f}s", ha='center')
    
    # 添加加速比
    ax1.text(0.5, max(times) * 0.5, f"Speedup: {speedup:.2f}x", 
            ha='center', fontsize=12, bbox=dict(facecolor='white', alpha=0.8))
    
    # GPU利用率对比
    ax2 = fig.add_subplot(132)
    utils = [trad_gpu_util, zmq_gpu_util]
    
    ax2.bar(methods, utils, color=['blue', 'green'])
    ax2.set_ylabel('GPU Utilization (%)')
    ax2.set_title('GPU Utilization Comparison')
    ax2.set_ylim(0, 100)
    
    # 添加数值标签
    for i, v in enumerate(utils):
        ax2.text(i, v + 1, f"{v:.2f}%", ha='center')
    
    # 添加利用率提升
    ax2.text(0.5, 50, f"Improvement: {zmq_gpu_util - trad_gpu_util:.2f}%", 
            ha='center', fontsize=12, bbox=dict(facecolor='white', alpha=0.8))
    
    # CPU-GPU重叠比例
    ax3 = fig.add_subplot(133)
    ax3.pie([overlap_percentage, 100-overlap_percentage], 
            labels=['Overlap', 'Non-overlap'],
            colors=['green', 'lightgray'],
            autopct='%1.1f%%',
            startangle=90)
    ax3.set_title('CPU-GPU Overlap Percentage')
    ax3.axis('equal')  # Equal aspect ratio ensures that pie is drawn as a circle
    
    fig.tight_layout()
    fig.savefig("vllm_simulation_comparison.png")
    print("性能对比图已保存为 vllm_simulation_comparison.png")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='模拟vLLM架构的简化版本')
    parser.add_argument('--requests', type=int, default=50, help='请求数量')
    args = parser.parse_args()
    
    # 比较性能
    compare_performance(args.requests)