File tree
- Lib
- asyncio
- test/test_asyncio
- Misc/NEWS.d/next/Library
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -6,6 +6,7 @@ | |||
| 6 | 6 | ||
| 7 | 7 | from . import exceptions | |
| 8 | 8 | from . import mixins | |
| 9 | + from . import tasks | ||
| 9 | 10 | ||
| 10 | 11 | ||
| 11 | 12 | class _ContextManagerMixin: | |
@@ -350,6 +351,7 @@ def __init__(self, value=1, *, loop=mixins._marker): | |||
| 350 | 351 | raise ValueError("Semaphore initial value must be >= 0") | |
| 351 | 352 | self._value = value | |
| 352 | 353 | self._waiters = collections.deque() | |
| 354 | + self._wakeup_scheduled = False | ||
| 353 | 355 | ||
| 354 | 356 | def __repr__(self): | |
| 355 | 357 | res = super().__repr__() | |
@@ -363,6 +365,7 @@ def _wake_up_next(self): | |||
| 363 | 365 | waiter = self._waiters.popleft() | |
| 364 | 366 | if not waiter.done(): | |
| 365 | 367 | waiter.set_result(None) | |
| 368 | + self._wakeup_scheduled = True | ||
| 366 | 369 | return | |
| 367 | 370 | ||
| 368 | 371 | def locked(self): | |
@@ -378,16 +381,17 @@ async def acquire(self): | |||
| 378 | 381 | called release() to make it larger than 0, and then return | |
| 379 | 382 | True. | |
| 380 | 383 | """ | |
| 381 | - while self._value <= 0: | ||
| 384 | + # _wakeup_scheduled is set if *another* task is scheduled to wakeup | ||
| 385 | + # but its acquire() is not resumed yet | ||
| 386 | + while self._wakeup_scheduled or self._value <= 0: | ||
| 382 | 387 | fut = self._get_loop().create_future() | |
| 383 | 388 | self._waiters.append(fut) | |
| 384 | 389 | try: | |
| 385 | 390 | await fut | |
| 386 | - except: | ||
| 387 | - # See the similar code in Queue.get. | ||
| 388 | - fut.cancel() | ||
| 389 | - if self._value > 0 and not fut.cancelled(): | ||
| 390 | - self._wake_up_next() | ||
| 391 | + # reset _wakeup_scheduled *after* waiting for a future | ||
| 392 | + self._wakeup_scheduled = False | ||
| 393 | + except exceptions.CancelledError: | ||
| 394 | + self._wake_up_next() | ||
| 391 | 395 | raise | |
| 392 | 396 | self._value -= 1 | |
| 393 | 397 | return True | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -933,6 +933,32 @@ async def test_release_no_waiters(self): | |||
| 933 | 933 | sem.release() | |
| 934 | 934 | self.assertFalse(sem.locked()) | |
| 935 | 935 | ||
| 936 | + async def test_acquire_fifo_order(self): | ||
| 937 | + sem = asyncio.Semaphore(1) | ||
| 938 | + result = [] | ||
| 939 | + | ||
| 940 | + async def coro(tag): | ||
| 941 | + await sem.acquire() | ||
| 942 | + result.append(f'{tag}_1') | ||
| 943 | + await asyncio.sleep(0.01) | ||
| 944 | + sem.release() | ||
| 945 | + | ||
| 946 | + await sem.acquire() | ||
| 947 | + result.append(f'{tag}_2') | ||
| 948 | + await asyncio.sleep(0.01) | ||
| 949 | + sem.release() | ||
| 950 | + | ||
| 951 | + t1 = asyncio.create_task(coro('c1')) | ||
| 952 | + t2 = asyncio.create_task(coro('c2')) | ||
| 953 | + t3 = asyncio.create_task(coro('c3')) | ||
| 954 | + | ||
| 955 | + await asyncio.gather(t1, t2, t3) | ||
| 956 | + | ||
| 957 | + self.assertEqual( | ||
| 958 | + ['c1_1', 'c2_1', 'c3_1', 'c1_2', 'c2_2', 'c3_2'], | ||
| 959 | + result | ||
| 960 | + ) | ||
| 961 | + | ||
| 936 | 962 | ||
| 937 | 963 | if __name__ == '__main__': | |
| 938 | 964 | unittest.main() | |
Lines changed: 1 addition & 0 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1 @@ | |||
| 1 | + Fix :class:`asyncio.Semaphore` re-aquiring FIFO order. | ||
0 commit comments