Function _dequeue_only_task

tests/benchmark/micro_benchmark.py:233–251

(args: tuple[str, str, str])

def _dequeue_only_task(args: tuple[str, str, str]) -> bool:
    store_url, worker_id, task_id = args
    console.print(f"[Dequeue-Only Task {task_id}] Dequeueing rollout for worker {worker_id}")
    store = agl.LightningStoreClient(store_url)

    async def _async_task() -> bool:
        attempted = await store.dequeue_rollout()  # no worker_id
        if attempted is None:
            console.print(f"[Dequeue-Only Task {task_id}] No rollout available to dequeue")
            return False
        return True

    try:
        return asyncio.run(_async_task())
    except Exception as e:
        console.print(f"Error dequeueing only worker {worker_id} for task {task_id}: {e}")
        return False
    finally:
        _close_store_client(store)


def dequeue_rollouts(store_url: str) -> BenchmarkSummary: