Update asyncio to 3.14.5 (#8012)

Lib/asyncio/__main__.py

@@ -86,22 +86,27 @@ class REPLThread(threading.Thread):
         global return_code
 
         try:
-            banner = (
-                f'asyncio REPL {sys.version} on {sys.platform}\n'
-                f'Use "await" directly instead of "asyncio.run()".\n'
-                f'Type "help", "copyright", "credits" or "license" '
-                f'for more information.\n'
-            )
+            if not sys.flags.quiet:
+                banner = (
+                    f'asyncio REPL {sys.version} on {sys.platform}\n'
+                    f'Use "await" directly instead of "asyncio.run()".\n'
+                    f'Type "help", "copyright", "credits" or "license" '
+                    f'for more information.\n'
+                )
 
-            console.write(banner)
+                console.write(banner)
 
-            if startup_path := os.getenv("PYTHONSTARTUP"):
+            if not sys.flags.isolated and (startup_path := os.getenv("PYTHONSTARTUP")):
                 sys.audit("cpython.run_startup", startup_path)
-
-                import tokenize
-                with tokenize.open(startup_path) as f:
-                    startup_code = compile(f.read(), startup_path, "exec")
+                try:
+                    import tokenize
+                    with tokenize.open(startup_path) as f:
+                        startup_code = compile(f.read(), startup_path, "exec")
                     exec(startup_code, console.locals)
+                except SystemExit:
+                    raise
+                except BaseException:
+                    console.showtraceback()
 
             ps1 = getattr(sys, "ps1", ">>> ")
             if CAN_USE_PYREPL:
@@ -236,4 +241,5 @@ if __name__ == '__main__':
             break
 
     console.write('exiting asyncio REPL...\n')
+    loop.close()
     sys.exit(return_code)

Lib/asyncio/base_events.py

@@ -1345,6 +1345,17 @@ class BaseEventLoop(events.AbstractEventLoop):
         # have a chance to get called before "ssl_protocol.connection_made()".
         transport.pause_reading()
 
+        # gh-142352: move buffered StreamReader data to SSLProtocol
+        if server_side:
+            from .streams import StreamReaderProtocol
+            if isinstance(protocol, StreamReaderProtocol):
+                stream_reader = getattr(protocol, '_stream_reader', None)
+                if stream_reader is not None:
+                    buffer = stream_reader._buffer
+                    if buffer:
+                        ssl_protocol._incoming.write(buffer)
+                        buffer.clear()
+
         transport.set_protocol(ssl_protocol)
         conmade_cb = self.call_soon(ssl_protocol.connection_made, transport)
         resume_cb = self.call_soon(transport.resume_reading)

Lib/asyncio/base_subprocess.py

@@ -265,7 +265,7 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
             # to avoid hanging forever in self._wait as otherwise _exit_waiters
             # would never be woken up, we wake them up here.
             for waiter in self._exit_waiters:
-                if not waiter.cancelled():
+                if not waiter.done():
                     waiter.set_result(self._returncode)
         if all(p is not None and p.disconnected
                for p in self._pipes.values()):
@@ -278,7 +278,7 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
         finally:
             # wake up futures waiting for wait()
             for waiter in self._exit_waiters:
-                if not waiter.cancelled():
+                if not waiter.done():
                     waiter.set_result(self._returncode)
             self._exit_waiters = None
             self._loop = None

Lib/asyncio/futures.py

@@ -392,7 +392,7 @@ def _chain_future(source, destination):
 
     def _call_check_cancel(destination):
         if destination.cancelled():
-            if source_loop is None or source_loop is dest_loop:
+            if source_loop is None or source_loop is events._get_running_loop():
                 source.cancel()
             else:
                 source_loop.call_soon_threadsafe(source.cancel)
@@ -401,7 +401,7 @@ def _chain_future(source, destination):
         if (destination.cancelled() and
                 dest_loop is not None and dest_loop.is_closed()):
             return
-        if dest_loop is None or dest_loop is source_loop:
+        if dest_loop is None or dest_loop is events._get_running_loop():
             _set_state(destination, source)
         else:
             if dest_loop.is_closed():

Lib/asyncio/queues.py

@@ -37,7 +37,7 @@ class Queue(mixins._LoopBoundMixin):
     is an integer greater than 0, then "await put()" will block when the
     queue reaches maxsize, until an item is removed by get().
 
-    Unlike the standard library Queue, you can reliably know this Queue's size
+    Unlike queue.Queue, you can reliably know this Queue's size
     with qsize(), since your single-threaded asyncio application won't be
     interrupted between calling qsize() and doing an operation on the Queue.
     """

Lib/asyncio/windows_utils.py

@@ -10,7 +10,6 @@ import itertools
 import msvcrt
 import os
 import subprocess
-import tempfile
 import warnings
 
 
@@ -24,6 +23,7 @@ BUFSIZE = 8192
 PIPE = subprocess.PIPE
 STDOUT = subprocess.STDOUT
 _mmap_counter = itertools.count()
+_MAX_PIPE_ATTEMPTS = 20
 
 
 # Replacement for os.pipe() using handles instead of fds
@@ -31,10 +31,6 @@ _mmap_counter = itertools.count()
 
 def pipe(*, duplex=False, overlapped=(True, True), bufsize=BUFSIZE):
     """Like os.pipe() but with overlapped support and using handles not fds."""
-    address = tempfile.mktemp(
-        prefix=r'\\.\pipe\python-pipe-{:d}-{:d}-'.format(
-            os.getpid(), next(_mmap_counter)))
-
     if duplex:
         openmode = _winapi.PIPE_ACCESS_DUPLEX
         access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE
@@ -56,9 +52,20 @@ def pipe(*, duplex=False, overlapped=(True, True), bufsize=BUFSIZE):
 
     h1 = h2 = None
     try:
-        h1 = _winapi.CreateNamedPipe(
-            address, openmode, _winapi.PIPE_WAIT,
-            1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL)
+        for attempts in itertools.count():
+            address = r'\\.\pipe\python-pipe-{:d}-{:d}-{}'.format(
+                os.getpid(), next(_mmap_counter), os.urandom(8).hex())
+            try:
+                h1 = _winapi.CreateNamedPipe(
+                    address, openmode, _winapi.PIPE_WAIT,
+                    1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL)
+                break
+            except OSError as e:
+                if attempts >= _MAX_PIPE_ATTEMPTS:
+                    raise
+                if e.winerror not in (_winapi.ERROR_PIPE_BUSY,
+                                      _winapi.ERROR_ACCESS_DENIED):
+                    raise
 
         h2 = _winapi.CreateFile(
             address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING,
@@ -104,8 +111,9 @@ class PipeHandle:
 
     def close(self, *, CloseHandle=_winapi.CloseHandle):
         if self._handle is not None:
-            CloseHandle(self._handle)
+            handle = self._handle
             self._handle = None
+            CloseHandle(handle)
 
     def __del__(self, _warn=warnings.warn):
         if self._handle is not None:

Lib/test/test_asyncio/test_sock_lowlevel.py

@@ -427,6 +427,27 @@ class BaseSockTestsMixin:
             self.loop.run_until_complete(
                 self._basetest_datagram_recvfrom_into(server_address))
 
+    async def _basetest_datagram_recvfrom_into_wrong_size(self, server_address):
+        # Call sock_sendto() with a size larger than the buffer
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+            sock.setblocking(False)
+
+            buf = bytearray(5000)
+            data = b'\x01' * 4096
+            wrong_size = len(buf) + 1
+            await self.loop.sock_sendto(sock, data, server_address)
+            with self.assertRaises(ValueError):
+                await self.loop.sock_recvfrom_into(
+                    sock, buf, wrong_size)
+
+            size, addr = await self.loop.sock_recvfrom_into(sock, buf)
+            self.assertEqual(buf[:size], data)
+
+    def test_recvfrom_into_wrong_size(self):
+        with test_utils.run_udp_echo_server() as server_address:
+            self.loop.run_until_complete(
+                self._basetest_datagram_recvfrom_into_wrong_size(server_address))
+
     async def _basetest_datagram_sendto_blocking(self, server_address):
         # Sad path, sock.sendto() raises BlockingIOError
         # This involves patching sock.sendto() to raise BlockingIOError but
@@ -642,6 +663,10 @@ if sys.platform == 'win32':
                 self._basetest_datagram_send_to_non_listening_address(
                     recvfrom_into))
 
+        @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON; AssertionError: ValueError not raised")
+        def test_recvfrom_into_wrong_size(self):
+            return super().test_recvfrom_into_wrong_size()
+
 else:
     import selectors
 

Lib/test/test_asyncio/test_ssl.py

@@ -27,6 +27,7 @@ from test.test_asyncio import utils as test_utils
 
 MACOS = (sys.platform == 'darwin')
 BUF_MULTIPLIER = 1024 if not MACOS else 64
+HANDSHAKE_TIMEOUT = support.LONG_TIMEOUT
 
 
 def tearDownModule():
@@ -257,15 +258,12 @@ class TestSSL(test_utils.TestCase):
             await fut
 
         async def start_server():
-            extras = {}
-            extras = dict(ssl_handshake_timeout=support.SHORT_TIMEOUT)
-
             srv = await asyncio.start_server(
                 handle_client,
                 '127.0.0.1', 0,
                 family=socket.AF_INET,
                 ssl=sslctx,
-                **extras)
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
 
             try:
                 srv_socks = srv.sockets
@@ -322,14 +320,11 @@ class TestSSL(test_utils.TestCase):
             sock.close()
 
         async def client(addr):
-            extras = {}
-            extras = dict(ssl_handshake_timeout=support.SHORT_TIMEOUT)
-
             reader, writer = await asyncio.open_connection(
                 *addr,
                 ssl=client_sslctx,
                 server_hostname='',
-                **extras)
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
 
             writer.write(A_DATA)
             self.assertEqual(await reader.readexactly(2), b'OK')
@@ -349,7 +344,8 @@ class TestSSL(test_utils.TestCase):
             reader, writer = await asyncio.open_connection(
                 sock=sock,
                 ssl=client_sslctx,
-                server_hostname='')
+                server_hostname='',
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
 
             writer.write(A_DATA)
             self.assertEqual(await reader.readexactly(2), b'OK')
@@ -448,7 +444,7 @@ class TestSSL(test_utils.TestCase):
                 *addr,
                 ssl=client_sslctx,
                 server_hostname='',
-                ssl_handshake_timeout=support.SHORT_TIMEOUT)
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
             writer.close()
             await self.wait_closed(writer)
 
@@ -610,7 +606,7 @@ class TestSSL(test_utils.TestCase):
 
         extras = {}
         if server_ssl:
-            extras = dict(ssl_handshake_timeout=support.SHORT_TIMEOUT)
+            extras = dict(ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
 
         f = loop.create_task(
             loop.connect_accepted_socket(
@@ -659,7 +655,8 @@ class TestSSL(test_utils.TestCase):
             reader, writer = await asyncio.open_connection(
                 *addr,
                 ssl=client_sslctx,
-                server_hostname='')
+                server_hostname='',
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
 
             self.assertEqual(await reader.readline(), b'A\n')
             writer.write(b'B')
@@ -1153,14 +1150,11 @@ class TestSSL(test_utils.TestCase):
             await fut
 
         async def start_server():
-            extras = {}
-
             srv = await self.loop.create_server(
                 server_protocol_factory,
                 '127.0.0.1', 0,
                 family=socket.AF_INET,
-                ssl=sslctx_1,
-                **extras)
+                ssl=sslctx_1)
 
             try:
                 srv_socks = srv.sockets
@@ -1210,14 +1204,11 @@ class TestSSL(test_utils.TestCase):
             sock.close()
 
         async def client(addr):
-            extras = {}
-            extras = dict(ssl_handshake_timeout=support.SHORT_TIMEOUT)
-
             reader, writer = await asyncio.open_connection(
                 *addr,
                 ssl=client_sslctx,
                 server_hostname='',
-                **extras)
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
 
             writer.write(A_DATA)
             self.assertEqual(await reader.readexactly(2), b'OK')
@@ -1287,7 +1278,8 @@ class TestSSL(test_utils.TestCase):
             reader, writer = await asyncio.open_connection(
                 *addr,
                 ssl=client_sslctx,
-                server_hostname='')
+                server_hostname='',
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
             sslprotocol = writer.transport._ssl_protocol
             writer.write(b'ping')
             data = await reader.readexactly(4)
@@ -1399,7 +1391,8 @@ class TestSSL(test_utils.TestCase):
             reader, writer = await asyncio.open_connection(
                 *addr,
                 ssl=client_sslctx,
-                server_hostname='')
+                server_hostname='',
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
             writer.write(b'ping')
             data = await reader.readexactly(4)
             self.assertEqual(data, b'pong')
@@ -1530,7 +1523,8 @@ class TestSSL(test_utils.TestCase):
             reader, writer = await asyncio.open_connection(
                 *addr,
                 ssl=client_sslctx,
-                server_hostname='')
+                server_hostname='',
+                ssl_handshake_timeout=HANDSHAKE_TIMEOUT)
             writer.write(b'ping')
             data = await reader.readexactly(4)
             self.assertEqual(data, b'pong')

Lib/test/test_asyncio/test_streams.py

@@ -819,6 +819,48 @@ class StreamTests(test_utils.TestCase):
         self.assertEqual(msg1, b"hello world 1!\n")
         self.assertEqual(msg2, b"hello world 2!\n")
 
+    @unittest.skipIf(ssl is None, 'No ssl module')
+    def test_start_tls_buffered_data(self):
+        # gh-142352: test start_tls() with buffered data
+
+        async def server_handler(client_reader, client_writer):
+            # Wait for TLS ClientHello to be buffered before start_tls().
+            await client_reader._wait_for_data('test_start_tls_buffered_data'),
+            self.assertTrue(client_reader._buffer)
+            await client_writer.start_tls(test_utils.simple_server_sslcontext())
+
+            line = await client_reader.readline()
+            self.assertEqual(line, b"ping\n")
+            client_writer.write(b"pong\n")
+            await client_writer.drain()
+            client_writer.close()
+            await client_writer.wait_closed()
+
+        async def client(addr):
+            reader, writer = await asyncio.open_connection(*addr)
+            await writer.start_tls(test_utils.simple_client_sslcontext())
+
+            writer.write(b"ping\n")
+            await writer.drain()
+            line = await reader.readline()
+            self.assertEqual(line, b"pong\n")
+            writer.close()
+            await writer.wait_closed()
+
+        async def run_test():
+            server = await asyncio.start_server(
+                server_handler, socket_helper.HOSTv4, 0)
+            server_addr = server.sockets[0].getsockname()
+
+            await client(server_addr)
+            server.close()
+            await server.wait_closed()
+
+        messages = []
+        self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx))
+        self.loop.run_until_complete(run_test())
+        self.assertEqual(messages, [])
+
     def test_streamreader_constructor_without_loop(self):
         with self.assertRaisesRegex(RuntimeError, 'no current event loop'):
             asyncio.StreamReader()

Lib/test/test_asyncio/test_subprocess.py

@@ -111,6 +111,37 @@ class SubprocessTransportTests(test_utils.TestCase):
         )
         transport.close()
 
+    def test_proc_exited_no_invalid_state_error_on_exit_waiters(self):
+        # gh-145541: when _connect_pipes hasn't completed (so
+        # _pipes_connected is False) and the process exits, _try_finish()
+        # sets the result on exit waiters. Then _call_connection_lost() must
+        # not call set_result() again on the same waiters.
+        self.loop.set_exception_handler(
+            lambda loop, context: self.fail(
+                f"unexpected exception: {context}")
+        )
+        waiter = self.loop.create_future()
+        transport, protocol = self.create_transport(waiter)
+
+        # Simulate a waiter registered via _wait() before the process exits.
+        exit_waiter = self.loop.create_future()
+        transport._exit_waiters.append(exit_waiter)
+
+        # _connect_pipes hasn't completed, so _pipes_connected is False.
+        self.assertFalse(transport._pipes_connected)
+
+        # Simulate process exit. _try_finish() will set the result on
+        # exit_waiter because _pipes_connected is False, and then schedule
+        # _call_connection_lost() because _pipes is empty (vacuously all
+        # disconnected). _call_connection_lost() must skip exit_waiter
+        # because it's already done.
+        transport._process_exited(6)
+        self.loop.run_until_complete(waiter)
+
+        self.assertEqual(exit_waiter.result(), 6)
+
+        transport.close()
+
 
 class SubprocessMixin:
 

Lib/test/test_asyncio/test_tasks.py

@@ -2947,7 +2947,6 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
         return super().test_log_destroyed_pending_task()
 
 
-
 @unittest.skipUnless(hasattr(futures, '_CFuture') and
                      hasattr(tasks, '_CTask'),
                      'requires the C _asyncio module')
@@ -3030,7 +3029,6 @@ class PyTask_PyFuture_SubclassTests(BaseTaskTests, test_utils.TestCase):
     all_tasks = staticmethod(tasks._py_all_tasks)
     current_task = staticmethod(tasks._py_current_task)
 
-
 @unittest.skipUnless(hasattr(tasks, '_CTask'),
                      'requires the C _asyncio module')
 class CTask_Future_Tests(test_utils.TestCase):
@@ -3063,6 +3061,26 @@ class BaseTaskIntrospectionTests:
     _enter_task = None
     _leave_task = None
     all_tasks = None
+    Task = None
+
+    def test_register_task_resurrection(self):
+        register_task = self._register_task
+        class EvilLoop:
+            def get_debug(self):
+                return False
+
+            def call_exception_handler(self, context):
+                register_task(context["task"])
+
+        async def coro_fn ():
+            pass
+
+        coro = coro_fn()
+        self.addCleanup(coro.close)
+        loop = EvilLoop()
+        with self.assertRaises(AttributeError):
+            self.Task(coro, loop=loop)
+
 
     def test__register_task_1(self):
         class TaskLike:
@@ -3193,6 +3211,7 @@ class PyIntrospectionTests(test_utils.TestCase, BaseTaskIntrospectionTests):
     _leave_task = staticmethod(tasks._py_leave_task)
     all_tasks = staticmethod(tasks._py_all_tasks)
     current_task = staticmethod(tasks._py_current_task)
+    Task = tasks._PyTask
 
 
 @unittest.skipUnless(hasattr(tasks, '_c_register_task'),
@@ -3205,10 +3224,12 @@ class CIntrospectionTests(test_utils.TestCase, BaseTaskIntrospectionTests):
         _leave_task = staticmethod(tasks._c_leave_task)
         all_tasks = staticmethod(tasks._c_all_tasks)
         current_task = staticmethod(tasks._c_current_task)
+        Task = tasks._CTask
     else:
         _register_task = _unregister_task = _enter_task = _leave_task = None
 
 
+
 class BaseCurrentLoopTests:
     current_task = None
 
@@ -3698,6 +3719,30 @@ class RunCoroutineThreadsafeTests(test_utils.TestCase):
         (loop, context), kwargs = callback.call_args
         self.assertEqual(context['exception'], exc_context.exception)
 
+    def test_run_coroutine_threadsafe_and_cancel(self):
+        task = None
+        thread_future = None
+        # Use a custom task factory to capture the created Task
+        def task_factory(loop, coro):
+            nonlocal task
+            task = asyncio.Task(coro, loop=loop)
+            return task
+
+        self.addCleanup(self.loop.set_task_factory,
+                        self.loop.get_task_factory())
+
+        async def target():
+            nonlocal thread_future
+            self.loop.set_task_factory(task_factory)
+            thread_future = asyncio.run_coroutine_threadsafe(asyncio.sleep(10), self.loop)
+            await asyncio.sleep(0)
+
+            thread_future.cancel()
+
+        self.loop.run_until_complete(target())
+        self.assertTrue(task.cancelled())
+        self.assertTrue(thread_future.cancelled())
+
 
 class SleepTests(test_utils.TestCase):
     def setUp(self):

Lib/test/test_asyncio/test_windows_utils.py

@@ -77,6 +77,30 @@ class PipeTests(unittest.TestCase):
         else:
             raise RuntimeError('expected ERROR_INVALID_HANDLE')
 
+    def test_pipe_handle_close_after_external_close(self):
+        # gh-149388: PipeHandle.close() must clear ``_handle`` before calling
+        # CloseHandle so that if CloseHandle raises on a stale handle the
+        # PipeHandle is still marked closed and __del__ / subsequent close()
+        # calls are silent no-ops.
+        h1, h2 = windows_utils.pipe(overlapped=(False, False))
+        try:
+            p = windows_utils.PipeHandle(h1)
+            # Simulate an external close of the underlying handle (e.g.
+            # a finalizer race or a concurrent close on the same object).
+            _winapi.CloseHandle(p.handle)
+            # First close() still propagates the OSError from CloseHandle,
+            # but must clear ``_handle`` first.
+            with self.assertRaises(OSError):
+                p.close()
+            self.assertIsNone(p.handle)
+            # Second close() is a no-op.
+            p.close()
+            # __del__ through GC is also a silent no-op — no unraisable.
+            del p
+            support.gc_collect()
+        finally:
+            _winapi.CloseHandle(h2)
+
 
 class PopenTests(unittest.TestCase):
 
@@ -129,5 +153,25 @@ class PopenTests(unittest.TestCase):
             pass
 
 
+class OverlappedRefleakTests(unittest.TestCase):
+
+    def test_wsasendto_failure(self):
+        ov = _overlapped.Overlapped()
+        buf = bytearray(4096)
+        with self.assertRaises(OSError):
+            ov.WSASendTo(0x1234, buf, 0, ("127.0.0.1", 1))
+
+    def test_wsarecvfrom_failure(self):
+        ov = _overlapped.Overlapped()
+        with self.assertRaises(OSError):
+            ov.WSARecvFrom(0x1234, 1024, 0)
+
+    def test_wsarecvfrominto_failure(self):
+        ov = _overlapped.Overlapped()
+        buf = bytearray(4096)
+        with self.assertRaises(OSError):
+            ov.WSARecvFromInto(0x1234, buf, len(buf), 0)
+
+
 if __name__ == '__main__':
     unittest.main()