Function run_load_test

tests/ce/performance/stress_tools.py:90–118

()

# ============ 主流程 ============
async def run_load_test():
    prompts = await load_data()
    queue = asyncio.Queue(maxsize=MAX_CONCURRENCY * 5)  # 限制队列大小，降低内存占用
    counter = Counter()
    latencies = []

    connector = aiohttp.TCPConnector(limit=MAX_CONCURRENCY * 2)  # 限制TCP连接
    timeout = aiohttp.ClientTimeout(total=TIMEOUT)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        with tqdm(total=TOTAL_REQUESTS, desc="压测进度") as pbar:
            # 启动 Worker
            workers = [
                asyncio.create_task(worker(f"W{i}", session, prompts, counter, latencies, pbar, queue))
                for i in range(MAX_CONCURRENCY)
            ]

            # 边生产边消费
            for i in range(TOTAL_REQUESTS):
                await queue.put(i)

            # 发送毒丸让 worker 退出
            for _ in workers:
                await queue.put(None)

            await queue.join()
            await asyncio.gather(*workers)

    generate_report(counter, latencies)


# ============ 报告输出 ============