diff --git a/Lib/asyncio/__main__.py b/Lib/asyncio/__main__.py index e07dd52a2..44e14771b 100644 --- a/Lib/asyncio/__main__.py +++ b/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) diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index 8cbb71f70..b83b84181 100644 --- a/Lib/asyncio/base_events.py +++ b/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) diff --git a/Lib/asyncio/base_subprocess.py b/Lib/asyncio/base_subprocess.py index 321a4e5d5..224b18838 100644 --- a/Lib/asyncio/base_subprocess.py +++ b/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 diff --git a/Lib/asyncio/futures.py b/Lib/asyncio/futures.py index d1df67073..e8a99556b 100644 --- a/Lib/asyncio/futures.py +++ b/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(): diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py index 084fccaaf..756216fac 100644 --- a/Lib/asyncio/queues.py +++ b/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. """ diff --git a/Lib/asyncio/windows_utils.py b/Lib/asyncio/windows_utils.py index ef277fac3..d6393f0b1 100644 --- a/Lib/asyncio/windows_utils.py +++ b/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: diff --git a/Lib/test/test_asyncio/test_sock_lowlevel.py b/Lib/test/test_asyncio/test_sock_lowlevel.py index df4ec7948..a34e15b4d 100644 --- a/Lib/test/test_asyncio/test_sock_lowlevel.py +++ b/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 diff --git a/Lib/test/test_asyncio/test_ssl.py b/Lib/test/test_asyncio/test_ssl.py index ab4a2316f..932b1dace 100644 --- a/Lib/test/test_asyncio/test_ssl.py +++ b/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') diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 5f0fc6a7a..739a82e57 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/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() diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py index bf3017407..c08eb7cf2 100644 --- a/Lib/test/test_asyncio/test_subprocess.py +++ b/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: diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py index 9fbad3d0e..5a4157c97 100644 --- a/Lib/test/test_asyncio/test_tasks.py +++ b/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): diff --git a/Lib/test/test_asyncio/test_windows_utils.py b/Lib/test/test_asyncio/test_windows_utils.py index 97f078ff9..509697613 100644 --- a/Lib/test/test_asyncio/test_windows_utils.py +++ b/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()