gh-155233: Fix asyncio.Barrier reusing a cancelled task's index

[sreehariannam]

Fixes gh-155233.

Barrier.wait() is documented to return "a unique and individual index number from 0 to parties-1" for each task that passes the barrier together. If a task's wait() is cancelled while the barrier is still filling (i.e. before enough parties have arrived for that round), a task that arrives afterward can be assigned the same index as another task already waiting in that same round:

import asyncio

async def main():
    b = asyncio.Barrier(3)
    results = []

    async def party(name):
        try:
            idx = await b.wait()
            results.append((name, idx))
        except asyncio.CancelledError:
            results.append((name, "cancelled"))
            raise

    t1 = asyncio.create_task(party("A"))
    t2 = asyncio.create_task(party("B"))
    await asyncio.sleep(0)
    await asyncio.sleep(0)

    t1.cancel()  # A leaves while the barrier is still filling
    try:
        await t1
    except asyncio.CancelledError:
        pass
    await asyncio.sleep(0)

    t3 = asyncio.create_task(party("C"))
    t4 = asyncio.create_task(party("D"))  # completes the round of 3
    await asyncio.gather(t2, t3, t4)

    print(results)

asyncio.run(main())

On current main:

[('A', 'cancelled'), ('D', 2), ('B', 1), ('C', 1)]

B and C both get index 1; index 0 is never handed out among the three tasks (B, C, D) that actually pass the barrier together. Full analysis and root cause are in gh-155233.

Root cause

index = self._count was computed eagerly at arrival, before a round's final membership was known. self._count is also decremented when a party leaves early (cancellation) during filling, so a later arrival could land on the exact value an already-waiting task had already captured as its own index.

Fix

Indices can only be assigned correctly once a round's exact release cohort is known, so this defers assignment to release time:

• Each arriving task appends a unique "ticket" (a plain object()) to self._present, an ordered list of currently-present parties for the round.
• _release() (called by the last arriving task) builds self._release_index = {ticket: i for i, ticket in enumerate(self._present)} — a snapshot of arrival order for exactly the parties in that round — before any of them get a chance to leave.
• Each task looks up its own index from that mapping once released.
• self._count/n_waiting/reset()'s occupancy check are now derived from len(self._present) instead of a separately-tracked int.

threading.Barrier doesn't need equivalent handling since a synchronous thread can't be cancelled out of a blocking wait the way an asyncio.Task can, so this is specific to the asyncio implementation.

Testing

• Added test_filling_tasks_cancel_one_index_not_reused, which fails against the old implementation ([1, 1, 2] instead of [0, 1, 2]) and passes against the fix.
• Updated one existing test (test_reset_barrier_when_tasks_half_draining) that reached into the now-renamed private barrier._count attribute.
• Full test.test_asyncio suite (2783 tests) passes.