Skip to content

Commit 49c43f7

Browse files
YvesDupblurb-it[bot]asvetlov
authored
gh-155233: Fix asyncio.Barrier reusing a cancelled task's index (alternative to GH-155235) (#158957)
* Initial commit: - add a private ``_index`` attr - add a test Co-authored-by: Sreehari Annam * 📜🤖 Added by blurb_it. * Apply suggestions from code review * Update Lib/asyncio/locks.py * Update Lib/asyncio/locks.py --------- Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com> Co-authored-by: Andrew Svetlov <andrew.svetlov@gmail.com>
1 parent 429fd0a commit 49c43f7

3 files changed

Lines changed: 38 additions & 2 deletions

File tree

‎Lib/asyncio/locks.py‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -489,6 +489,8 @@ def __init__(self, parties):
489489
self._parties = parties
490490
self._state = _BarrierState.FILLING
491491
self._count = 0 # count tasks in Barrier
492+
# index number returned when barrier drains
493+
self._index = self._parties - 1
492494

493495
def __repr__(self):
494496
res = super().__repr__()
@@ -515,13 +517,14 @@ async def wait(self):
515517
async with self._cond:
516518
await self._block() # Block while the barrier drains or resets.
517519
try:
518-
index = self._count
519520
self._count += 1
520-
if index + 1 == self._parties:
521+
if self._count == self._parties:
521522
# We release the barrier
522523
await self._release()
523524
else:
524525
await self._wait()
526+
index = self._index
527+
self._index = (self._index + 1) % self._parties
525528
return index
526529
finally:
527530
self._count -= 1
@@ -569,6 +572,7 @@ def _exit(self):
569572
if self._count == 0:
570573
if self._state in (_BarrierState.RESETTING, _BarrierState.DRAINING):
571574
self._state = _BarrierState.FILLING
575+
self._index = self._parties - 1
572576
self._cond.notify_all()
573577

574578
async def reset(self):
@@ -584,6 +588,7 @@ async def reset(self):
584588
self._state = _BarrierState.RESETTING
585589
else:
586590
self._state = _BarrierState.FILLING
591+
self._index = self._parties - 1
587592
self._cond.notify_all()
588593

589594
async def abort(self):

‎Lib/test/test_asyncio/test_locks.py‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1820,6 +1820,36 @@ async def coro():
18201820

18211821
self.assertEqual(barrier1.n_waiting, 0)
18221822

1823+
async def test_filling_tasks_cancel_one_index_still_unique(self):
1824+
# See gh-155233: a task cancelled while the barrier is still
1825+
# filling used to leave its index available for reuse.
1826+
# Index counter is now calculated on demand during the draining phasis.
1827+
self.N = 3
1828+
barrier = asyncio.Barrier(self.N)
1829+
results = []
1830+
1831+
async def coro():
1832+
i = await barrier.wait()
1833+
results.append(i)
1834+
1835+
t1 = asyncio.create_task(coro())
1836+
t2 = asyncio.create_task(coro())
1837+
await asyncio.sleep(0)
1838+
self.assertEqual(barrier.n_waiting, 2)
1839+
1840+
t1.cancel()
1841+
with self.assertRaises(asyncio.CancelledError):
1842+
await t1
1843+
await asyncio.sleep(0)
1844+
self.assertEqual(barrier.n_waiting, 1)
1845+
1846+
t3 = asyncio.create_task(coro())
1847+
t4 = asyncio.create_task(coro())
1848+
await asyncio.gather(t2, t3, t4)
1849+
1850+
self.assertEqual(sorted(results), list(range(self.N)))
1851+
self.assertEqual(barrier.n_waiting, 0)
1852+
18231853

18241854
if __name__ == '__main__':
18251855
unittest.main()
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Fix :meth:`asyncio.Barrier.wait` returning duplicate indices when a waiting task is cancelled before the barrier is released.```

0 commit comments

Comments
 (0)