Align asyncio initialization and future state with Python 3.15 (#9136)

* Align asyncio initialization and future state with Python 3.15

Reject native Future and Task reinitialization, validate task contexts, and restore task registration after failed eager context entry. Generate matching asyncio and concurrent Future sources and tests with update_lib from CPython d220409a13b77de7a6c30abe16ea90db01528edc. Preserve the existing pending-task reinitialization regression.

Assisted-by: OpenAI Codex:model version unavailable

* Remove asyncio snippet duplicating canonical reinitialization coverage

Keep the exact-final Lib tests as the reinitialization coverage instead of adding a new snippet for the same contract. Native and library implementation bytes remain unchanged.

Assisted-by: OpenAI Codex:model version unavailable

* Validate both objects in asyncio awaited-by helpers

Match CPython by ignoring non-native Future/Task waiters in add and discard. Keep the existing native reentrant hashing fixture reachable using Future subclasses and use native instances for its membership checks.

Assisted-by: OpenAI Codex:model version unavailable

Lib/asyncio/futures.py

@@ -79,6 +79,10 @@ class Future:
         loop object used by the future. If it's not provided, the future uses
         the default event loop.
         """
+        if self._loop is not None:
+            raise RuntimeError(f"{self.__class__.__name__} object is already "
+                                "initialized")
+
         if loop is None:
             self._loop = events.get_event_loop()
         else:
@@ -351,22 +355,19 @@ def _set_concurrent_future_state(concurrent, source):
 def _copy_future_state(source, dest):
     """Internal helper to copy state from another Future.
 
-    The other Future may be a concurrent.futures.Future.
+    The other Future must be a concurrent.futures.Future.
     """
-    assert source.done()
     if dest.cancelled():
         return
     assert not dest.done()
-    if source.cancelled():
+    done, cancelled, result, exception = source._get_snapshot()
+    assert done
+    if cancelled:
         dest.cancel()
+    elif exception is not None:
+        dest.set_exception(_convert_future_exc(exception))
     else:
-        exception = source.exception()
-        if exception is not None:
-            dest.set_exception(_convert_future_exc(exception))
-        else:
-            result = source.result()
-            dest.set_result(result)
-
+        dest.set_result(result)
 
 def _chain_future(source, destination):
     """Chain two futures so that when one completes, so does the other.

Lib/asyncio/tasks.py

@@ -103,11 +103,20 @@ class Task(futures._PyFuture):  # Inherit Python Task implementation
         self._coro = coro
         if context is None:
             self._context = contextvars.copy_context()
+        elif not isinstance(context, contextvars.Context):
+            # gh-157301: the passed value must be a contextvars.Context
+            self._log_destroy_pending = False
+            raise TypeError('a contextvars.Context was expected, '
+                            f'got {type(context).__name__}')
         else:
             self._context = context
 
         if eager_start and self._loop.is_running():
-            self.__eager_start()
+            try:
+                self.__eager_start()
+            except:
+                self._log_destroy_pending = False
+                raise
         else:
             self._loop.call_soon(self.__step, context=self._context)
             _py_register_task(self)
@@ -440,15 +449,13 @@ def _release_waiter(waiter, *args):
 async def wait_for(fut, timeout):
     """Wait for the single Future or coroutine to complete, with timeout.
 
-    Coroutine will be wrapped in Task.
-
     Returns result of the Future or coroutine.  When a timeout occurs,
-    it cancels the task and raises TimeoutError.  To avoid the task
-    cancellation, wrap it in shield().
+    it cancels fut and raises TimeoutError.  To prevent fut from being
+    cancelled, wrap it in shield().
 
-    If the wait is cancelled, the task is also cancelled.
+    If the wait is cancelled, fut is also cancelled.
 
-    If the task suppresses the cancellation and returns a value instead,
+    If fut suppresses the cancellation and returns a value instead,
     that value is returned.
 
     This function is a coroutine.
@@ -524,6 +531,7 @@ async def _wait(fs, timeout, return_when, loop):
             timeout_handle.cancel()
         for f in fs:
             f.remove_done_callback(_on_completion)
+            futures.future_discard_from_awaited_by(f, cur_task)
 
     done, pending = set(), set()
     for f in fs:
@@ -542,6 +550,10 @@ async def _cancel_and_wait(fut):
     cb = functools.partial(_release_waiter, waiter)
     fut.add_done_callback(cb)
 
+    # gh-157058: awaiting the waiter leaves no edge on fut, add it here
+    cur_task = current_task()
+    futures.future_add_to_awaited_by(fut, cur_task)
+
     try:
         fut.cancel()
         # We cannot wait on *fut* directly to make
@@ -549,6 +561,7 @@ async def _cancel_and_wait(fut):
         await waiter
     finally:
         fut.remove_done_callback(cb)
+        futures.future_discard_from_awaited_by(fut, cur_task)
 
 
 class _AsCompletedIterator:
@@ -563,9 +576,11 @@ class _AsCompletedIterator:
         self._timeout_handle = None
 
         loop = events.get_event_loop()
+        self._cur_task = current_task()
         todo = {ensure_future(aw, loop=loop) for aw in set(aws)}
         for f in todo:
             f.add_done_callback(self._handle_completion)
+            futures.future_add_to_awaited_by(f, self._cur_task)
         if todo and timeout is not None:
             self._timeout_handle = (
                 loop.call_later(timeout, self._handle_timeout)
@@ -596,6 +611,7 @@ class _AsCompletedIterator:
     def _handle_timeout(self):
         for f in self._todo:
             f.remove_done_callback(self._handle_completion)
+            futures.future_discard_from_awaited_by(f, self._cur_task)
             self._done.put_nowait(None)  # Sentinel for _wait_for_one().
         self._todo.clear()  # Can't do todo.remove(f) in the loop.
 
@@ -603,6 +619,7 @@ class _AsCompletedIterator:
         if not self._todo:
             return  # _handle_timeout() was here first.
         self._todo.remove(f)
+        futures.future_discard_from_awaited_by(f, self._cur_task)
         self._done.put_nowait(f)
         if not self._todo and self._timeout_handle is not None:
             self._timeout_handle.cancel()
@@ -767,6 +784,11 @@ class _GatheringFuture(futures.Future):
         return ret
 
 
+def _discard_awaited_by(children, waiter, outer):
+    for fut in children:
+        futures.future_discard_from_awaited_by(fut, waiter)
+
+
 def gather(*coros_or_futures, return_exceptions=False):
     """Return a future aggregating results from the given coroutines/futures.
 
@@ -900,6 +922,10 @@ def gather(*coros_or_futures, return_exceptions=False):
         children.append(fut)
 
     outer = _GatheringFuture(children, loop=loop)
+    if cur_task is not None:
+        # gh-157213: a child outliving gather() must lose the awaited-by edge
+        outer.add_done_callback(
+            functools.partial(_discard_awaited_by, children, cur_task))
     # Run done callbacks after GatheringFuture created so any post-processing
     # can be performed at this point
     # optimization: in the special case that *all* futures finished eagerly,
@@ -996,6 +1022,9 @@ def shield(arg):
             # Keep only one callback to log on cancel
             inner.remove_done_callback(_log_on_exception)
             inner.add_done_callback(_log_on_exception)
+            if cur_task is not None:
+                inner.remove_done_callback(_clear_awaited_by_callback)
+                futures.future_discard_from_awaited_by(inner, cur_task)
 
     if cur_task is not None:
         inner.add_done_callback(_clear_awaited_by_callback)

Lib/concurrent/futures/_base.py

@@ -563,6 +563,33 @@ class Future(object):
             self._condition.notify_all()
         self._invoke_callbacks()
 
+    def _get_snapshot(self):
+        """Get a snapshot of the future's current state.
+
+        This method atomically retrieves the state in one lock acquisition,
+        which is significantly faster than multiple method calls.
+
+        Returns:
+            Tuple of (done, cancelled, result, exception)
+            - done: True if the future is done (cancelled or finished)
+            - cancelled: True if the future was cancelled
+            - result: The result if available and not cancelled
+            - exception: The exception if available and not cancelled
+        """
+        # Fast path: check if already finished without lock
+        if self._state == FINISHED:
+            return True, False, self._result, self._exception
+
+        # Need lock for other states since they can change
+        with self._condition:
+            # We have to check the state again after acquiring the lock
+            # because it may have changed in the meantime.
+            if self._state == FINISHED:
+                return True, False, self._result, self._exception
+            if self._state in {CANCELLED, CANCELLED_AND_NOTIFIED}:
+                return True, True, None, None
+            return False, False, None, None
+
     __class_getitem__ = classmethod(types.GenericAlias)
 
 class Executor(object):
@@ -676,5 +703,5 @@ class Executor(object):
 
 class BrokenExecutor(RuntimeError):
     """
-    Raised when a executor has become non-functional after a severe failure.
+    Raised when an executor has become non-functional after a severe failure.
     """

Lib/test/test_asyncio/test_futures.py

@@ -413,7 +413,7 @@ class BaseFutureTests:
     def test_copy_state(self):
         from asyncio.futures import _copy_future_state
 
-        f = self._new_future(loop=self.loop)
+        f = concurrent.futures.Future()
         f.set_result(10)
 
         newf = self._new_future(loop=self.loop)
@@ -421,7 +421,7 @@ class BaseFutureTests:
         self.assertTrue(newf.done())
         self.assertEqual(newf.result(), 10)
 
-        f_exception = self._new_future(loop=self.loop)
+        f_exception = concurrent.futures.Future()
         f_exception.set_exception(RuntimeError())
 
         newf_exception = self._new_future(loop=self.loop)
@@ -429,7 +429,7 @@ class BaseFutureTests:
         self.assertTrue(newf_exception.done())
         self.assertRaises(RuntimeError, newf_exception.result)
 
-        f_cancelled = self._new_future(loop=self.loop)
+        f_cancelled = concurrent.futures.Future()
         f_cancelled.cancel()
 
         newf_cancelled = self._new_future(loop=self.loop)
@@ -441,7 +441,7 @@ class BaseFutureTests:
         except BaseException as e:
             f_exc = e
 
-        f_conexc = self._new_future(loop=self.loop)
+        f_conexc = concurrent.futures.Future()
         f_conexc.set_exception(f_exc)
 
         newf_conexc = self._new_future(loop=self.loop)
@@ -454,6 +454,56 @@ class BaseFutureTests:
         newf_tb = ''.join(traceback.format_tb(newf_exc.__traceback__))
         self.assertEqual(newf_tb.count('raise concurrent.futures.InvalidStateError'), 1)
 
+    def test_copy_state_from_concurrent_futures(self):
+        """Test _copy_future_state from concurrent.futures.Future.
+
+        This tests the optimized path using _get_snapshot when available.
+        """
+        from asyncio.futures import _copy_future_state
+
+        # Test with a result
+        f_concurrent = concurrent.futures.Future()
+        f_concurrent.set_result(42)
+        f_asyncio = self._new_future(loop=self.loop)
+        _copy_future_state(f_concurrent, f_asyncio)
+        self.assertTrue(f_asyncio.done())
+        self.assertEqual(f_asyncio.result(), 42)
+
+        # Test with an exception
+        f_concurrent_exc = concurrent.futures.Future()
+        f_concurrent_exc.set_exception(ValueError("test exception"))
+        f_asyncio_exc = self._new_future(loop=self.loop)
+        _copy_future_state(f_concurrent_exc, f_asyncio_exc)
+        self.assertTrue(f_asyncio_exc.done())
+        with self.assertRaises(ValueError) as cm:
+            f_asyncio_exc.result()
+        self.assertEqual(str(cm.exception), "test exception")
+
+        # Test with cancelled state
+        f_concurrent_cancelled = concurrent.futures.Future()
+        f_concurrent_cancelled.cancel()
+        f_asyncio_cancelled = self._new_future(loop=self.loop)
+        _copy_future_state(f_concurrent_cancelled, f_asyncio_cancelled)
+        self.assertTrue(f_asyncio_cancelled.cancelled())
+
+        # Test that destination already cancelled prevents copy
+        f_concurrent_result = concurrent.futures.Future()
+        f_concurrent_result.set_result(10)
+        f_asyncio_precancelled = self._new_future(loop=self.loop)
+        f_asyncio_precancelled.cancel()
+        _copy_future_state(f_concurrent_result, f_asyncio_precancelled)
+        self.assertTrue(f_asyncio_precancelled.cancelled())
+
+        # Test exception type conversion
+        f_concurrent_invalid = concurrent.futures.Future()
+        f_concurrent_invalid.set_exception(concurrent.futures.InvalidStateError("invalid"))
+        f_asyncio_invalid = self._new_future(loop=self.loop)
+        _copy_future_state(f_concurrent_invalid, f_asyncio_invalid)
+        self.assertTrue(f_asyncio_invalid.done())
+        with self.assertRaises(asyncio.exceptions.InvalidStateError) as cm:
+            f_asyncio_invalid.result()
+        self.assertEqual(str(cm.exception), "invalid")
+
     def test_iter(self):
         fut = self._new_future(loop=self.loop)
 
@@ -700,6 +750,10 @@ class BaseFutureTests:
         self.assertIsNotNone(exc)
         self.assertListEqual(gc.get_referrers(exc), [])
 
+    def test_future_disallow_multiple_initialization(self):
+        f = self._new_future(loop=self.loop)
+        with self.assertRaises(RuntimeError, msg="is already initialized"):
+            f.__init__(loop=self.loop)
 
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')
@@ -1041,33 +1095,6 @@ class BaseFutureDoneCallbackTests():
             fut.add_done_callback(fut_callback_0)
             self.assertRaises(ReachableCode, fut.set_result, "boom")
 
-    def test_use_after_free_on_fut_context_0_with_evil__getattribute__(self):
-        # see: https://github.com/python/cpython/issues/125984
-
-        class EvilEventLoop(SimpleEvilEventLoop):
-            def call_soon(self, *args, **kwargs):
-                super().call_soon(*args, **kwargs)
-                raise ReachableCode
-
-            def __getattribute__(self, name):
-                if name == 'call_soon':
-                    # resets the future's event loop
-                    fut.__init__(loop=SimpleEvilEventLoop())
-                return object.__getattribute__(self, name)
-
-        evil_loop = EvilEventLoop()
-        with mock.patch.object(self, 'loop', evil_loop):
-            fut = self._new_future()
-            self.assertIs(fut.get_loop(), evil_loop)
-
-            fut_callback_0 = mock.Mock()
-            fut_context_0 = mock.Mock()
-            fut.add_done_callback(fut_callback_0, context=fut_context_0)
-            del fut_context_0
-            del fut_callback_0
-            self.assertRaises(ReachableCode, fut.set_result, "boom")
-
-
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')
 class CFutureDoneCallbackTests(BaseFutureDoneCallbackTests,

Lib/test/test_asyncio/test_tasks.py

@@ -1218,6 +1218,38 @@ class BaseTaskTests:
         loop.advance_time(10)
         loop.run_until_complete(asyncio.wait([a, b]))
 
+    def test_wait_discards_awaited_by_for_pending(self):
+        # gh-152569: wait() must remove itself from the await-graph of every
+        # future once it returns, including futures that never resolved.
+        async def coro():
+            immortal = self.loop.create_future()
+            done = self.new_task(self.loop, asyncio.sleep(0))
+            await asyncio.wait({done, immortal},
+                               return_when=asyncio.FIRST_COMPLETED)
+            self.assertFalse(immortal._asyncio_awaited_by)
+            immortal.cancel()
+
+        self.loop.run_until_complete(self.new_task(self.loop, coro()))
+
+    def test_gather_discards_awaited_by_for_pending(self):
+        # gh-157213: a child outliving gather() must lose the awaited-by edge
+        async def fail():
+            raise ValueError
+
+        async def survivor():
+            await asyncio.Future()
+
+        async def coro():
+            t = self.new_task(self.loop, survivor())
+            with self.assertRaises(ValueError):
+                await asyncio.gather(t, fail())
+            self.assertFalse(t._asyncio_awaited_by)
+            t.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await t
+
+        self.loop.run_until_complete(self.new_task(self.loop, coro()))
+
     def test_wait_really_done(self):
         # there is possibility that some tasks in the pending list
         # became done but their callbacks haven't all been called yet
@@ -2115,6 +2147,8 @@ class BaseTaskTests:
         test_utils.run_briefly(self.loop)
         self.assertTrue(outer.cancelled())
         self.assertEqual(0, 0 if outer._callbacks is None else len(outer._callbacks))
+        self.assertFalse(inner._asyncio_awaited_by)
+        self.assertTrue({f for f, _ctx in inner._callbacks or []} <= {asyncio.tasks._log_on_exception})
 
     def test_shield_cancel_outer_result(self):
         mock_handler = mock.Mock()
@@ -2140,6 +2174,21 @@ class BaseTaskTests:
         test_utils.run_briefly(self.loop)
         mock_handler.assert_called_once()
 
+    def test_shield_cancel_outer_in_task(self):
+        inner = self.new_future(self.loop)
+
+        async def coro():
+            outer = asyncio.shield(inner)
+            self.assertNotEqual(0, len(inner._callbacks))
+            outer.cancel()
+            await asyncio.sleep(0)
+            self.assertTrue(outer.cancelled())
+
+        task = self.new_task(self.loop, coro())
+        self.loop.run_until_complete(task)
+        self.assertFalse(inner._asyncio_awaited_by)
+        self.assertTrue({f for f, _ctx in inner._callbacks or []} <= {asyncio.tasks._log_on_exception})
+
     def test_shield_duplicate_log_once(self):
         mock_handler = mock.Mock()
         self.loop.set_exception_handler(mock_handler)
@@ -2581,6 +2630,68 @@ class BaseTaskTests:
         finally:
             loop.close()
 
+    def test_context_not_a_context(self):
+        # gh-157301
+        async def coro():
+            pass
+
+        loop = asyncio.new_event_loop()
+        c = coro()
+        try:
+            with self.assertRaises(TypeError):
+                self.new_task(loop, c, context='not a context')
+        finally:
+            c.close()
+            loop.close()
+
+    def test_context_not_a_context_leaves_loop_usable(self):
+        # gh-157301
+        async def coro():
+            pass
+
+        async def main():
+            c = coro()
+            try:
+                with self.assertRaises(TypeError):
+                    self.new_task(loop, c, context='not a context',
+                                  eager_start=True)
+            finally:
+                c.close()
+            await asyncio.sleep(0)
+
+        loop = asyncio.new_event_loop()
+        loop.call_later(support.SHORT_TIMEOUT, loop.stop)
+        try:
+            loop.run_until_complete(self.new_task(loop, main()))
+        finally:
+            loop.close()
+
+    def test_context_already_entered_leaves_loop_usable(self):
+        # gh-157301
+        async def coro():
+            pass
+
+        async def main():
+            ctx = contextvars.copy_context()
+
+            def inside():
+                c = coro()
+                try:
+                    with self.assertRaises(RuntimeError):
+                        self.new_task(loop, c, context=ctx, eager_start=True)
+                finally:
+                    c.close()
+
+            ctx.run(inside)
+            await asyncio.sleep(0)
+
+        loop = asyncio.new_event_loop()
+        loop.call_later(support.SHORT_TIMEOUT, loop.stop)
+        try:
+            loop.run_until_complete(self.new_task(loop, main()))
+        finally:
+            loop.close()
+
     def test_context_2(self):
         cvar = contextvars.ContextVar('cvar', default='nope')
 
@@ -2739,6 +2850,18 @@ class BaseTaskTests:
             self.assertEqual(name, "example")
             await t
 
+    def test_eager_start_true_no_loop(self):
+        # gh-154695: eager_start must use the resolved loop, not loop=None.
+        async def asyncfn():
+            return 42
+
+        async def main():
+            t = self.__class__.Task(asyncfn(), eager_start=True)
+            self.assertTrue(t.done())
+            self.assertEqual(await t, 42)
+
+        asyncio.run(main(), loop_factory=asyncio.EventLoop)
+
     def test_eager_start_false(self):
         name = None
 
@@ -2776,28 +2899,17 @@ class BaseTaskTests:
         finally:
             loop.close()
 
-    def test_proper_refcounts(self):
-        # see: https://github.com/python/cpython/issues/126083
-        class Break:
-            def __str__(self):
-                raise RuntimeError("break")
+    def test_task_disallow_multiple_initialization(self):
+        async def foo():
+            pass
 
-        obj = object()
-        initial_refcount = sys.getrefcount(obj)
-
-        coro = coroutine_function()
-        with contextlib.closing(asyncio.EventLoop()) as loop:
-            task = asyncio.Task.__new__(asyncio.Task)
-            for _ in range(5):
-                with self.assertRaisesRegex(RuntimeError, 'break'):
-                    task.__init__(coro, loop=loop, context=obj, name=Break())
-
-            coro.close()
-            task._log_destroy_pending = False
-            del task
-
-            self.assertEqual(sys.getrefcount(obj), initial_refcount)
+        coro = foo()
+        self.addCleanup(coro.close)
+        task = self.new_task(self.loop, coro)
+        task._log_destroy_pending = False
 
+        with self.assertRaises(RuntimeError, msg="is already initialized"):
+            task.__init__(coro, loop=self.loop)
 
 def add_subclass_tests(cls):
     BaseTask = cls.Task
@@ -2921,19 +3033,6 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
     all_tasks = getattr(tasks, '_c_all_tasks', None)
     current_task = staticmethod(getattr(tasks, '_c_current_task', None))
 
-    @support.refcount_test
-    def test_refleaks_in_task___init__(self):
-        gettotalrefcount = support.get_attribute(sys, 'gettotalrefcount')
-        async def coro():
-            pass
-        task = self.new_task(self.loop, coro())
-        self.loop.run_until_complete(task)
-        refs_before = gettotalrefcount()
-        for i in range(100):
-            task.__init__(coro(), loop=self.loop)
-            self.loop.run_until_complete(task)
-        self.assertAlmostEqual(gettotalrefcount() - refs_before, 0, delta=10)
-
     def test_del__log_destroy_pending_segfault(self):
         async def coro():
             pass
@@ -2942,6 +3041,16 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
         with self.assertRaises(AttributeError):
             del task._log_destroy_pending
 
+    def test_get_context_uninitialized_segfault(self):
+        # https://github.com/python/cpython/issues/154871
+
+        class UninitializedTask(self.Task):
+            def __init__(self, *args, **kwargs):
+                pass
+
+        task = UninitializedTask()
+        self.assertIsNone(task.get_context())
+
 
 @unittest.skipUnless(hasattr(futures, '_CFuture') and
                      hasattr(tasks, '_CTask'),

Lib/test/test_concurrent_futures/test_future.py

@@ -6,6 +6,7 @@ from concurrent.futures._base import (
     PENDING, RUNNING, CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED, Future)
 
 from test import support
+from test.support import threading_helper
 
 from .util import (
     PENDING_FUTURE, RUNNING_FUTURE, CANCELLED_FUTURE,
@@ -282,6 +283,62 @@ class FutureTests(BaseTestCase):
 
         self.assertEqual(f.exception(), e)
 
+    def test_get_snapshot(self):
+        """Test the _get_snapshot method for atomic state retrieval."""
+        # Test with a pending future
+        f = Future()
+        done, cancelled, result, exception = f._get_snapshot()
+        self.assertFalse(done)
+        self.assertFalse(cancelled)
+        self.assertIsNone(result)
+        self.assertIsNone(exception)
+
+        # Test with a finished future (successful result)
+        f = Future()
+        f.set_result(42)
+        done, cancelled, result, exception = f._get_snapshot()
+        self.assertTrue(done)
+        self.assertFalse(cancelled)
+        self.assertEqual(result, 42)
+        self.assertIsNone(exception)
+
+        # Test with a finished future (exception)
+        f = Future()
+        exc = ValueError("test error")
+        f.set_exception(exc)
+        done, cancelled, result, exception = f._get_snapshot()
+        self.assertTrue(done)
+        self.assertFalse(cancelled)
+        self.assertIsNone(result)
+        self.assertIs(exception, exc)
+
+        # Test with a cancelled future
+        f = Future()
+        f.cancel()
+        done, cancelled, result, exception = f._get_snapshot()
+        self.assertTrue(done)
+        self.assertTrue(cancelled)
+        self.assertIsNone(result)
+        self.assertIsNone(exception)
+
+        # Test concurrent access (basic thread safety check)
+        f = Future()
+        f.set_result(100)
+        results = []
+
+        def get_snapshot():
+            for _ in range(1000):
+                snapshot = f._get_snapshot()
+                results.append(snapshot)
+
+        threads = [threading.Thread(target=get_snapshot) for _ in range(4)]
+        with threading_helper.start_threads(threads):
+            pass
+        # All snapshots should be identical for a finished future
+        expected = (True, False, 100, None)
+        for result in results:
+            self.assertEqual(result, expected)
+
 
 def setUpModule():
     setup_module()

crates/stdlib/src/_asyncio.rs

@@ -188,6 +188,13 @@ pub(crate) mod _asyncio {
             loop_: Option<PyObjectRef>,
             vm: &VirtualMachine,
         ) -> PyResult<()> {
+            if zelf.fut_loop.read().is_some() {
+                return Err(vm.new_runtime_error(format!(
+                    "{} object is already initialized",
+                    zelf.class().fully_qualified_name(vm)?
+                )));
+            }
+
             // Get the event loop
             let loop_obj = match loop_ {
                 Some(l) if !vm.is_none(&l) => l,
@@ -1183,6 +1190,13 @@ pub(crate) mod _asyncio {
     )]
     impl PyTask {
         fn py_init(zelf: &Py<Self>, args: TaskInitArgs, vm: &VirtualMachine) -> PyResult<()> {
+            if zelf.base.fut_loop.read().is_some() {
+                return Err(vm.new_runtime_error(format!(
+                    "{} object is already initialized",
+                    zelf.class().fully_qualified_name(vm)?
+                )));
+            }
+
             // Validate coroutine
             if !is_coroutine(args.coro.clone(), vm)? {
                 return Err(vm.new_type_error(format!(
@@ -1217,8 +1231,18 @@ pub(crate) mod _asyncio {
 
             // Get or create context
             let context = match args.context {
-                Some(c) => c,
-                None => get_copy_context(vm)?,
+                Some(c) if !vm.is_none(&c) => {
+                    if !c.class().is(PyContext::class(&vm.ctx)) {
+                        zelf.task_log_destroy_pending
+                            .store(false, Ordering::Relaxed);
+                        return Err(vm.new_type_error(format!(
+                            "a contextvars.Context was expected, got {}",
+                            c.class().fully_qualified_name(vm)?
+                        )));
+                    }
+                    c
+                }
+                _ => get_copy_context(vm)?,
             };
             *zelf.task_context.write() = Some(context);
 
@@ -2043,7 +2067,25 @@ pub(crate) mod _asyncio {
                     match ctx.downcast::<PyContext>() {
                         Ok(ctx) => {
                             // Only exit a context that was actually entered.
-                            PyContext::enter(&ctx, vm)?;
+                            if let Err(error) = PyContext::enter(&ctx, vm) {
+                                zelf.task_log_destroy_pending
+                                    .store(false, Ordering::Relaxed);
+                                let _ = _swap_current_task(
+                                    LoopTaskArgs {
+                                        loop_: loop_obj,
+                                        task: prev_task,
+                                    },
+                                    vm,
+                                );
+                                let _ = _unregister_eager_task(
+                                    TaskArg {
+                                        task: task_obj.clone(),
+                                    },
+                                    vm,
+                                );
+                                let _ = _unregister_task(TaskArg { task: task_obj }, vm);
+                                return Err(error);
+                            }
                             let result = vm.call_method(&c, "send", (vm.ctx.none(),));
                             PyContext::exit(&ctx, vm)?;
                             result
@@ -2839,8 +2881,12 @@ pub(crate) mod _asyncio {
         waiter: PyObjectRef,
         vm: &VirtualMachine,
     ) -> PyResult<()> {
-        // Only operate on native Future/Task objects (including subclasses).
+        // Both arguments must be native Future/Task objects (including subclasses).
         // Non-native objects are silently ignored.
+        if waiter.downcast_ref::<PyTask>().is_none() && waiter.downcast_ref::<PyFuture>().is_none()
+        {
+            return Ok(());
+        }
         if let Some(task) = fut.downcast_ref::<PyTask>() {
             return task.base.awaited_by_add(waiter, vm);
         }
@@ -2856,8 +2902,12 @@ pub(crate) mod _asyncio {
         waiter: PyObjectRef,
         vm: &VirtualMachine,
     ) -> PyResult<()> {
-        // Only operate on native Future/Task objects (including subclasses).
+        // Both arguments must be native Future/Task objects (including subclasses).
         // Non-native objects are silently ignored.
+        if waiter.downcast_ref::<PyTask>().is_none() && waiter.downcast_ref::<PyFuture>().is_none()
+        {
+            return Ok(());
+        }
         if let Some(task) = fut.downcast_ref::<PyTask>() {
             return task.base.awaited_by_discard(&waiter, vm);
         }

extra_tests/snippets/stdlib_asyncio.py

@@ -76,9 +76,9 @@ finally:
 # to the same future; the field must not be locked while that runs.
 
 
-class Reentrant:
+class Reentrant(_asyncio.Future):
     def __hash__(self):
-        _asyncio.future_add_to_awaited_by(awaited, Reentrant())
+        _asyncio.future_add_to_awaited_by(awaited, Reentrant(loop=object()))
         return 1
 
     def __eq__(self, other):
@@ -86,15 +86,15 @@ class Reentrant:
 
 
 awaited = _asyncio.Future(loop=object())
-_asyncio.future_add_to_awaited_by(awaited, Reentrant())
+_asyncio.future_add_to_awaited_by(awaited, Reentrant(loop=object()))
 with assert_raises(RecursionError):
     # converting the single waiter into a set hashes both of them
-    _asyncio.future_add_to_awaited_by(awaited, Reentrant())
+    _asyncio.future_add_to_awaited_by(awaited, Reentrant(loop=object()))
 
 plain = _asyncio.Future(loop=object())
-waiter = object()
+waiter = _asyncio.Future(loop=object())
 _asyncio.future_add_to_awaited_by(plain, waiter)
-_asyncio.future_add_to_awaited_by(plain, object())
+_asyncio.future_add_to_awaited_by(plain, _asyncio.Future(loop=object()))
 assert waiter in plain._asyncio_awaited_by
 _asyncio.future_discard_from_awaited_by(plain, waiter)
 assert waiter not in plain._asyncio_awaited_by