Synchronize Coroutines with asyncio.Barrier

Python's asyncio.Barrier is a synchronization primitive introduced in Python 3.11 that forces a predefined number of concurrent coroutines to block until all of them reach a common rendezvous point. This article covers the mechanics of asyncio.Barrier, how the wait() method orchestrates simultaneous execution, how to run an optional completion action, and how to handle error conditions such as broken barriers.

Core Mechanics of asyncio.Barrier

When initializing an asyncio.Barrier(parties, action=None), you define a fixed threshold via the parties argument. This number specifies exactly how many coroutines must call the barrier's wait() method before any of them are permitted to advance.

Each time a coroutine executes await barrier.wait(), the barrier performs internal accounting:

  1. It registers the arriving coroutine and decrements the count of remaining required parties.
  2. If the count of arrived coroutines is less than parties, the coroutine pauses execution and yields control back to the event loop.
  3. When the final (\(N\)-th) coroutine invokes wait(), the barrier trips open.
  4. If an action callable was provided at initialization, one of the coroutines executes this callback while the others continue waiting.
  5. All blocked coroutines are simultaneously woken up, returning an integer index representing their arrival order (from 0 to parties - 1).
  6. The barrier automatically resets to its initial state, allowing it to be reused for subsequent synchronization cycles.

Practical Example

The following example demonstrates three independent worker coroutines performing phase-based work. Neither worker can proceed to Phase 2 until all workers have completed Phase 1.

import asyncio

async def worker(barrier: asyncio.Barrier, worker_id: int):
    print(f"Worker {worker_id}: Completing Phase 1...")
    await asyncio.sleep(worker_id * 0.5)  # Simulate variable execution time
    print(f"Worker {worker_id}: Reached the rendezvous point.")
    
    # Wait for all workers to arrive
    arrival_index = await barrier.wait()
    
    print(f"Worker {worker_id}: Passing barrier (arrival order: {arrival_index}). Starting Phase 2...")

async def main():
    parties = 3
    
    # Optional action executed once all parties arrive
    def on_release():
        print("--- All workers arrived. Barrier released. ---")

    barrier = asyncio.Barrier(parties, action=on_release)

    async with asyncio.TaskGroup() as tg:
        for i in range(parties):
            tg.create_task(worker(barrier, i + 1))

if __name__ == "__main__":
    asyncio.run(main())

Error Handling and State Management

asyncio.Barrier tracks state to prevent deadlocks when tasks fail or get canceled: