| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -378,6 +378,7 @@ def __init__(self, value=1, *, loop=None): | |||
| 378 | 378 | warnings.warn("The loop argument is deprecated since Python 3.8, " | |
| 379 | 379 | "and scheduled for removal in Python 3.10.", | |
| 380 | 380 | DeprecationWarning, stacklevel=2) | |
| 381 | + self._wakeup_scheduled = False | ||
| 381 | 382 | ||
| 382 | 383 | def __repr__(self): | |
| 383 | 384 | res = super().__repr__() | |
@@ -391,6 +392,7 @@ def _wake_up_next(self): | |||
| 391 | 392 | waiter = self._waiters.popleft() | |
| 392 | 393 | if not waiter.done(): | |
| 393 | 394 | waiter.set_result(None) | |
| 395 | + self._wakeup_scheduled = True | ||
| 394 | 396 | return | |
| 395 | 397 | ||
| 396 | 398 | def locked(self): | |
@@ -406,16 +408,17 @@ async def acquire(self): | |||
| 406 | 408 | called release() to make it larger than 0, and then return | |
| 407 | 409 | True. | |
| 408 | 410 | """ | |
| 409 | - while self._value <= 0: | ||
| 411 | + # _wakeup_scheduled is set if *another* task is scheduled to wakeup | ||
| 412 | + # but its acquire() is not resumed yet | ||
| 413 | + while self._wakeup_scheduled or self._value <= 0: | ||
| 410 | 414 | fut = self._loop.create_future() | |
| 411 | 415 | self._waiters.append(fut) | |
| 412 | 416 | try: | |
| 413 | 417 | await fut | |
| 414 | - except: | ||
| 415 | - # See the similar code in Queue.get. | ||
| 416 | - fut.cancel() | ||
| 417 | - if self._value > 0 and not fut.cancelled(): | ||
| 418 | - self._wake_up_next() | ||
| 418 | + # reset _wakeup_scheduled *after* waiting for a future | ||
| 419 | + self._wakeup_scheduled = False | ||
| 420 | + except exceptions.CancelledError: | ||
| 421 | + self._wake_up_next() | ||
| 419 | 422 | raise | |
| 420 | 423 | self._value -= 1 | |
| 421 | 424 | return True | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -961,6 +961,32 @@ async def test_release_no_waiters(self): | |||
| 961 | 961 | sem.release() | |
| 962 | 962 | self.assertFalse(sem.locked()) | |
| 963 | 963 | ||
| 964 | + async def test_acquire_fifo_order(self): | ||
| 965 | + sem = asyncio.Semaphore(1) | ||
| 966 | + result = [] | ||
| 967 | + | ||
| 968 | + async def coro(tag): | ||
| 969 | + await sem.acquire() | ||
| 970 | + result.append(f'{tag}_1') | ||
| 971 | + await asyncio.sleep(0.01) | ||
| 972 | + sem.release() | ||
| 973 | + | ||
| 974 | + await sem.acquire() | ||
| 975 | + result.append(f'{tag}_2') | ||
| 976 | + await asyncio.sleep(0.01) | ||
| 977 | + sem.release() | ||
| 978 | + | ||
| 979 | + t1 = asyncio.create_task(coro('c1')) | ||
| 980 | + t2 = asyncio.create_task(coro('c2')) | ||
| 981 | + t3 = asyncio.create_task(coro('c3')) | ||
| 982 | + | ||
| 983 | + await asyncio.gather(t1, t2, t3) | ||
| 984 | + | ||
| 985 | + self.assertEqual( | ||
| 986 | + ['c1_1', 'c2_1', 'c3_1', 'c1_2', 'c2_2', 'c3_2'], | ||
| 987 | + result | ||
| 988 | + ) | ||
| 989 | + | ||
| 964 | 990 | ||
| 965 | 991 | if __name__ == '__main__': | |
| 966 | 992 | unittest.main() | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1 @@ | |||
| 1 | + Fix :class:`asyncio.Semaphore` re-aquiring FIFO order. | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments