From 8dd6655ac9de9d199e23459e07a73f38bdcf8802 Mon Sep 17 00:00:00 2001 From: "Jeong, YunWon" Date: Tue, 30 Dec 2025 13:47:33 +0900 Subject: [PATCH 1/3] PythonFinallizationError --- crates/vm/src/exceptions.rs | 13 +++++++++++++ crates/vm/src/vm/vm_new.rs | 1 + 2 files changed, 14 insertions(+) diff --git a/crates/vm/src/exceptions.rs b/crates/vm/src/exceptions.rs index b1a654d58c4..5d0c1c6f3ed 100644 --- a/crates/vm/src/exceptions.rs +++ b/crates/vm/src/exceptions.rs @@ -493,6 +493,7 @@ pub struct ExceptionZoo { pub runtime_error: &'static Py, pub not_implemented_error: &'static Py, pub recursion_error: &'static Py, + pub python_finalization_error: &'static Py, pub syntax_error: &'static Py, pub incomplete_input_error: &'static Py, pub indentation_error: &'static Py, @@ -804,6 +805,7 @@ impl ExceptionZoo { let runtime_error = PyRuntimeError::init_builtin_type(); let not_implemented_error = PyNotImplementedError::init_builtin_type(); let recursion_error = PyRecursionError::init_builtin_type(); + let python_finalization_error = PyPythonFinalizationError::init_builtin_type(); let syntax_error = PySyntaxError::init_builtin_type(); let incomplete_input_error = PyIncompleteInputError::init_builtin_type(); @@ -879,6 +881,7 @@ impl ExceptionZoo { runtime_error, not_implemented_error, recursion_error, + python_finalization_error, syntax_error, incomplete_input_error, indentation_error, @@ -992,6 +995,11 @@ impl ExceptionZoo { extend_exception!(PyRuntimeError, ctx, excs.runtime_error); extend_exception!(PyNotImplementedError, ctx, excs.not_implemented_error); extend_exception!(PyRecursionError, ctx, excs.recursion_error); + extend_exception!( + PyPythonFinalizationError, + ctx, + excs.python_finalization_error + ); extend_exception!(PySyntaxError, ctx, excs.syntax_error, { "msg" => ctx.new_static_getset( @@ -2111,6 +2119,11 @@ pub(super) mod types { #[repr(transparent)] pub struct PyRecursionError(PyRuntimeError); + #[pyexception(name, base = PyRuntimeError, ctx = "python_finalization_error", impl)] + #[derive(Debug)] + #[repr(transparent)] + pub struct PyPythonFinalizationError(PyRuntimeError); + #[pyexception(name, base = PyException, ctx = "syntax_error")] #[derive(Debug)] #[repr(transparent)] diff --git a/crates/vm/src/vm/vm_new.rs b/crates/vm/src/vm/vm_new.rs index ba09c8ecf69..4f25d6e2036 100644 --- a/crates/vm/src/vm/vm_new.rs +++ b/crates/vm/src/vm/vm_new.rs @@ -600,5 +600,6 @@ impl VirtualMachine { define_exception_fn!(fn new_zero_division_error, zero_division_error, ZeroDivisionError); define_exception_fn!(fn new_overflow_error, overflow_error, OverflowError); define_exception_fn!(fn new_runtime_error, runtime_error, RuntimeError); + define_exception_fn!(fn new_python_finalization_error, python_finalization_error, PythonFinalizationError); define_exception_fn!(fn new_memory_error, memory_error, MemoryError); } From f306915f844801658d268c753b67ca8d59a4dbf7 Mon Sep 17 00:00:00 2001 From: "Jeong, YunWon" Date: Tue, 30 Dec 2025 13:48:02 +0900 Subject: [PATCH 2/3] fix posixprocess --- crates/stdlib/src/posixsubprocess.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/crates/stdlib/src/posixsubprocess.rs b/crates/stdlib/src/posixsubprocess.rs index 1ebf6619c24..a6badc081d2 100644 --- a/crates/stdlib/src/posixsubprocess.rs +++ b/crates/stdlib/src/posixsubprocess.rs @@ -33,6 +33,18 @@ mod _posixsubprocess { #[pyfunction] fn fork_exec(args: ForkExecArgs<'_>, vm: &VirtualMachine) -> PyResult { + // Check for interpreter shutdown when preexec_fn is used + if args.preexec_fn.is_some() + && vm + .state + .finalizing + .load(std::sync::atomic::Ordering::Acquire) + { + return Err(vm.new_python_finalization_error( + "preexec_fn not supported at interpreter shutdown".to_owned(), + )); + } + let extra_groups = args .groups_list .as_ref() From 406376cc06f3fa70f7aacab92ab5fd01ee295622 Mon Sep 17 00:00:00 2001 From: "Jeong, YunWon" Date: Tue, 30 Dec 2025 13:46:34 +0900 Subject: [PATCH 3/3] upgrade multiprocessing from 3.13.11 --- Lib/multiprocessing/connection.py | 2 +- Lib/multiprocessing/resource_tracker.py | 253 ++++++++++++++++-------- Lib/multiprocessing/util.py | 81 +++++++- 3 files changed, 257 insertions(+), 79 deletions(-) diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index 8caddd204d7..abd88adf76e 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -74,7 +74,7 @@ def arbitrary_address(family): if family == 'AF_INET': return ('localhost', 0) elif family == 'AF_UNIX': - return tempfile.mktemp(prefix='listener-', dir=util.get_temp_dir()) + return tempfile.mktemp(prefix='sock-', dir=util.get_temp_dir()) elif family == 'AF_PIPE': return tempfile.mktemp(prefix=r'\\.\pipe\pyc-%d-%d-' % (os.getpid(), next(_mmap_counter)), dir="") diff --git a/Lib/multiprocessing/resource_tracker.py b/Lib/multiprocessing/resource_tracker.py index 05633ac21a2..22e3bbcf21b 100644 --- a/Lib/multiprocessing/resource_tracker.py +++ b/Lib/multiprocessing/resource_tracker.py @@ -15,11 +15,15 @@ # this resource tracker process, "killall python" would probably leave unlinked # resources. +import base64 import os import signal import sys import threading import warnings +from collections import deque + +import json from . import spawn from . import util @@ -66,6 +70,14 @@ def __init__(self): self._fd = None self._pid = None self._exitcode = None + self._reentrant_messages = deque() + + # True to use colon-separated lines, rather than JSON lines, + # for internal communication. (Mainly for testing). + # Filenames not supported by the simple format will always be sent + # using JSON. + # The reader should understand all formats. + self._use_simple_format = True def _reentrant_call_error(self): # gh-109629: this happens if an explicit call to the ResourceTracker @@ -102,7 +114,7 @@ def _stop_locked( # This shouldn't happen (it might when called by a finalizer) # so we check for it anyway. if self._lock._recursion_count() > 1: - return self._reentrant_call_error() + raise self._reentrant_call_error() if self._fd is None: # not running return @@ -113,7 +125,12 @@ def _stop_locked( close(self._fd) self._fd = None - _, status = waitpid(self._pid, 0) + try: + _, status = waitpid(self._pid, 0) + except ChildProcessError: + self._pid = None + self._exitcode = None + return self._pid = None @@ -132,76 +149,119 @@ def ensure_running(self): This can be run from any process. Usually a child process will use the resource created by its parent.''' + return self._ensure_running_and_write() + + def _teardown_dead_process(self): + os.close(self._fd) + + # Clean-up to avoid dangling processes. + try: + # _pid can be None if this process is a child from another + # python process, which has started the resource_tracker. + if self._pid is not None: + os.waitpid(self._pid, 0) + except ChildProcessError: + # The resource_tracker has already been terminated. + pass + self._fd = None + self._pid = None + self._exitcode = None + + warnings.warn('resource_tracker: process died unexpectedly, ' + 'relaunching. Some resources might leak.') + + def _launch(self): + fds_to_pass = [] + try: + fds_to_pass.append(sys.stderr.fileno()) + except Exception: + pass + r, w = os.pipe() + try: + fds_to_pass.append(r) + # process will out live us, so no need to wait on pid + exe = spawn.get_executable() + args = [ + exe, + *util._args_from_interpreter_flags(), + '-c', + f'from multiprocessing.resource_tracker import main;main({r})', + ] + # bpo-33613: Register a signal mask that will block the signals. + # This signal mask will be inherited by the child that is going + # to be spawned and will protect the child from a race condition + # that can make the child die before it registers signal handlers + # for SIGINT and SIGTERM. The mask is unregistered after spawning + # the child. + prev_sigmask = None + try: + if _HAVE_SIGMASK: + prev_sigmask = signal.pthread_sigmask(signal.SIG_BLOCK, _IGNORED_SIGNALS) + pid = util.spawnv_passfds(exe, args, fds_to_pass) + finally: + if prev_sigmask is not None: + signal.pthread_sigmask(signal.SIG_SETMASK, prev_sigmask) + except: + os.close(w) + raise + else: + self._fd = w + self._pid = pid + finally: + os.close(r) + + def _make_probe_message(self): + """Return a probe message.""" + if self._use_simple_format: + return b'PROBE:0:noop\n' + return ( + json.dumps( + {"cmd": "PROBE", "rtype": "noop"}, + ensure_ascii=True, + separators=(",", ":"), + ) + + "\n" + ).encode("ascii") + + def _ensure_running_and_write(self, msg=None): with self._lock: if self._lock._recursion_count() > 1: # The code below is certainly not reentrant-safe, so bail out - return self._reentrant_call_error() + if msg is None: + raise self._reentrant_call_error() + return self._reentrant_messages.append(msg) + if self._fd is not None: # resource tracker was launched before, is it still running? - if self._check_alive(): - # => still alive - return - # => dead, launch it again - os.close(self._fd) - - # Clean-up to avoid dangling processes. + if msg is None: + to_send = self._make_probe_message() + else: + to_send = msg try: - # _pid can be None if this process is a child from another - # python process, which has started the resource_tracker. - if self._pid is not None: - os.waitpid(self._pid, 0) - except ChildProcessError: - # The resource_tracker has already been terminated. - pass - self._fd = None - self._pid = None - self._exitcode = None + self._write(to_send) + except OSError: + self._teardown_dead_process() + self._launch() - warnings.warn('resource_tracker: process died unexpectedly, ' - 'relaunching. Some resources might leak.') + msg = None # message was sent in probe + else: + self._launch() - fds_to_pass = [] - try: - fds_to_pass.append(sys.stderr.fileno()) - except Exception: - pass - cmd = 'from multiprocessing.resource_tracker import main;main(%d)' - r, w = os.pipe() + while True: try: - fds_to_pass.append(r) - # process will out live us, so no need to wait on pid - exe = spawn.get_executable() - args = [exe] + util._args_from_interpreter_flags() - args += ['-c', cmd % r] - # bpo-33613: Register a signal mask that will block the signals. - # This signal mask will be inherited by the child that is going - # to be spawned and will protect the child from a race condition - # that can make the child die before it registers signal handlers - # for SIGINT and SIGTERM. The mask is unregistered after spawning - # the child. - prev_sigmask = None - try: - if _HAVE_SIGMASK: - prev_sigmask = signal.pthread_sigmask(signal.SIG_BLOCK, _IGNORED_SIGNALS) - pid = util.spawnv_passfds(exe, args, fds_to_pass) - finally: - if prev_sigmask is not None: - signal.pthread_sigmask(signal.SIG_SETMASK, prev_sigmask) - except: - os.close(w) - raise - else: - self._fd = w - self._pid = pid - finally: - os.close(r) + reentrant_msg = self._reentrant_messages.popleft() + except IndexError: + break + self._write(reentrant_msg) + if msg is not None: + self._write(msg) def _check_alive(self): '''Check that the pipe has not been closed by sending a probe.''' try: # We cannot use send here as it calls ensure_running, creating # a cycle. - os.write(self._fd, b'PROBE:0:noop\n') + os.write(self._fd, self._make_probe_message()) except OSError: return False else: @@ -215,27 +275,42 @@ def unregister(self, name, rtype): '''Unregister name of resource with resource tracker.''' self._send('UNREGISTER', name, rtype) - def _send(self, cmd, name, rtype): - try: - self.ensure_running() - except ReentrantCallError: - # The code below might or might not work, depending on whether - # the resource tracker was already running and still alive. - # Better warn the user. - # (XXX is warnings.warn itself reentrant-safe? :-) - warnings.warn( - f"ResourceTracker called reentrantly for resource cleanup, " - f"which is unsupported. " - f"The {rtype} object {name!r} might leak.") - msg = '{0}:{1}:{2}\n'.format(cmd, name, rtype).encode('ascii') - if len(msg) > 512: - # posix guarantees that writes to a pipe of less than PIPE_BUF - # bytes are atomic, and that PIPE_BUF >= 512 - raise ValueError('msg too long') + def _write(self, msg): nbytes = os.write(self._fd, msg) - assert nbytes == len(msg), "nbytes {0:n} but len(msg) {1:n}".format( - nbytes, len(msg)) + assert nbytes == len(msg), f"{nbytes=} != {len(msg)=}" + def _send(self, cmd, name, rtype): + if self._use_simple_format and '\n' not in name: + msg = f"{cmd}:{name}:{rtype}\n".encode("ascii") + if len(msg) > 512: + # posix guarantees that writes to a pipe of less than PIPE_BUF + # bytes are atomic, and that PIPE_BUF >= 512 + raise ValueError('msg too long') + self._ensure_running_and_write(msg) + return + + # POSIX guarantees that writes to a pipe of less than PIPE_BUF (512 on Linux) + # bytes are atomic. Therefore, we want the message to be shorter than 512 bytes. + # POSIX shm_open() and sem_open() require the name, including its leading slash, + # to be at most NAME_MAX bytes (255 on Linux) + # With json.dump(..., ensure_ascii=True) every non-ASCII byte becomes a 6-char + # escape like \uDC80. + # As we want the overall message to be kept atomic and therefore smaller than 512, + # we encode encode the raw name bytes with URL-safe Base64 - so a 255 long name + # will not exceed 340 bytes. + b = name.encode('utf-8', 'surrogateescape') + if len(b) > 255: + raise ValueError('shared memory name too long (max 255 bytes)') + b64 = base64.urlsafe_b64encode(b).decode('ascii') + + payload = {"cmd": cmd, "rtype": rtype, "base64_name": b64} + msg = (json.dumps(payload, ensure_ascii=True, separators=(",", ":")) + "\n").encode("ascii") + + # The entire JSON message is guaranteed < PIPE_BUF (512 bytes) by construction. + assert len(msg) <= 512, f"internal error: message too long ({len(msg)} bytes)" + assert msg.startswith(b'{') + + self._ensure_running_and_write(msg) _resource_tracker = ResourceTracker() ensure_running = _resource_tracker.ensure_running @@ -244,6 +319,30 @@ def _send(self, cmd, name, rtype): getfd = _resource_tracker.getfd +def _decode_message(line): + if line.startswith(b'{'): + try: + obj = json.loads(line.decode('ascii')) + except Exception as e: + raise ValueError("malformed resource_tracker message: %r" % (line,)) from e + + cmd = obj["cmd"] + rtype = obj["rtype"] + b64 = obj.get("base64_name", "") + + if not isinstance(cmd, str) or not isinstance(rtype, str) or not isinstance(b64, str): + raise ValueError("malformed resource_tracker fields: %r" % (obj,)) + + try: + name = base64.urlsafe_b64decode(b64).decode('utf-8', 'surrogateescape') + except ValueError as e: + raise ValueError("malformed resource_tracker base64_name: %r" % (b64,)) from e + else: + cmd, rest = line.strip().decode('ascii').split(':', maxsplit=1) + name, rtype = rest.rsplit(':', maxsplit=1) + return cmd, rtype, name + + def main(fd): '''Run resource tracker.''' # protect the process from ^C and "killall python" etc @@ -266,7 +365,7 @@ def main(fd): with open(fd, 'rb') as f: for line in f: try: - cmd, name, rtype = line.strip().decode('ascii').split(':') + cmd, rtype, name = _decode_message(line) cleanup_func = _CLEANUP_FUNCS.get(rtype, None) if cleanup_func is None: raise ValueError( diff --git a/Lib/multiprocessing/util.py b/Lib/multiprocessing/util.py index 75dde02d88c..b8bfea045df 100644 --- a/Lib/multiprocessing/util.py +++ b/Lib/multiprocessing/util.py @@ -34,6 +34,7 @@ DEBUG = 10 INFO = 20 SUBWARNING = 25 +WARNING = 30 LOGGER_NAME = 'multiprocessing' DEFAULT_LOGGING_FORMAT = '[%(levelname)s/%(processName)s] %(message)s' @@ -53,6 +54,10 @@ def info(msg, *args): if _logger: _logger.log(INFO, msg, *args, stacklevel=2) +def _warn(msg, *args): + if _logger: + _logger.log(WARNING, msg, *args, stacklevel=2) + def sub_warning(msg, *args): if _logger: _logger.log(SUBWARNING, msg, *args, stacklevel=2) @@ -121,6 +126,23 @@ def is_abstract_socket_namespace(address): # Function returning a temp directory which will be removed on exit # +# Maximum length of a NULL-terminated [1] socket file path is usually +# between 92 and 108 [2], but Linux is known to use a size of 108 [3]. +# BSD-based systems usually use a size of 104 or 108 and Windows does +# not create AF_UNIX sockets. +# +# [1]: https://github.com/python/cpython/issues/140734 +# [2]: https://pubs.opengroup.org/onlinepubs/9799919799/basedefs/sys_un.h.html +# [3]: https://man7.org/linux/man-pages/man7/unix.7.html + +if sys.platform == 'linux': + _SUN_PATH_MAX = 108 +elif sys.platform.startswith(('openbsd', 'freebsd')): + _SUN_PATH_MAX = 104 +else: + # On Windows platforms, we do not create AF_UNIX sockets. + _SUN_PATH_MAX = None if os.name == 'nt' else 92 + def _remove_temp_dir(rmtree, tempdir): rmtree(tempdir) @@ -130,12 +152,69 @@ def _remove_temp_dir(rmtree, tempdir): if current_process is not None: current_process._config['tempdir'] = None +def _get_base_temp_dir(tempfile): + """Get a temporary directory where socket files will be created. + + To prevent additional imports, pass a pre-imported 'tempfile' module. + """ + if os.name == 'nt': + return None + # Most of the time, the default temporary directory is /tmp. Thus, + # listener sockets files "$TMPDIR/pymp-XXXXXXXX/sock-XXXXXXXX" do + # not have a path length exceeding SUN_PATH_MAX. + # + # If users specify their own temporary directory, we may be unable + # to create those files. Therefore, we fall back to the system-wide + # temporary directory /tmp, assumed to exist on POSIX systems. + # + # See https://github.com/python/cpython/issues/132124. + base_tempdir = tempfile.gettempdir() + # Files created in a temporary directory are suffixed by a string + # generated by tempfile._RandomNameSequence, which, by design, + # is 8 characters long. + # + # Thus, the socket file path length (without NULL terminator) will be: + # + # len(base_tempdir + '/pymp-XXXXXXXX' + '/sock-XXXXXXXX') + sun_path_len = len(base_tempdir) + 14 + 14 + # Strict inequality to account for the NULL terminator. + # See https://github.com/python/cpython/issues/140734. + if sun_path_len < _SUN_PATH_MAX: + return base_tempdir + # Fallback to the default system-wide temporary directory. + # This ignores user-defined environment variables. + # + # On POSIX systems, /tmp MUST be writable by any application [1]. + # We however emit a warning if this is not the case to prevent + # obscure errors later in the execution. + # + # On some legacy systems, /var/tmp and /usr/tmp can be present + # and will be used instead. + # + # [1]: https://refspecs.linuxfoundation.org/FHS_3.0/fhs/ch03s18.html + dirlist = ['/tmp', '/var/tmp', '/usr/tmp'] + try: + base_system_tempdir = tempfile._get_default_tempdir(dirlist) + except FileNotFoundError: + _warn("Process-wide temporary directory %s will not be usable for " + "creating socket files and no usable system-wide temporary " + "directory was found in %s", base_tempdir, dirlist) + # At this point, the system-wide temporary directory is not usable + # but we may assume that the user-defined one is, even if we will + # not be able to write socket files out there. + return base_tempdir + _warn("Ignoring user-defined temporary directory: %s", base_tempdir) + # at most max(map(len, dirlist)) + 14 + 14 = 36 characters + assert len(base_system_tempdir) + 14 + 14 < _SUN_PATH_MAX + return base_system_tempdir + def get_temp_dir(): # get name of a temp directory which will be automatically cleaned up tempdir = process.current_process()._config.get('tempdir') if tempdir is None: import shutil, tempfile - tempdir = tempfile.mkdtemp(prefix='pymp-') + base_tempdir = _get_base_temp_dir(tempfile) + tempdir = tempfile.mkdtemp(prefix='pymp-', dir=base_tempdir) info('created temp directory %s', tempdir) # keep a strong reference to shutil.rmtree(), since the finalizer # can be called late during Python shutdown