From 066333a9448bd60479c401b17c6aad04d4fcdb9f Mon Sep 17 00:00:00 2001 From: Youzhi Luo Date: Thu, 11 Jun 2026 16:37:46 +0000 Subject: [PATCH] fix(verl): fire all tasks in the batch concurrently instead of capped by semaphore --- .../backends/verl/loop_manager.py | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/src/agentcore_rl_toolkit/backends/verl/loop_manager.py b/src/agentcore_rl_toolkit/backends/verl/loop_manager.py index 9c439c7..d874f09 100644 --- a/src/agentcore_rl_toolkit/backends/verl/loop_manager.py +++ b/src/agentcore_rl_toolkit/backends/verl/loop_manager.py @@ -160,19 +160,13 @@ async def _dispatch_and_collect( """ max_time = self.config.actor_rollout_ref.rollout.agentcore.max_rollout_time target_size = len(rollout_batch_input) - max_concurrent = self.config.actor_rollout_ref.rollout.agentcore.max_pool_connections - semaphore = asyncio.Semaphore(max_concurrent) async_client = AsyncGatewayClient(self._gateway_url) start_time = time.time() - async def _run_with_semaphore(payload, uid): - async with semaphore: - return await self._process_single_task( - async_client, payload, uid, rollout_sampling_params or {}, max_time - ) - + # All tasks in the batch are fired concurrently; the ACR TPS limiter inside + # RolloutClient controls AgentCore invoke request rate. tasks = [] for i in range(target_size): non_tensor_item = rollout_batch_input[i].non_tensor_batch @@ -185,7 +179,11 @@ async def _run_with_semaphore(payload, uid): ) uid = payload.pop("uid") - tasks.append(asyncio.create_task(_run_with_semaphore(payload, uid))) + tasks.append( + asyncio.create_task( + self._process_single_task(async_client, payload, uid, rollout_sampling_params or {}, max_time) + ) + ) try: results = await asyncio.gather(*tasks, return_exceptions=True)