【Bug已解决】[AsyncGRPO] aiohttp limits to 100 reqs when max_inflight_tasks ＞ 100 解决方案

【Bug已解决】[AsyncGRPO] aiohttp limits to 100 reqs when max_inflight_tasks 100 解决方案原始报错[AsyncGRPO] aiohttp limits to 100 reqs when max_inflight_tasks 100 场景异步 GRPO 用 aiohttp 作为 HTTP 客户端向推理引擎并发发请求rollout 生成。配置里max_inflight_tasks设到 200期望同时跑 200 个并发任务但实际同时只有约 100 个在飞其余排队等待。根因是 aiohttp 的ClientSession默认连接池上限就是 100TCPConnector(limit100)并发超过 100 就被池子卡住第 101 个得等前面的释放。要支持更多 inflight必须显式调高连接池上限。 关键词aiohttp、连接池上限、TCPConnector、limit、concurrency、max_inflight_tasks、信号量、异步并发、100 默认限制。一、现象长什么样并发上不去卡在 100配置max_inflight_tasks 200期望 200 个任务同时向引擎发请求实际观测并发数稳定在 ~100多的任务在排队吞吐上不去用asyncio跑理论上能并发 200但 aiohttp 这层卡住调大max_inflight_tasks没用瓶颈不在任务调度而在 HTTP 连接池改小到 100 以内一切正常一过 100 就出现第 101 个起开始等日志/报错没有明显的连接超限只是默默排队——典型隐性限制。核心问题aiohttp 默认连接池上限 100成了并发天花板与max_inflight_tasks脱节。二、背景aiohttp 的默认连接池上限为什么是 100aiohttp 的ClientSession内部有一个TCPConnector管理连接复用它有两个默认上限limit整个连接池的总并发连接数默认100limit_per_host单个 host 的并发连接数默认也是100实际上是limit的值旧版默认 100。当你的 200 个任务都打同一个推理引擎 host并发连接请求超过limit/limit_per_host多出来的请求会在连接器里排队等已有连接释放。所以即使你用asyncio起了 200 个协程真正同时建连的只有 100 个。这不是 bug是默认值保守——避免无节制建连打爆服务端。但当你明确要跑max_inflight_tasks200就必须把连接池上限调到匹配值否则max_inflight_tasks形同虚设。三、根因连接池上限与 max_inflight_tasks 脱节根因拆解默认 100ClientSession()没传connector用默认limit100未联动配置max_inflight_tasks调大但TCPConnector.limit没跟着调单 host 上限全部请求打同一引擎 hostlimit_per_host也卡在 100隐性排队超 100 的请求静默排队无报错难发现吞吐天花板并发被池子限死加任务不加速误调任务数开发者调大max_inflight_tasks以为够了瓶颈在更底层。下面用最小模型复现并发被池子上限卡在 100再给调高连接池上限的修复。四、最小可运行复现import asyncio class FakeConnector: def __init__(self, limit100): self.limit limit self._sem asyncio.Semaphore(limit) # 模拟连接池上限 async def request(self, i): async with self._sem: # 同时最多 limit 个 await asyncio.sleep(0.01) return fresp{i} async def run(connector, n): # 错连接池默认 100n200 时只有 100 同时飞 tasks [asyncio.create_task(connector.request(i)) for i in range(n)] results await asyncio.gather(*tasks) return results if __name__ __main__: # 用信号量计数观测真实并发峰值 peak {v: 0, cur: 0} async def instrumented(connector, n): async def one(i): peak[cur] 1 peak[v] max(peak[v], peak[cur]) try: return await connector.request(i) finally: peak[cur] - 1 await asyncio.gather(*[one(i) for i in range(n)]) asyncio.run(instrumented(FakeConnector(limit100), 200)) print(默认 limit100 时真实并发峰值:, peak[v]) # 100被卡运行可见并发峰值停在 100正是 aiohttp 默认上限的现场。五、方案显式调高 TCPConnector 的 limit 与 limit_per_host第一层创建ClientSession时传自定义TCPConnector把limit和limit_per_host调到与max_inflight_tasks匹配import aiohttp def make_session(max_inflight_tasks): # 把连接池上限与 max_inflight_tasks 联动 connector aiohttp.TCPConnector( limitmax_inflight_tasks, # 总连接上限 limit_per_hostmax_inflight_tasks, # 单 host 上限推理引擎通常单 host ) return aiohttp.ClientSession(connectorconnector) # 真实用法示意 # session make_session(max_inflight_tasks200) # async with session.post(ENGINE_URL, jsonpayload) as resp: ...limit与limit_per_host都设为 200200 个并发连接不再被池子卡住。六、方案用信号量显式控制 inflight与连接池一致第二层除了连接池再在任务调度层用asyncio.Semaphore(max_inflight_tasks)显式限制同时发起的请求数两层上限保持一致避免任务层想发 200、连接池只放 100的脱节async def bounded_request(session, sem, i, payload): async with sem: # 与连接池上限一致 async with session.post(ENGINE_URL, jsonpayload) as resp: return await resp.json() def build_semaphore(max_inflight_tasks): return asyncio.Semaphore(max_inflight_tasks) # 启动 200 个任务信号量保证同时不超过 200连接池也放 200二者匹配信号量在任务发起层兜底连接池在建连层兜底双层一致才不脱节。七、方案配置一处指定避免两处不一致第三层把max_inflight_tasks作为单一真相连接池 limit、信号量都从它派生杜绝两处各写各的数字CONFIG {max_inflight_tasks: 200} def build_runtime(): n CONFIG[max_inflight_tasks] connector aiohttp.TCPConnector(limitn, limit_per_hostn) session aiohttp.ClientSession(connectorconnector) sem asyncio.Semaphore(n) return session, sem # 任何地方想改并发只改 CONFIG[max_inflight_tasks] 一处单一来源保证连接池上限 信号量 配置改一处全链路生效不会再出现配置 200、池子 100。八、验证把并发随 limit 提升锁进测试def test_default_cap_100(): peak {v: 0, cur: 0} async def inst(n): async def one(i): peak[cur] 1; peak[v] max(peak[v], peak[cur]) async with FakeConnector(limit100)._sem: await asyncio.sleep(0.001) peak[cur] - 1 await asyncio.gather(*[one(i) for i in range(n)]) asyncio.run(inst(200)) assert peak[v] 100 def test_raised_limit_allows_more(): peak {v: 0, cur: 0} async def inst(n, limit): sem asyncio.Semaphore(limit) async def one(i): peak[cur] 1; peak[v] max(peak[v], peak[cur]) async with sem: await asyncio.sleep(0.001) peak[cur] - 1 await asyncio.gather(*[one(i) for i in range(n)]) asyncio.run(inst(200, 200)) assert peak[v] 200 if __name__ __main__: test_default_cap_100() test_raised_limit_allows_more() print(aiohttp 并发上限测试通过。)九、排查清单并发卡在 100按顺序查默认池ClientSession 是否没传 connector用了默认 limit100是则卡。limit 联动TCPConnector.limit 是否随 max_inflight_tasks 调高没调则脱节。单 host请求是否都打同一引擎 hostlimit_per_host 也要调高。隐性排队超 100 的请求是否静默排队无报错是则隐性限制。双层一致任务层信号量与连接池上限是否一致不一致仍卡。配置单一max_inflight_tasks 是否单一来源派生各处上限否则易错配。峰值观测能否观测真实并发峰值能确认是否真被 100 卡住。十、小结aiohttp 在 max_inflight_tasks100 时限制到 100是aiohttp 默认连接池上限 100 成了并发天花板且与max_inflight_tasks配置脱节所有请求打同一引擎 hostTCPConnector默认limitlimit_per_host100超出的请求静默排队调大任务数无效。修复三层调高连接池创建ClientSession传TCPConnector(limit, limit_per_host)与max_inflight_tasks匹配信号量兜底任务调度层用asyncio.Semaphore(max_inflight_tasks)与连接池双层一致配置单一max_inflight_tasks作单一真相连接池与信号量都从它派生改一处全生效。核心原则异步 HTTP 客户端的并发上限由连接池决定不是由你的任务数决定。凡是设了 max_inflight_tasks100 却并发卡在 100的现场都应把 aiohttp 的TCPConnector.limit/limit_per_host调到匹配值并与任务信号量保持一致——让配置真正生效。