Mailman 3 March 2022 - Python-checkins

bpo-43352: Add a Barrier object in asyncio lib (GH-24903)
by asvetlov March 25, 2022

March 25, 2022

https://github.com/python/cpython/commit/d03acd7270d66ddb8e987f9743405147ec… commit: d03acd7270d66ddb8e987f9743405147ecc15087 branch: main author: Duprat <yduprat(a)gmail.com> committer: asvetlov <andrew.svetlov(a)gmail.com> date: 2022-03-26T00:01:21+02:00 summary: bpo-43352: Add a Barrier object in asyncio lib (GH-24903) Co-authored-by: Yury Selivanov <yury(a)edgedb.com> Co-authored-by: Andrew Svetlov <andrew.svetlov(a)gmail.com> files: A Misc/NEWS.d/next/Library/2021-03-31-15-22-45.bpo-43352.nSjMuE.rst M Doc/library/asyncio-api-index.rst M Doc/library/asyncio-sync.rst M Lib/asyncio/exceptions.py M Lib/asyncio/locks.py M Lib/test/test_asyncio/test_locks.py diff --git a/Doc/library/asyncio-api-index.rst b/Doc/library/asyncio-api-index.rst index 8bc7943a71739..a4e38e469d82f 100644 --- a/Doc/library/asyncio-api-index.rst +++ b/Doc/library/asyncio-api-index.rst @@ -186,11 +186,16 @@ Threading-like synchronization primitives that can be used in Tasks. * - :class:`BoundedSemaphore` - A bounded semaphore. + * - :class:`Barrier` + - A barrier object. + .. rubric:: Examples * :ref:`Using asyncio.Event <asyncio_example_sync_event>`. +* :ref:`Using asyncio.Barrier <asyncio_example_barrier>`. + * See also the documentation of asyncio :ref:`synchronization primitives <asyncio-sync>`. @@ -206,6 +211,9 @@ Exceptions * - :exc:`asyncio.CancelledError` - Raised when a Task is cancelled. See also :meth:`Task.cancel`. + * - :exc:`asyncio.BrokenBarrierError` + - Raised when a Barrier is broken. See also :meth:`Barrier.wait`. + .. rubric:: Examples diff --git a/Doc/library/asyncio-sync.rst b/Doc/library/asyncio-sync.rst index f4063db2ee86e..141733ee2c800 100644 --- a/Doc/library/asyncio-sync.rst +++ b/Doc/library/asyncio-sync.rst @@ -28,6 +28,7 @@ asyncio has the following basic synchronization primitives: * :class:`Condition` * :class:`Semaphore` * :class:`BoundedSemaphore` +* :class:`Barrier` --------- @@ -340,6 +341,115 @@ BoundedSemaphore .. versionchanged:: 3.10 Removed the *loop* parameter. + +Barrier +======= + +.. class:: Barrier(parties, action=None) + + A barrier object. Not thread-safe. + + A barrier is a simple synchronization primitive that allows to block until + *parties* number of tasks are waiting on it. + Tasks can wait on the :meth:`~Barrier.wait` method and would be blocked until + the specified number of tasks end up waiting on :meth:`~Barrier.wait`. + At that point all of the waiting tasks would unblock simultaneously. + + :keyword:`async with` can be used as an alternative to awaiting on + :meth:`~Barrier.wait`. + + The barrier can be reused any number of times. + + .. _asyncio_example_barrier: + + Example:: + + async def example_barrier(): + # barrier with 3 parties + b = asyncio.Barrier(3) + + # create 2 new waiting tasks + asyncio.create_task(b.wait()) + asyncio.create_task(b.wait()) + + await asyncio.sleep(0) + print(b) + + # The third .wait() call passes the barrier + await b.wait() + print(b) + print("barrier passed") + + await asyncio.sleep(0) + print(b) + + asyncio.run(example_barrier()) + + Result of this example is:: + + <asyncio.locks.Barrier object at 0x... [filling, waiters:2/3]> + <asyncio.locks.Barrier object at 0x... [draining, waiters:0/3]> + barrier passed + <asyncio.locks.Barrier object at 0x... [filling, waiters:0/3]> + + .. versionadded:: 3.11 + + .. coroutinemethod:: wait() + + Pass the barrier. When all the tasks party to the barrier have called + this function, they are all unblocked simultaneously. + + When a waiting or blocked task in the barrier is cancelled, + this task exits the barrier which stays in the same state. + If the state of the barrier is "filling", the number of waiting task + decreases by 1. + + The return value is an integer in the range of 0 to ``parties-1``, different + for each task. This can be used to select a task to do some special + housekeeping, e.g.:: + + ... + async with barrier as position: + if position == 0: + # Only one task print this + print('End of *draining phasis*') + + This method may raise a :class:`BrokenBarrierError` exception if the + barrier is broken or reset while a task is waiting. + It could raise a :exc:`CancelledError` if a task is cancelled. + + .. coroutinemethod:: reset() + + Return the barrier to the default, empty state. Any tasks waiting on it + will receive the :class:`BrokenBarrierError` exception. + + If a barrier is broken it may be better to just leave it and create a new one. + + .. coroutinemethod:: abort() + + Put the barrier into a broken state. This causes any active or future + calls to :meth:`wait` to fail with the :class:`BrokenBarrierError`. + Use this for example if one of the taks needs to abort, to avoid infinite + waiting tasks. + + .. attribute:: parties + + The number of tasks required to pass the barrier. + + .. attribute:: n_waiting + + The number of tasks currently waiting in the barrier while filling. + + .. attribute:: broken + + A boolean that is ``True`` if the barrier is in the broken state. + + +.. exception:: BrokenBarrierError + + This exception, a subclass of :exc:`RuntimeError`, is raised when the + :class:`Barrier` object is reset or broken. + --------- diff --git a/Lib/asyncio/exceptions.py b/Lib/asyncio/exceptions.py index c764c9ffcfc19..5ece595aad647 100644 --- a/Lib/asyncio/exceptions.py +++ b/Lib/asyncio/exceptions.py @@ -1,7 +1,8 @@ """asyncio exceptions.""" -__all__ = ('CancelledError', 'InvalidStateError', 'TimeoutError', +__all__ = ('BrokenBarrierError', + 'CancelledError', 'InvalidStateError', 'TimeoutError', 'IncompleteReadError', 'LimitOverrunError', 'SendfileNotAvailableError') @@ -55,3 +56,7 @@ def __init__(self, message, consumed): def __reduce__(self): return type(self), (self.args[0], self.consumed) + + +class BrokenBarrierError(RuntimeError): + """Barrier is broken by barrier.abort() call.""" diff --git a/Lib/asyncio/locks.py b/Lib/asyncio/locks.py index 9b4612197de1d..e71130274dd6f 100644 --- a/Lib/asyncio/locks.py +++ b/Lib/asyncio/locks.py @@ -1,14 +1,15 @@ """Synchronization primitives.""" -__all__ = ('Lock', 'Event', 'Condition', 'Semaphore', 'BoundedSemaphore') +__all__ = ('Lock', 'Event', 'Condition', 'Semaphore', + 'BoundedSemaphore', 'Barrier') import collections +import enum from . import exceptions from . import mixins from . import tasks - class _ContextManagerMixin: async def __aenter__(self): await self.acquire() @@ -416,3 +417,155 @@ def release(self): if self._value >= self._bound_value: raise ValueError('BoundedSemaphore released too many times') super().release() + + + +class _BarrierState(enum.Enum): + FILLING = 'filling' + DRAINING = 'draining' + RESETTING = 'resetting' + BROKEN = 'broken' + + +class Barrier(mixins._LoopBoundMixin): + """Asyncio equivalent to threading.Barrier + + Implements a Barrier primitive. + Useful for synchronizing a fixed number of tasks at known synchronization + points. Tasks block on 'wait()' and are simultaneously awoken once they + have all made their call. + """ + + def __init__(self, parties): + """Create a barrier, initialised to 'parties' tasks.""" + if parties < 1: + raise ValueError('parties must be > 0') + + self._cond = Condition() # notify all tasks when state changes + + self._parties = parties + self._state = _BarrierState.FILLING + self._count = 0 # count tasks in Barrier + + def __repr__(self): + res = super().__repr__() + extra = f'{self._state.value}' + if not self.broken: + extra += f', waiters:{self.n_waiting}/{self.parties}' + return f'<{res[1:-1]} [{extra}]>' + + async def __aenter__(self): + # wait for the barrier reaches the parties number + # when start draining release and return index of waited task + return await self.wait() + + async def __aexit__(self, *args): + pass + + async def wait(self): + """Wait for the barrier. + + When the specified number of tasks have started waiting, they are all + simultaneously awoken. + Returns an unique and individual index number from 0 to 'parties-1'. + """ + async with self._cond: + await self._block() # Block while the barrier drains or resets. + try: + index = self._count + self._count += 1 + if index + 1 == self._parties: + # We release the barrier + await self._release() + else: + await self._wait() + return index + finally: + self._count -= 1 + # Wake up any tasks waiting for barrier to drain. + self._exit() + + async def _block(self): + # Block until the barrier is ready for us, + # or raise an exception if it is broken. + # + # It is draining or resetting, wait until done + # unless a CancelledError occurs + await self._cond.wait_for( + lambda: self._state not in ( + _BarrierState.DRAINING, _BarrierState.RESETTING + ) + ) + + # see if the barrier is in a broken state + if self._state is _BarrierState.BROKEN: + raise exceptions.BrokenBarrierError("Barrier aborted") + + async def _release(self): + # Release the tasks waiting in the barrier. + + # Enter draining state. + # Next waiting tasks will be blocked until the end of draining. + self._state = _BarrierState.DRAINING + self._cond.notify_all() + + async def _wait(self): + # Wait in the barrier until we are released. Raise an exception + # if the barrier is reset or broken. + + # wait for end of filling + # unless a CancelledError occurs + await self._cond.wait_for(lambda: self._state is not _BarrierState.FILLING) + + if self._state in (_BarrierState.BROKEN, _BarrierState.RESETTING): + raise exceptions.BrokenBarrierError("Abort or reset of barrier") + + def _exit(self): + # If we are the last tasks to exit the barrier, signal any tasks + # waiting for the barrier to drain. + if self._count == 0: + if self._state in (_BarrierState.RESETTING, _BarrierState.DRAINING): + self._state = _BarrierState.FILLING + self._cond.notify_all() + + async def reset(self): + """Reset the barrier to the initial state. + + Any tasks currently waiting will get the BrokenBarrier exception + raised. + """ + async with self._cond: + if self._count > 0: + if self._state is not _BarrierState.RESETTING: + #reset the barrier, waking up tasks + self._state = _BarrierState.RESETTING + else: + self._state = _BarrierState.FILLING + self._cond.notify_all() + + async def abort(self): + """Place the barrier into a 'broken' state. + + Useful in case of error. Any currently waiting tasks and tasks + attempting to 'wait()' will have BrokenBarrierError raised. + """ + async with self._cond: + self._state = _BarrierState.BROKEN + self._cond.notify_all() + + @property + def parties(self): + """Return the number of tasks required to trip the barrier.""" + return self._parties + + @property + def n_waiting(self): + """Return the number of tasks currently waiting at the barrier.""" + if self._state is _BarrierState.FILLING: + return self._count + return 0 + + @property + def broken(self): + """Return True if the barrier is in a broken state.""" + return self._state is _BarrierState.BROKEN diff --git a/Lib/test/test_asyncio/test_locks.py b/Lib/test/test_asyncio/test_locks.py index 920b3b5717a2c..415cbe570fde9 100644 --- a/Lib/test/test_asyncio/test_locks.py +++ b/Lib/test/test_asyncio/test_locks.py @@ -1,4 +1,4 @@ -"""Tests for lock.py""" +"""Tests for locks.py""" import unittest from unittest import mock @@ -9,7 +9,10 @@ STR_RGX_REPR = ( r'^<(?P<class>.*?) object at (?P<address>.*?)' r'\[(?P<extras>' - r'(set|unset|locked|unlocked)(, value:\d)?(, waiters:\d+)?' + r'(set|unset|locked|unlocked|filling|draining|resetting|broken)' + r'(, value:\d)?' + r'(, waiters:\d+)?' + r'(, waiters:\d+\/\d+)?' # barrier r')\]>\Z' ) RGX_REPR = re.compile(STR_RGX_REPR) @@ -943,5 +946,576 @@ async def coro(tag): ) +class BarrierTests(unittest.IsolatedAsyncioTestCase): + + async def asyncSetUp(self): + await super().asyncSetUp() + self.N = 5 + + def make_tasks(self, n, coro): + tasks = [asyncio.create_task(coro()) for _ in range(n)] + return tasks + + async def gather_tasks(self, n, coro): + tasks = self.make_tasks(n, coro) + res = await asyncio.gather(*tasks) + return res, tasks + + async def test_barrier(self): + barrier = asyncio.Barrier(self.N) + self.assertIn("filling", repr(barrier)) + with self.assertRaisesRegex( + TypeError, + "object Barrier can't be used in 'await' expression", + ): + await barrier + + self.assertIn("filling", repr(barrier)) + + async def test_repr(self): + barrier = asyncio.Barrier(self.N) + + self.assertTrue(RGX_REPR.match(repr(barrier))) + self.assertIn("filling", repr(barrier)) + + waiters = [] + async def wait(barrier): + await barrier.wait() + + incr = 2 + for i in range(incr): + waiters.append(asyncio.create_task(wait(barrier))) + await asyncio.sleep(0) + + self.assertTrue(RGX_REPR.match(repr(barrier))) + self.assertTrue(f"waiters:{incr}/{self.N}" in repr(barrier)) + self.assertIn("filling", repr(barrier)) + + # create missing waiters + for i in range(barrier.parties - barrier.n_waiting): + waiters.append(asyncio.create_task(wait(barrier))) + await asyncio.sleep(0) + + self.assertTrue(RGX_REPR.match(repr(barrier))) + self.assertIn("draining", repr(barrier)) + + # add a part of waiters + for i in range(incr): + waiters.append(asyncio.create_task(wait(barrier))) + await asyncio.sleep(0) + # and reset + await barrier.reset() + + self.assertTrue(RGX_REPR.match(repr(barrier))) + self.assertIn("resetting", repr(barrier)) + + # add a part of waiters again + for i in range(incr): + waiters.append(asyncio.create_task(wait(barrier))) + await asyncio.sleep(0) + # and abort + await barrier.abort() + + self.assertTrue(RGX_REPR.match(repr(barrier))) + self.assertIn("broken", repr(barrier)) + self.assertTrue(barrier.broken) + + # suppress unhandled exceptions + await asyncio.gather(*waiters, return_exceptions=True) + + async def test_barrier_parties(self): + self.assertRaises(ValueError, lambda: asyncio.Barrier(0)) + self.assertRaises(ValueError, lambda: asyncio.Barrier(-4)) + + self.assertIsInstance(asyncio.Barrier(self.N), asyncio.Barrier) + + async def test_context_manager(self): + self.N = 3 + barrier = asyncio.Barrier(self.N) + results = [] + + async def coro(): + async with barrier as i: + results.append(i) + + await self.gather_tasks(self.N, coro) + + self.assertListEqual(sorted(results), list(range(self.N))) + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_filling_one_task(self): + barrier = asyncio.Barrier(1) + + async def f(): + async with barrier as i: + return True + + ret = await f() + + self.assertTrue(ret) + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_filling_one_task_twice(self): + barrier = asyncio.Barrier(1) + + t1 = asyncio.create_task(barrier.wait()) + await asyncio.sleep(0) + self.assertEqual(barrier.n_waiting, 0) + + t2 = asyncio.create_task(barrier.wait()) + await asyncio.sleep(0) + + self.assertEqual(t1.result(), t2.result()) + self.assertEqual(t1.done(), t2.done()) + + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_filling_task_by_task(self): + self.N = 3 + barrier = asyncio.Barrier(self.N) + + t1 = asyncio.create_task(barrier.wait()) + await asyncio.sleep(0) + self.assertEqual(barrier.n_waiting, 1) + self.assertIn("filling", repr(barrier)) + + t2 = asyncio.create_task(barrier.wait()) + await asyncio.sleep(0) + self.assertEqual(barrier.n_waiting, 2) + self.assertIn("filling", repr(barrier)) + + t3 = asyncio.create_task(barrier.wait()) + await asyncio.sleep(0) + + await asyncio.wait([t1, t2, t3]) + + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_filling_tasks_wait_twice(self): + barrier = asyncio.Barrier(self.N) + results = [] + + async def coro(): + async with barrier: + results.append(True) + + async with barrier: + results.append(False) + + await self.gather_tasks(self.N, coro) + + self.assertEqual(len(results), self.N*2) + self.assertEqual(results.count(True), self.N) + self.assertEqual(results.count(False), self.N) + + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_filling_tasks_check_return_value(self): + barrier = asyncio.Barrier(self.N) + results1 = [] + results2 = [] + + async def coro(): + async with barrier: + results1.append(True) + + async with barrier as i: + results2.append(True) + return i + + res, _ = await self.gather_tasks(self.N, coro) + + self.assertEqual(len(results1), self.N) + self.assertTrue(all(results1)) + self.assertEqual(len(results2), self.N) + self.assertTrue(all(results2)) + self.assertListEqual(sorted(res), list(range(self.N))) + + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_draining_state(self): + barrier = asyncio.Barrier(self.N) + results = [] + + async def coro(): + async with barrier: + # barrier state change to filling for the last task release + results.append("draining" in repr(barrier)) + + await self.gather_tasks(self.N, coro) + + self.assertEqual(len(results), self.N) + self.assertEqual(results[-1], False) + self.assertTrue(all(results[:self.N-1])) + + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_blocking_tasks_while_draining(self): + rewait = 2 + barrier = asyncio.Barrier(self.N) + barrier_nowaiting = asyncio.Barrier(self.N - rewait) + results = [] + rewait_n = rewait + counter = 0 + + async def coro(): + nonlocal rewait_n + + # first time waiting + await barrier.wait() + + # after wainting once for all tasks + if rewait_n > 0: + rewait_n -= 1 + # wait again only for rewait tasks + await barrier.wait() + else: + # wait for end of draining state` + await barrier_nowaiting.wait() + # wait for other waiting tasks + await barrier.wait() + + # a success means that barrier_nowaiting + # was waited for exactly N-rewait=3 times + await self.gather_tasks(self.N, coro) + + async def test_filling_tasks_cancel_one(self): + self.N = 3 + barrier = asyncio.Barrier(self.N) + results = [] + + async def coro(): + await barrier.wait() + results.append(True) + + t1 = asyncio.create_task(coro()) + await asyncio.sleep(0) + self.assertEqual(barrier.n_waiting, 1) + + t2 = asyncio.create_task(coro()) + await asyncio.sleep(0) + self.assertEqual(barrier.n_waiting, 2) + + t1.cancel() + await asyncio.sleep(0) + self.assertEqual(barrier.n_waiting, 1) + with self.assertRaises(asyncio.CancelledError): + await t1 + self.assertTrue(t1.cancelled()) + + t3 = asyncio.create_task(coro()) + await asyncio.sleep(0) + self.assertEqual(barrier.n_waiting, 2) + + t4 = asyncio.create_task(coro()) + await asyncio.gather(t2, t3, t4) + + self.assertEqual(len(results), self.N) + self.assertTrue(all(results)) + + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_reset_barrier(self): + barrier = asyncio.Barrier(1) + + asyncio.create_task(barrier.reset()) + await asyncio.sleep(0) + + self.assertEqual(barrier.n_waiting, 0) + self.assertFalse(barrier.broken) + + async def test_reset_barrier_while_tasks_waiting(self): + barrier = asyncio.Barrier(self.N) + results = [] + + async def coro(): + try: + await barrier.wait() + except asyncio.BrokenBarrierError: + results.append(True) + + async def coro_reset(): + await barrier.reset() + + # N-1 tasks waiting on barrier with N parties + tasks = self.make_tasks(self.N-1, coro) + await asyncio.sleep(0) + + # reset the barrier + asyncio.create_task(coro_reset()) + await asyncio.gather(*tasks) + + self.assertEqual(len(results), self.N-1) + self.assertTrue(all(results)) + self.assertEqual(barrier.n_waiting, 0) + self.assertNotIn("resetting", repr(barrier)) + self.assertFalse(barrier.broken) + + async def test_reset_barrier_when_tasks_half_draining(self): + barrier = asyncio.Barrier(self.N) + results1 = [] + rest_of_tasks = self.N//2 + + async def coro(): + try: + await barrier.wait() + except asyncio.BrokenBarrierError: + # catch here waiting tasks + results1.append(True) + else: + # here drained task ouside the barrier + if rest_of_tasks == barrier._count: + # tasks outside the barrier + await barrier.reset() + + await self.gather_tasks(self.N, coro) + + self.assertEqual(results1, [True]*rest_of_tasks) + self.assertEqual(barrier.n_waiting, 0) + self.assertNotIn("resetting", repr(barrier)) + self.assertFalse(barrier.broken) + + async def test_reset_barrier_when_tasks_half_draining_half_blocking(self): + barrier = asyncio.Barrier(self.N) + results1 = [] + results2 = [] + blocking_tasks = self.N//2 + count = 0 + + async def coro(): + nonlocal count + try: + await barrier.wait() + except asyncio.BrokenBarrierError: + # here catch still waiting tasks + results1.append(True) + + # so now waiting again to reach nb_parties + await barrier.wait() + else: + count += 1 + if count > blocking_tasks: + # reset now: raise asyncio.BrokenBarrierError for waiting tasks + await barrier.reset() + + # so now waiting again to reach nb_parties + await barrier.wait() + else: + try: + await barrier.wait() + except asyncio.BrokenBarrierError: + # here no catch - blocked tasks go to wait + results2.append(True) + + await self.gather_tasks(self.N, coro) + + self.assertEqual(results1, [True]*blocking_tasks) + self.assertEqual(results2, []) + self.assertEqual(barrier.n_waiting, 0) + self.assertNotIn("resetting", repr(barrier)) + self.assertFalse(barrier.broken) + + async def test_reset_barrier_while_tasks_waiting_and_waiting_again(self): + barrier = asyncio.Barrier(self.N) + results1 = [] + results2 = [] + + async def coro1(): + try: + await barrier.wait() + except asyncio.BrokenBarrierError: + results1.append(True) + finally: + await barrier.wait() + results2.append(True) + + async def coro2(): + async with barrier: + results2.append(True) + + tasks = self.make_tasks(self.N-1, coro1) + + # reset barrier, N-1 waiting tasks raise an BrokenBarrierError + asyncio.create_task(barrier.reset()) + await asyncio.sleep(0) + + # complete waiting tasks in the `finally` + asyncio.create_task(coro2()) + + await asyncio.gather(*tasks) + + self.assertFalse(barrier.broken) + self.assertEqual(len(results1), self.N-1) + self.assertTrue(all(results1)) + self.assertEqual(len(results2), self.N) + self.assertTrue(all(results2)) + + self.assertEqual(barrier.n_waiting, 0) + + + async def test_reset_barrier_while_tasks_draining(self): + barrier = asyncio.Barrier(self.N) + results1 = [] + results2 = [] + results3 = [] + count = 0 + + async def coro(): + nonlocal count + + i = await barrier.wait() + count += 1 + if count == self.N: + # last task exited from barrier + await barrier.reset() + + # wit here to reach the `parties` + await barrier.wait() + else: + try: + # second waiting + await barrier.wait() + + # N-1 tasks here + results1.append(True) + except Exception as e: + # never goes here + results2.append(True) + + # Now, pass the barrier again + # last wait, must be completed + k = await barrier.wait() + results3.append(True) + + await self.gather_tasks(self.N, coro) + + self.assertFalse(barrier.broken) + self.assertTrue(all(results1)) + self.assertEqual(len(results1), self.N-1) + self.assertEqual(len(results2), 0) + self.assertEqual(len(results3), self.N) + self.assertTrue(all(results3)) + + self.assertEqual(barrier.n_waiting, 0) + + async def test_abort_barrier(self): + barrier = asyncio.Barrier(1) + + asyncio.create_task(barrier.abort()) + await asyncio.sleep(0) + + self.assertEqual(barrier.n_waiting, 0) + self.assertTrue(barrier.broken) + + async def test_abort_barrier_when_tasks_half_draining_half_blocking(self): + barrier = asyncio.Barrier(self.N) + results1 = [] + results2 = [] + blocking_tasks = self.N//2 + count = 0 + + async def coro(): + nonlocal count + try: + await barrier.wait() + except asyncio.BrokenBarrierError: + # here catch tasks waiting to drain + results1.append(True) + else: + count += 1 + if count > blocking_tasks: + # abort now: raise asyncio.BrokenBarrierError for all tasks + await barrier.abort() + else: + try: + await barrier.wait() + except asyncio.BrokenBarrierError: + # here catch blocked tasks (already drained) + results2.append(True) + + await self.gather_tasks(self.N, coro) + + self.assertTrue(barrier.broken) + self.assertEqual(results1, [True]*blocking_tasks) + self.assertEqual(results2, [True]*(self.N-blocking_tasks-1)) + self.assertEqual(barrier.n_waiting, 0) + self.assertNotIn("resetting", repr(barrier)) + + async def test_abort_barrier_when_exception(self): + # test from threading.Barrier: see `lock_tests.test_reset` + barrier = asyncio.Barrier(self.N) + results1 = [] + results2 = [] + + async def coro(): + try: + async with barrier as i : + if i == self.N//2: + raise RuntimeError + async with barrier: + results1.append(True) + except asyncio.BrokenBarrierError: + results2.append(True) + except RuntimeError: + await barrier.abort() + + await self.gather_tasks(self.N, coro) + + self.assertTrue(barrier.broken) + self.assertEqual(len(results1), 0) + self.assertEqual(len(results2), self.N-1) + self.assertTrue(all(results2)) + self.assertEqual(barrier.n_waiting, 0) + + async def test_abort_barrier_when_exception_then_resetting(self): + # test from threading.Barrier: see `lock_tests.test_abort_and_reset`` + barrier1 = asyncio.Barrier(self.N) + barrier2 = asyncio.Barrier(self.N) + results1 = [] + results2 = [] + results3 = [] + + async def coro(): + try: + i = await barrier1.wait() + if i == self.N//2: + raise RuntimeError + await barrier1.wait() + results1.append(True) + except asyncio.BrokenBarrierError: + results2.append(True) + except RuntimeError: + await barrier1.abort() + + # Synchronize and reset the barrier. Must synchronize first so + # that everyone has left it when we reset, and after so that no + # one enters it before the reset. + i = await barrier2.wait() + if i == self.N//2: + await barrier1.reset() + await barrier2.wait() + await barrier1.wait() + results3.append(True) + + await self.gather_tasks(self.N, coro) + + self.assertFalse(barrier1.broken) + self.assertEqual(len(results1), 0) + self.assertEqual(len(results2), self.N-1) + self.assertTrue(all(results2)) + self.assertEqual(len(results3), self.N) + self.assertTrue(all(results3)) + + self.assertEqual(barrier1.n_waiting, 0) + + if __name__ == '__main__': unittest.main() diff --git a/Misc/NEWS.d/next/Library/2021-03-31-15-22-45.bpo-43352.nSjMuE.rst b/Misc/NEWS.d/next/Library/2021-03-31-15-22-45.bpo-43352.nSjMuE.rst new file mode 100644 index 0000000000000..e53ba28b64099 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2021-03-31-15-22-45.bpo-43352.nSjMuE.rst @@ -0,0 +1 @@ +Add an Barrier object in synchronization primitives of *asyncio* Lib in order to be consistant with Barrier from *threading* and *multiprocessing* libs*

1 0

bpo-47118: Fix asyncio.Runner tests error (32117)
by asvetlov March 25, 2022

March 25, 2022

https://github.com/python/cpython/commit/20e6e5636a06fe5e1472062918d0a302d8… commit: 20e6e5636a06fe5e1472062918d0a302d82a71c3 branch: main author: Andrew Svetlov <andrew.svetlov(a)gmail.com> committer: asvetlov <andrew.svetlov(a)gmail.com> date: 2022-03-25T19:59:29+02:00 summary: bpo-47118: Fix asyncio.Runner tests error (32117) files: M Lib/test/test_asyncio/test_runners.py diff --git a/Lib/test/test_asyncio/test_runners.py b/Lib/test/test_asyncio/test_runners.py index 2919412ab81db..c0bd1a242c86c 100644 --- a/Lib/test/test_asyncio/test_runners.py +++ b/Lib/test/test_asyncio/test_runners.py @@ -296,7 +296,7 @@ async def get_context(): self.assertEqual(-1, runner.run(f(1))) self.assertEqual(1, runner.run(f(2))) - self.assertEqual({cvar: 2}, dict(runner.run(get_context()))) + self.assertEqual(2, runner.run(get_context()).get(cvar)) def test_recursine_run(self): async def g():

1 0

Fix some typos in whatsnew (GH-32098)
by Mariatta March 25, 2022

March 25, 2022

https://github.com/python/cpython/commit/c07ca1aab6e1928e9eefe9dfec7e7e5ae9… commit: c07ca1aab6e1928e9eefe9dfec7e7e5ae982b420 branch: main author: Kurt McKee <contactme(a)kurtmckee.org> committer: Mariatta <Mariatta(a)users.noreply.github.com> date: 2022-03-25T10:32:05-07:00 summary: Fix some typos in whatsnew (GH-32098) * Fix some typos and phrasing * Convert the `hash()` text to a link * Remove definite article files: M Doc/whatsnew/3.11.rst diff --git a/Doc/whatsnew/3.11.rst b/Doc/whatsnew/3.11.rst index 8c120ec45fe3f..2c09d3583ea15 100644 --- a/Doc/whatsnew/3.11.rst +++ b/Doc/whatsnew/3.11.rst @@ -193,9 +193,9 @@ Other CPython Implementation Changes support :class:`typing.SupportsComplex` and :class:`typing.SupportsBytes` protocols. (Contributed by Mark Dickinson and Dong-hee Na in :issue:`24234`.) -* ``siphash13`` is added as a new internal hashing algorithms. It's has similar security +* ``siphash13`` is added as a new internal hashing algorithms. It has similar security properties as ``siphash24`` but it is slightly faster for long inputs. ``str``, ``bytes``, - and some other types now use it as default algorithm for ``hash()``. :pep:`552` + and some other types now use it as default algorithm for :func:`hash`. :pep:`552` hash-based pyc files now use ``siphash13``, too. (Contributed by Inada Naoki in :issue:`29410`.) @@ -210,9 +210,9 @@ Other CPython Implementation Changes have been removed as their values can be derived from ``exc_value``. (Contributed by Irit Katriel in :issue:`45711`.) -* A new command line option for the Windows installer ``AppendPath`` has beend added. +* A new command line option for the Windows installer ``AppendPath`` has been added. It behaves similiar to ``PrependPath`` but appends the install and scripts directories - instead of prepending it. + instead of prepending them. (Contributed by Bastian Neuburger in :issue:`44934`.)

1 0

bpo-47053: Reduce deoptimization in BINARY_OP_INPLACE_ADD_UNICODE (GH-31318)
by markshannon March 25, 2022

March 25, 2022

https://github.com/python/cpython/commit/cca43b7d64f47ea921d0f7a347ae1a839c… commit: cca43b7d64f47ea921d0f7a347ae1a839c5463c3 branch: main author: Dennis Sweeney <36520290+sweeneyde(a)users.noreply.github.com> committer: markshannon <mark(a)hotpy.org> date: 2022-03-25T16:13:19Z summary: bpo-47053: Reduce deoptimization in BINARY_OP_INPLACE_ADD_UNICODE (GH-31318) * Don't deopt if refcounts are too big * Detect more at specialization time files: A Misc/NEWS.d/next/Core and Builtins/2022-03-17-22-47-29.bpo-47053.QAXk8Q.rst M Include/internal/pycore_code.h M Python/ceval.c M Python/specialize.c diff --git a/Include/internal/pycore_code.h b/Include/internal/pycore_code.h index 82dc9e4bdc62f..0f6613b6c1e3d 100644 --- a/Include/internal/pycore_code.h +++ b/Include/internal/pycore_code.h @@ -263,7 +263,7 @@ extern int _Py_Specialize_Call(PyObject *callable, _Py_CODEUNIT *instr, extern int _Py_Specialize_Precall(PyObject *callable, _Py_CODEUNIT *instr, int nargs, PyObject *kwnames, int oparg); extern void _Py_Specialize_BinaryOp(PyObject *lhs, PyObject *rhs, _Py_CODEUNIT *instr, - int oparg); + int oparg, PyObject **locals); extern void _Py_Specialize_CompareOp(PyObject *lhs, PyObject *rhs, _Py_CODEUNIT *instr, int oparg); extern void _Py_Specialize_UnpackSequence(PyObject *seq, _Py_CODEUNIT *instr, diff --git a/Misc/NEWS.d/next/Core and Builtins/2022-03-17-22-47-29.bpo-47053.QAXk8Q.rst b/Misc/NEWS.d/next/Core and Builtins/2022-03-17-22-47-29.bpo-47053.QAXk8Q.rst new file mode 100644 index 0000000000000..097105b1c20f4 --- /dev/null +++ b/Misc/NEWS.d/next/Core and Builtins/2022-03-17-22-47-29.bpo-47053.QAXk8Q.rst @@ -0,0 +1 @@ +Reduce de-optimization in the specialized ``BINARY_OP_INPLACE_ADD_UNICODE`` opcode. diff --git a/Python/ceval.c b/Python/ceval.c index e1d961fee6a83..4824b192f4e48 100644 --- a/Python/ceval.c +++ b/Python/ceval.c @@ -2002,10 +2002,10 @@ _PyEval_EvalFrameDefault(PyThreadState *tstate, _PyInterpreterFrame *frame, int PyObject *right = TOP(); DEOPT_IF(!PyUnicode_CheckExact(left), BINARY_OP); DEOPT_IF(Py_TYPE(right) != Py_TYPE(left), BINARY_OP); - DEOPT_IF(Py_REFCNT(left) != 2, BINARY_OP); _Py_CODEUNIT true_next = next_instr[INLINE_CACHE_ENTRIES_BINARY_OP]; int next_oparg = _Py_OPARG(true_next); - assert(_Py_OPCODE(true_next) == STORE_FAST); + assert(_Py_OPCODE(true_next) == STORE_FAST || + _Py_OPCODE(true_next) == STORE_FAST__LOAD_FAST); /* In the common case, there are 2 references to the value * stored in 'variable' when the v = v + ... is performed: one * on the value stack (in 'v') and one still stored in the @@ -2016,7 +2016,8 @@ _PyEval_EvalFrameDefault(PyThreadState *tstate, _PyInterpreterFrame *frame, int DEOPT_IF(var != left, BINARY_OP); STAT_INC(BINARY_OP, hit); GETLOCAL(next_oparg) = NULL; - Py_DECREF(left); + assert(Py_REFCNT(left) >= 2); + Py_DECREF(left); // XXX never need to dealloc STACK_SHRINK(1); PyUnicode_Append(&TOP(), right); Py_DECREF(right); @@ -5378,7 +5379,7 @@ _PyEval_EvalFrameDefault(PyThreadState *tstate, _PyInterpreterFrame *frame, int PyObject *lhs = SECOND(); PyObject *rhs = TOP(); next_instr--; - _Py_Specialize_BinaryOp(lhs, rhs, next_instr, oparg); + _Py_Specialize_BinaryOp(lhs, rhs, next_instr, oparg, &GETLOCAL(0)); DISPATCH(); } else { diff --git a/Python/specialize.c b/Python/specialize.c index ce091a2c5f078..720bc7f241586 100644 --- a/Python/specialize.c +++ b/Python/specialize.c @@ -1742,7 +1742,7 @@ binary_op_fail_kind(int oparg, PyObject *lhs, PyObject *rhs) void _Py_Specialize_BinaryOp(PyObject *lhs, PyObject *rhs, _Py_CODEUNIT *instr, - int oparg) + int oparg, PyObject **locals) { assert(_PyOpcode_Caches[BINARY_OP] == INLINE_CACHE_ENTRIES_BINARY_OP); _PyBinaryOpCache *cache = (_PyBinaryOpCache *)(instr + 1); @@ -1754,7 +1754,9 @@ _Py_Specialize_BinaryOp(PyObject *lhs, PyObject *rhs, _Py_CODEUNIT *instr, } if (PyUnicode_CheckExact(lhs)) { _Py_CODEUNIT next = instr[INLINE_CACHE_ENTRIES_BINARY_OP + 1]; - if (_Py_OPCODE(next) == STORE_FAST && Py_REFCNT(lhs) == 2) { + bool to_store = (_Py_OPCODE(next) == STORE_FAST || + _Py_OPCODE(next) == STORE_FAST__LOAD_FAST); + if (to_store && locals[_Py_OPARG(next)] == lhs) { _Py_SET_OPCODE(*instr, BINARY_OP_INPLACE_ADD_UNICODE); goto success; }

1 0

bpo-42197: Don't create `f_locals` dictionary unless we actually need it. (GH-32055)
by markshannon March 25, 2022

March 25, 2022

https://github.com/python/cpython/commit/d7163bb35d1ed46bde9affcd4eb267dfd0… commit: d7163bb35d1ed46bde9affcd4eb267dfd0b703dd branch: main author: Mark Shannon <mark(a)hotpy.org> committer: markshannon <mark(a)hotpy.org> date: 2022-03-25T12:57:50Z summary: bpo-42197: Don't create `f_locals` dictionary unless we actually need it. (GH-32055) * `PyFrame_FastToLocalsWithError` and `PyFrame_LocalsToFast` are no longer called during profile and tracing. (Contributed by Fabio Zadrozny) * Make accesses to a frame's `f_locals` safe from C code, not relying on calls to `PyFrame_FastToLocals` or `PyFrame_LocalsToFast`. * Document new `PyFrame_GetLocals` C-API function. files: A Misc/NEWS.d/next/Core and Builtins/2022-03-22-15-12-28.bpo-42197.SwrrFO.rst M Doc/c-api/frame.rst M Doc/whatsnew/3.11.rst M Include/cpython/frameobject.h M Include/internal/pycore_frame.h M Objects/frameobject.c M Python/sysmodule.c diff --git a/Doc/c-api/frame.rst b/Doc/c-api/frame.rst index 0e36e6e1fd706..0c11bc163b417 100644 --- a/Doc/c-api/frame.rst +++ b/Doc/c-api/frame.rst @@ -41,6 +41,17 @@ See also :ref:`Reflection <reflection>`. .. versionadded:: 3.9 +.. c:function:: PyObject* PyFrame_GetLocals(PyFrameObject *frame) + + Get the *frame*'s ``f_locals`` attribute (:class:`dict`). + + Return a :term:`strong reference`. + + *frame* must not be ``NULL``. + + .. versionadded:: 3.11 + + .. c:function:: int PyFrame_GetLineNumber(PyFrameObject *frame) Return the line number that *frame* is currently executing. diff --git a/Doc/whatsnew/3.11.rst b/Doc/whatsnew/3.11.rst index b2fdb4852360d..8c120ec45fe3f 100644 --- a/Doc/whatsnew/3.11.rst +++ b/Doc/whatsnew/3.11.rst @@ -969,7 +969,7 @@ Porting to Python 3.11 Code using ``f_lasti`` with ``PyCode_Addr2Line()`` must use :c:func:`PyFrame_GetLineNumber` instead. * ``f_lineno``: use :c:func:`PyFrame_GetLineNumber` - * ``f_locals``: use ``PyObject_GetAttrString((PyObject*)frame, "f_locals")``. + * ``f_locals``: use :c:func:`PyFrame_GetLocals`. * ``f_stackdepth``: removed. * ``f_state``: no public API (renamed to ``f_frame.f_state``). * ``f_trace``: no public API. @@ -983,6 +983,12 @@ Porting to Python 3.11 computed lazily. The :c:func:`PyFrame_GetBack` function must be called instead. + Debuggers that accessed the ``f_locals`` directly *must* call + `:c:func:`PyFrame_GetLocals` instead. They no longer need to call + `:c:func:`PyFrame_FastToLocalsWithError` or :c:func:`PyFrame_LocalsToFast`, + in fact they should not call those functions. The necessary updating of the + frame is now managed by the virtual machine. + Code defining ``PyFrame_GetCode()`` on Python 3.8 and older:: #if PY_VERSION_HEX < 0x030900B1 diff --git a/Include/cpython/frameobject.h b/Include/cpython/frameobject.h index 9b697fb3cbaf5..d54d3652a0dbc 100644 --- a/Include/cpython/frameobject.h +++ b/Include/cpython/frameobject.h @@ -23,3 +23,4 @@ PyAPI_FUNC(int) PyFrame_FastToLocalsWithError(PyFrameObject *f); PyAPI_FUNC(void) PyFrame_FastToLocals(PyFrameObject *); PyAPI_FUNC(PyFrameObject *) PyFrame_GetBack(PyFrameObject *frame); +PyAPI_FUNC(PyObject *) PyFrame_GetLocals(PyFrameObject *frame); diff --git a/Include/internal/pycore_frame.h b/Include/internal/pycore_frame.h index 14fba8cd1f941..211831a6e497f 100644 --- a/Include/internal/pycore_frame.h +++ b/Include/internal/pycore_frame.h @@ -15,6 +15,7 @@ struct _frame { int f_lineno; /* Current line number. Only valid if non-zero */ char f_trace_lines; /* Emit per-line trace events? */ char f_trace_opcodes; /* Emit per-opcode trace events? */ + char f_fast_as_locals; /* Have the fast locals of this frame been converted to a dict? */ /* The frame data, if this frame object owns the frame */ PyObject *_f_frame_data[1]; }; diff --git a/Misc/NEWS.d/next/Core and Builtins/2022-03-22-15-12-28.bpo-42197.SwrrFO.rst b/Misc/NEWS.d/next/Core and Builtins/2022-03-22-15-12-28.bpo-42197.SwrrFO.rst new file mode 100644 index 0000000000000..d54002a80e44f --- /dev/null +++ b/Misc/NEWS.d/next/Core and Builtins/2022-03-22-15-12-28.bpo-42197.SwrrFO.rst @@ -0,0 +1,2 @@ +:c:func:`PyFrame_FastToLocalsWithError` and :c:func:`PyFrame_LocalsToFast` are no longer +called during profiling nor tracing. C code can access the ``f_locals`` attribute of :c:type:`PyFrameObject` by calling :c:func:`PyFrame_GetLocals`. diff --git a/Objects/frameobject.c b/Objects/frameobject.c index 5c6a8bcb9008d..13dfbf6b9db41 100644 --- a/Objects/frameobject.c +++ b/Objects/frameobject.c @@ -840,6 +840,7 @@ _PyFrame_New_NoTrack(PyCodeObject *code) f->f_trace = NULL; f->f_trace_lines = 1; f->f_trace_opcodes = 0; + f->f_fast_as_locals = 0; f->f_lineno = 0; return f; } @@ -1004,7 +1005,11 @@ PyFrame_FastToLocalsWithError(PyFrameObject *f) PyErr_BadInternalCall(); return -1; } - return _PyFrame_FastToLocalsWithError(f->f_frame); + int err = _PyFrame_FastToLocalsWithError(f->f_frame); + if (err == 0) { + f->f_fast_as_locals = 1; + } + return err; } void @@ -1028,8 +1033,9 @@ _PyFrame_LocalsToFast(_PyInterpreterFrame *frame, int clear) PyObject *error_type, *error_value, *error_traceback; PyCodeObject *co; locals = frame->f_locals; - if (locals == NULL) + if (locals == NULL) { return; + } fast = _PyFrame_GetLocalsArray(frame); co = frame->f_code; @@ -1088,13 +1094,12 @@ _PyFrame_LocalsToFast(_PyInterpreterFrame *frame, int clear) void PyFrame_LocalsToFast(PyFrameObject *f, int clear) { - if (f == NULL || _PyFrame_GetState(f) == FRAME_CLEARED) { - return; + if (f && f->f_fast_as_locals && _PyFrame_GetState(f) != FRAME_CLEARED) { + _PyFrame_LocalsToFast(f->f_frame, clear); + f->f_fast_as_locals = 0; } - _PyFrame_LocalsToFast(f->f_frame, clear); } - PyCodeObject * PyFrame_GetCode(PyFrameObject *frame) { @@ -1118,6 +1123,12 @@ PyFrame_GetBack(PyFrameObject *frame) return back; } +PyObject* +PyFrame_GetLocals(PyFrameObject *frame) +{ + return frame_getlocals(frame, NULL); +} + PyObject* _PyEval_BuiltinsFromGlobals(PyThreadState *tstate, PyObject *globals) { diff --git a/Python/sysmodule.c b/Python/sysmodule.c index c89f81f689f7e..6322af5f5ca81 100644 --- a/Python/sysmodule.c +++ b/Python/sysmodule.c @@ -924,15 +924,19 @@ static PyObject * call_trampoline(PyThreadState *tstate, PyObject* callback, PyFrameObject *frame, int what, PyObject *arg) { - if (PyFrame_FastToLocalsWithError(frame) < 0) { - return NULL; - } PyObject *stack[3]; stack[0] = (PyObject *)frame; stack[1] = whatstrings[what]; stack[2] = (arg != NULL) ? arg : Py_None; + /* Discard any previous modifications the frame's fast locals */ + if (frame->f_fast_as_locals) { + if (PyFrame_FastToLocalsWithError(frame) < 0) { + return NULL; + } + } + /* call the Python-level function */ PyObject *result = _PyObject_FastCallTstate(tstate, callback, stack, 3);

1 0

Add an issue template config for the migration (GH-32101)
by ezio-melotti March 25, 2022

March 25, 2022

https://github.com/python/cpython/commit/b68431fadb3150134ac6ccbf501cdfeaf4… commit: b68431fadb3150134ac6ccbf501cdfeaf4c75678 branch: main author: Ezio Melotti <ezio.melotti(a)gmail.com> committer: ezio-melotti <ezio.melotti(a)gmail.com> date: 2022-03-25T00:45:50+01:00 summary: Add an issue template config for the migration (GH-32101) files: A .github/ISSUE_TEMPLATE/config.yml diff --git a/.github/ISSUE_TEMPLATE/config.yml b/.github/ISSUE_TEMPLATE/config.yml new file mode 100644 index 0000000000000..6a222c618aa37 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/config.yml @@ -0,0 +1,5 @@ +blank_issues_enabled: false +contact_links: + - name: ⚠ GitHub Issues Migration in progress ⚠ + url: https://discuss.python.org/t/github-issues-migration-status-update/14573 + about: Check status updates on the migration

1 0

bpo-40280: Add wasm32-emscripten and wasm32-wasi SOABI (GH-32095)
by miss-islington March 24, 2022

March 24, 2022

https://github.com/python/cpython/commit/8a0a9e5b1928fab7d9819c8d6498ef5c0b… commit: 8a0a9e5b1928fab7d9819c8d6498ef5c0b9383af branch: main author: Christian Heimes <christian(a)python.org> committer: miss-islington <31488909+miss-islington(a)users.noreply.github.com> date: 2022-03-24T14:09:42-07:00 summary: bpo-40280: Add wasm32-emscripten and wasm32-wasi SOABI (GH-32095) Shared extension on Emscripten now have suffix ``.cpython-311-wasm32-emscripten.so`` (JS loader) and ``.cpython-311-wasm32-emscripten.wasm`` (WebAssembly code). files: A Misc/NEWS.d/next/Build/2022-03-24-12-12-35.bpo-40280.eAQWrM.rst M configure M configure.ac diff --git a/Misc/NEWS.d/next/Build/2022-03-24-12-12-35.bpo-40280.eAQWrM.rst b/Misc/NEWS.d/next/Build/2022-03-24-12-12-35.bpo-40280.eAQWrM.rst new file mode 100644 index 0000000000000..7c7dc6d24fae4 --- /dev/null +++ b/Misc/NEWS.d/next/Build/2022-03-24-12-12-35.bpo-40280.eAQWrM.rst @@ -0,0 +1,2 @@ +Add SOABI ``wasm32-emscripten`` for Emscripten and ``wasm32-wasi`` for WASI +on 32bit WASM as well as ``wasm64`` counter parts. diff --git a/configure b/configure index 128e36a80062a..00f5c7ba4cc72 100755 --- a/configure +++ b/configure @@ -6084,6 +6084,22 @@ cat > conftest.c <<EOF darwin #elif defined(__VXWORKS__) vxworks +#elif defined(__wasm32__) +# if defined(__EMSCRIPTEN__) + wasm32-emscripten +# elif defined(__wasi__) + wasm32-wasi +# else +# error unknown wasm32 platform +# endif +#elif defined(__wasm64__) +# if defined(__EMSCRIPTEN) + wasm64-emscripten +# elif defined(__wasi__) + wasm64-wasi +# else +# error unknown wasm64 platform +# endif #else # error unknown platform triplet #endif diff --git a/configure.ac b/configure.ac index 6fd95f7704ace..094866136dda7 100644 --- a/configure.ac +++ b/configure.ac @@ -973,6 +973,22 @@ cat > conftest.c <<EOF darwin #elif defined(__VXWORKS__) vxworks +#elif defined(__wasm32__) +# if defined(__EMSCRIPTEN__) + wasm32-emscripten +# elif defined(__wasi__) + wasm32-wasi +# else +# error unknown wasm32 platform +# endif +#elif defined(__wasm64__) +# if defined(__EMSCRIPTEN) + wasm64-emscripten +# elif defined(__wasi__) + wasm64-wasi +# else +# error unknown wasm64 platform +# endif #else # error unknown platform triplet #endif

1 0

bpo-47062: Implement asyncio.Runner context manager (GH-31799)
by asvetlov March 24, 2022

March 24, 2022

https://github.com/python/cpython/commit/4119d2d7c9e25acd4f16994fb92d656f8b… commit: 4119d2d7c9e25acd4f16994fb92d656f8b7816d7 branch: main author: Andrew Svetlov <andrew.svetlov(a)gmail.com> committer: asvetlov <andrew.svetlov(a)gmail.com> date: 2022-03-24T21:51:16+02:00 summary: bpo-47062: Implement asyncio.Runner context manager (GH-31799) Co-authored-by: Zachary Ware <zach(a)python.org> files: A Doc/library/asyncio-runner.rst A Misc/NEWS.d/next/Library/2022-03-18-22-46-18.bpo-47062.RNc99_.rst M Doc/library/asyncio-task.rst M Doc/library/asyncio.rst M Lib/asyncio/runners.py M Lib/test/test_asyncio/test_runners.py M Lib/unittest/async_case.py diff --git a/Doc/library/asyncio-runner.rst b/Doc/library/asyncio-runner.rst new file mode 100644 index 0000000000000..2f4de9edaa400 --- /dev/null +++ b/Doc/library/asyncio-runner.rst @@ -0,0 +1,121 @@ +.. currentmodule:: asyncio + + +======= +Runners +======= + +**Source code:** :source:`Lib/asyncio/runners.py` + + +This section outlines high-level asyncio primitives to run asyncio code. + +They are built on top of an :ref:`event loop <asyncio-event-loop>` with the aim +to simplify async code usage for common wide-spread scenarios. + +.. contents:: + :depth: 1 + :local: + + + +Running an asyncio Program +========================== + +.. function:: run(coro, *, debug=None) + + Execute the :term:`coroutine` *coro* and return the result. + + This function runs the passed coroutine, taking care of + managing the asyncio event loop, *finalizing asynchronous + generators*, and closing the threadpool. + + This function cannot be called when another asyncio event loop is + running in the same thread. + + If *debug* is ``True``, the event loop will be run in debug mode. ``False`` disables + debug mode explicitly. ``None`` is used to respect the global + :ref:`asyncio-debug-mode` settings. + + This function always creates a new event loop and closes it at + the end. It should be used as a main entry point for asyncio + programs, and should ideally only be called once. + + Example:: + + async def main(): + await asyncio.sleep(1) + print('hello') + + asyncio.run(main()) + + .. versionadded:: 3.7 + + .. versionchanged:: 3.9 + Updated to use :meth:`loop.shutdown_default_executor`. + + .. versionchanged:: 3.10 + + *debug* is ``None`` by default to respect the global debug mode settings. + + +Runner context manager +====================== + +.. class:: Runner(*, debug=None, factory=None) + + A context manager that simplifies *multiple* async function calls in the same + context. + + Sometimes several top-level async functions should be called in the same :ref:`event + loop <asyncio-event-loop>` and :class:`contextvars.Context`. + + If *debug* is ``True``, the event loop will be run in debug mode. ``False`` disables + debug mode explicitly. ``None`` is used to respect the global + :ref:`asyncio-debug-mode` settings. + + *factory* could be used for overriding the loop creation. + :func:`asyncio.new_event_loop` is used if ``None``. + + Basically, :func:`asyncio.run()` example can be rewritten with the runner usage:: + + async def main(): + await asyncio.sleep(1) + print('hello') + + with asyncio.Runner() as runner: + runner.run(main()) + + .. versionadded:: 3.11 + + .. method:: run(coro, *, context=None) + + Run a :term:`coroutine <coroutine>` *coro* in the embedded loop. + + Return the coroutine's result or raise its exception. + + An optional keyword-only *context* argument allows specifying a + custom :class:`contextvars.Context` for the *coro* to run in. + The runner's default context is used if ``None``. + + This function cannot be called when another asyncio event loop is + running in the same thread. + + .. method:: close() + + Close the runner. + + Finalize asynchronous generators, shutdown default executor, close the event loop + and release embedded :class:`contextvars.Context`. + + .. method:: get_loop() + + Return the event loop associated with the runner instance. + + .. note:: + + :class:`Runner` uses the lazy initialization strategy, its constructor doesn't + initialize underlying low-level structures. + + Embedded *loop* and *context* are created at the :keyword:`with` body entering + or the first call of :meth:`run` or :meth:`get_loop`. diff --git a/Doc/library/asyncio-task.rst b/Doc/library/asyncio-task.rst index 21a4cb5820bd1..c104ac5b9a3b8 100644 --- a/Doc/library/asyncio-task.rst +++ b/Doc/library/asyncio-task.rst @@ -204,43 +204,6 @@ A good example of a low-level function that returns a Future object is :meth:`loop.run_in_executor`. -Running an asyncio Program -========================== - -.. function:: run(coro, *, debug=False) - - Execute the :term:`coroutine` *coro* and return the result. - - This function runs the passed coroutine, taking care of - managing the asyncio event loop, *finalizing asynchronous - generators*, and closing the threadpool. - - This function cannot be called when another asyncio event loop is - running in the same thread. - - If *debug* is ``True``, the event loop will be run in debug mode. - - This function always creates a new event loop and closes it at - the end. It should be used as a main entry point for asyncio - programs, and should ideally only be called once. - - Example:: - - async def main(): - await asyncio.sleep(1) - print('hello') - - asyncio.run(main()) - - .. versionadded:: 3.7 - - .. versionchanged:: 3.9 - Updated to use :meth:`loop.shutdown_default_executor`. - - .. note:: - The source code for ``asyncio.run()`` can be found in - :source:`Lib/asyncio/runners.py`. - Creating Tasks ============== diff --git a/Doc/library/asyncio.rst b/Doc/library/asyncio.rst index 94a853259d348..8b3a060ffad52 100644 --- a/Doc/library/asyncio.rst +++ b/Doc/library/asyncio.rst @@ -67,6 +67,7 @@ Additionally, there are **low-level** APIs for :caption: High-level APIs :maxdepth: 1 + asyncio-runner.rst asyncio-task.rst asyncio-stream.rst asyncio-sync.rst diff --git a/Lib/asyncio/runners.py b/Lib/asyncio/runners.py index 9a5e9a48479ef..975509c7d645d 100644 --- a/Lib/asyncio/runners.py +++ b/Lib/asyncio/runners.py @@ -1,10 +1,112 @@ -__all__ = 'run', +__all__ = ('Runner', 'run') +import contextvars +import enum from . import coroutines from . import events from . import tasks +class _State(enum.Enum): + CREATED = "created" + INITIALIZED = "initialized" + CLOSED = "closed" + + +class Runner: + """A context manager that controls event loop life cycle. + + The context manager always creates a new event loop, + allows to run async functions inside it, + and properly finalizes the loop at the context manager exit. + + If debug is True, the event loop will be run in debug mode. + If factory is passed, it is used for new event loop creation. + + asyncio.run(main(), debug=True) + + is a shortcut for + + with asyncio.Runner(debug=True) as runner: + runner.run(main()) + + The run() method can be called multiple times within the runner's context. + + This can be useful for interactive console (e.g. IPython), + unittest runners, console tools, -- everywhere when async code + is called from existing sync framework and where the preferred single + asyncio.run() call doesn't work. + + """ + + # Note: the class is final, it is not intended for inheritance. + + def __init__(self, *, debug=None, factory=None): + self._state = _State.CREATED + self._debug = debug + self._factory = factory + self._loop = None + self._context = None + + def __enter__(self): + self._lazy_init() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + def close(self): + """Shutdown and close event loop.""" + if self._state is not _State.INITIALIZED: + return + try: + loop = self._loop + _cancel_all_tasks(loop) + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.run_until_complete(loop.shutdown_default_executor()) + finally: + loop.close() + self._loop = None + self._state = _State.CLOSED + + def get_loop(self): + """Return embedded event loop.""" + self._lazy_init() + return self._loop + + def run(self, coro, *, context=None): + """Run a coroutine inside the embedded event loop.""" + if not coroutines.iscoroutine(coro): + raise ValueError("a coroutine was expected, got {!r}".format(coro)) + + if events._get_running_loop() is not None: + # fail fast with short traceback + raise RuntimeError( + "Runner.run() cannot be called from a running event loop") + + self._lazy_init() + + if context is None: + context = self._context + task = self._loop.create_task(coro, context=context) + return self._loop.run_until_complete(task) + + def _lazy_init(self): + if self._state is _State.CLOSED: + raise RuntimeError("Runner is closed") + if self._state is _State.INITIALIZED: + return + if self._factory is None: + self._loop = events.new_event_loop() + else: + self._loop = self._factory() + if self._debug is not None: + self._loop.set_debug(self._debug) + self._context = contextvars.copy_context() + self._state = _State.INITIALIZED + + + def run(main, *, debug=None): """Execute the coroutine and return the result. @@ -30,26 +132,12 @@ async def main(): asyncio.run(main()) """ if events._get_running_loop() is not None: + # fail fast with short traceback raise RuntimeError( "asyncio.run() cannot be called from a running event loop") - if not coroutines.iscoroutine(main): - raise ValueError("a coroutine was expected, got {!r}".format(main)) - - loop = events.new_event_loop() - try: - events.set_event_loop(loop) - if debug is not None: - loop.set_debug(debug) - return loop.run_until_complete(main) - finally: - try: - _cancel_all_tasks(loop) - loop.run_until_complete(loop.shutdown_asyncgens()) - loop.run_until_complete(loop.shutdown_default_executor()) - finally: - events.set_event_loop(None) - loop.close() + with Runner(debug=debug) as runner: + return runner.run(main) def _cancel_all_tasks(loop): diff --git a/Lib/test/test_asyncio/test_runners.py b/Lib/test/test_asyncio/test_runners.py index 112273662b20b..2919412ab81db 100644 --- a/Lib/test/test_asyncio/test_runners.py +++ b/Lib/test/test_asyncio/test_runners.py @@ -1,4 +1,7 @@ import asyncio +import contextvars +import gc +import re import unittest from unittest import mock @@ -186,5 +189,135 @@ async def main(): self.assertFalse(spinner.ag_running) +class RunnerTests(BaseTest): + + def test_non_debug(self): + with asyncio.Runner(debug=False) as runner: + self.assertFalse(runner.get_loop().get_debug()) + + def test_debug(self): + with asyncio.Runner(debug=True) as runner: + self.assertTrue(runner.get_loop().get_debug()) + + def test_custom_factory(self): + loop = mock.Mock() + with asyncio.Runner(factory=lambda: loop) as runner: + self.assertIs(runner.get_loop(), loop) + + def test_run(self): + async def f(): + await asyncio.sleep(0) + return 'done' + + with asyncio.Runner() as runner: + self.assertEqual('done', runner.run(f())) + loop = runner.get_loop() + + with self.assertRaisesRegex( + RuntimeError, + "Runner is closed" + ): + runner.get_loop() + + self.assertTrue(loop.is_closed()) + + def test_run_non_coro(self): + with asyncio.Runner() as runner: + with self.assertRaisesRegex( + ValueError, + "a coroutine was expected" + ): + runner.run(123) + + def test_run_future(self): + with asyncio.Runner() as runner: + with self.assertRaisesRegex( + ValueError, + "a coroutine was expected" + ): + fut = runner.get_loop().create_future() + runner.run(fut) + + def test_explicit_close(self): + runner = asyncio.Runner() + loop = runner.get_loop() + runner.close() + with self.assertRaisesRegex( + RuntimeError, + "Runner is closed" + ): + runner.get_loop() + + self.assertTrue(loop.is_closed()) + + def test_double_close(self): + runner = asyncio.Runner() + loop = runner.get_loop() + + runner.close() + self.assertTrue(loop.is_closed()) + + # the second call is no-op + runner.close() + self.assertTrue(loop.is_closed()) + + def test_second_with_block_raises(self): + ret = [] + + async def f(arg): + ret.append(arg) + + runner = asyncio.Runner() + with runner: + runner.run(f(1)) + + with self.assertRaisesRegex( + RuntimeError, + "Runner is closed" + ): + with runner: + runner.run(f(2)) + + self.assertEqual([1], ret) + + def test_run_keeps_context(self): + cvar = contextvars.ContextVar("cvar", default=-1) + + async def f(val): + old = cvar.get() + await asyncio.sleep(0) + cvar.set(val) + return old + + async def get_context(): + return contextvars.copy_context() + + with asyncio.Runner() as runner: + self.assertEqual(-1, runner.run(f(1))) + self.assertEqual(1, runner.run(f(2))) + + self.assertEqual({cvar: 2}, dict(runner.run(get_context()))) + + def test_recursine_run(self): + async def g(): + pass + + async def f(): + runner.run(g()) + + with asyncio.Runner() as runner: + with self.assertWarnsRegex( + RuntimeWarning, + "coroutine .+ was never awaited", + ): + with self.assertRaisesRegex( + RuntimeError, + re.escape( + "Runner.run() cannot be called from a running event loop" + ), + ): + runner.run(f()) + + if __name__ == '__main__': unittest.main() diff --git a/Lib/unittest/async_case.py b/Lib/unittest/async_case.py index 25adc3deff63d..85b938fb293af 100644 --- a/Lib/unittest/async_case.py +++ b/Lib/unittest/async_case.py @@ -34,7 +34,7 @@ class IsolatedAsyncioTestCase(TestCase): def __init__(self, methodName='runTest'): super().__init__(methodName) - self._asyncioTestLoop = None + self._asyncioRunner = None self._asyncioTestContext = contextvars.copy_context() async def asyncSetUp(self): @@ -75,76 +75,44 @@ def _callCleanup(self, function, *args, **kwargs): self._callMaybeAsync(function, *args, **kwargs) def _callAsync(self, func, /, *args, **kwargs): - assert self._asyncioTestLoop is not None, 'asyncio test loop is not initialized' + assert self._asyncioRunner is not None, 'asyncio runner is not initialized' assert inspect.iscoroutinefunction(func), f'{func!r} is not an async function' - task = self._asyncioTestLoop.create_task( + return self._asyncioRunner.run( func(*args, **kwargs), - context=self._asyncioTestContext, + context=self._asyncioTestContext ) - return self._asyncioTestLoop.run_until_complete(task) def _callMaybeAsync(self, func, /, *args, **kwargs): - assert self._asyncioTestLoop is not None, 'asyncio test loop is not initialized' + assert self._asyncioRunner is not None, 'asyncio runner is not initialized' if inspect.iscoroutinefunction(func): - task = self._asyncioTestLoop.create_task( + return self._asyncioRunner.run( func(*args, **kwargs), context=self._asyncioTestContext, ) - return self._asyncioTestLoop.run_until_complete(task) else: return self._asyncioTestContext.run(func, *args, **kwargs) - def _setupAsyncioLoop(self): - assert self._asyncioTestLoop is None, 'asyncio test loop already initialized' - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - loop.set_debug(True) - self._asyncioTestLoop = loop + def _setupAsyncioRunner(self): + assert self._asyncioRunner is None, 'asyncio runner is already initialized' + runner = asyncio.Runner(debug=True) + self._asyncioRunner = runner - def _tearDownAsyncioLoop(self): - assert self._asyncioTestLoop is not None, 'asyncio test loop is not initialized' - loop = self._asyncioTestLoop - self._asyncioTestLoop = None - - try: - # cancel all tasks - to_cancel = asyncio.all_tasks(loop) - if not to_cancel: - return - - for task in to_cancel: - task.cancel() - - loop.run_until_complete( - asyncio.gather(*to_cancel, return_exceptions=True)) - - for task in to_cancel: - if task.cancelled(): - continue - if task.exception() is not None: - loop.call_exception_handler({ - 'message': 'unhandled exception during test shutdown', - 'exception': task.exception(), - 'task': task, - }) - # shutdown asyncgens - loop.run_until_complete(loop.shutdown_asyncgens()) - finally: - asyncio.set_event_loop(None) - loop.close() + def _tearDownAsyncioRunner(self): + runner = self._asyncioRunner + runner.close() def run(self, result=None): - self._setupAsyncioLoop() + self._setupAsyncioRunner() try: return super().run(result) finally: - self._tearDownAsyncioLoop() + self._tearDownAsyncioRunner() def debug(self): - self._setupAsyncioLoop() + self._setupAsyncioRunner() super().debug() - self._tearDownAsyncioLoop() + self._tearDownAsyncioRunner() def __del__(self): - if self._asyncioTestLoop is not None: - self._tearDownAsyncioLoop() + if self._asyncioRunner is not None: + self._tearDownAsyncioRunner() diff --git a/Misc/NEWS.d/next/Library/2022-03-18-22-46-18.bpo-47062.RNc99_.rst b/Misc/NEWS.d/next/Library/2022-03-18-22-46-18.bpo-47062.RNc99_.rst new file mode 100644 index 0000000000000..7d5bfc114a8d1 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2022-03-18-22-46-18.bpo-47062.RNc99_.rst @@ -0,0 +1 @@ +Implement :class:`asyncio.Runner` context manager.

1 0

bpo-46841: remove no-longer-used macro UPDATE_PREV_INSTR_OPARG (GH-32100)
by iritkatriel March 24, 2022

March 24, 2022

https://github.com/python/cpython/commit/2f49b97cc5426087b46515254b9a97a22e… commit: 2f49b97cc5426087b46515254b9a97a22ee8c807 branch: main author: Irit Katriel <1055913+iritkatriel(a)users.noreply.github.com> committer: iritkatriel <1055913+iritkatriel(a)users.noreply.github.com> date: 2022-03-24T18:54:35Z summary: bpo-46841: remove no-longer-used macro UPDATE_PREV_INSTR_OPARG (GH-32100) files: M Python/ceval.c diff --git a/Python/ceval.c b/Python/ceval.c index 42925b5b63048..e1d961fee6a83 100644 --- a/Python/ceval.c +++ b/Python/ceval.c @@ -1433,8 +1433,6 @@ eval_frame_handle_pending(PyThreadState *tstate) #define DEOPT_IF(cond, instname) if (cond) { goto instname ## _miss; } -#define UPDATE_PREV_INSTR_OPARG(instr, oparg) ((uint8_t*)(instr))[-1] = (oparg) - #define GLOBALS() frame->f_globals #define BUILTINS() frame->f_builtins

1 0

bpo-40465: Document random module changes in 3.11 What's new (#31818)
by serhiy-storchaka March 24, 2022

March 24, 2022

https://github.com/python/cpython/commit/12c0012cf97d21bc637056983ede0eaf4c… commit: 12c0012cf97d21bc637056983ede0eaf4c0d9c33 branch: main author: Tomáš Hrnčiar <thrnciar(a)redhat.com> committer: serhiy-storchaka <storchaka(a)gmail.com> date: 2022-03-24T15:22:58+02:00 summary: bpo-40465: Document random module changes in 3.11 What's new (#31818) files: M Doc/library/random.rst M Doc/whatsnew/3.11.rst diff --git a/Doc/library/random.rst b/Doc/library/random.rst index 96c6300ea16f5..72881b56a4b18 100644 --- a/Doc/library/random.rst +++ b/Doc/library/random.rst @@ -257,7 +257,7 @@ Functions for sequences .. versionchanged:: 3.11 The *population* must be a sequence. Automatic conversion of sets - to lists is longer supported. + to lists is no longer supported. .. _real-valued-distributions: diff --git a/Doc/whatsnew/3.11.rst b/Doc/whatsnew/3.11.rst index fe6c2e24c3d24..b2fdb4852360d 100644 --- a/Doc/whatsnew/3.11.rst +++ b/Doc/whatsnew/3.11.rst @@ -705,6 +705,11 @@ Changes in the Python API deprecated since Python 3.6. (Contributed by Serhiy Storchaka in :issue:`47066`.) +* The *population* parameter of :func:`random.sample` must be a sequence. + Automatic conversion of sets to lists is no longer supported. If the sample size + is larger than the population size, a :exc:`ValueError` is raised. + (Contributed by Raymond Hettinger in :issue:`40465`.) + Build Changes =============

1 0