diff --git a/Doc/library/multiprocessing.rst b/Doc/library/multiprocessing.rst index 43fad57139057cd..3656561817e52d2 100644 --- a/Doc/library/multiprocessing.rst +++ b/Doc/library/multiprocessing.rst @@ -541,7 +541,7 @@ or creating these objects. ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ .. class:: Process(group=None, target=None, name=None, args=(), kwargs={}, \ - *, daemon=None) + *, daemon=None, env=None) Process objects represent activity that is run in a separate process. The :class:`Process` class has equivalents of all the methods of @@ -558,6 +558,26 @@ or creating these objects. to ``True`` or ``False``. If ``None`` (the default), this flag will be inherited from the creating process. + *env* defaults to ``None``, which inherits the parent's environment. + Otherwise, it is a mapping of strings to strings or bytes to bytes + that replaces the environment used to start the child interpreter. + The mapping is copied during construction; do not modify it concurrently + with construction. A non-``None`` value requires POSIX ``spawn``; + other platforms and start methods raise :exc:`ValueError` at :meth:`start`. + It does not override the inherited :data:`sys.path` or interpreter flags. + + On POSIX, a resource tracker launched by this :meth:`start` call also + uses an explicit *env*. A running tracker is reused without changing + its environment. If no explicit environment is supplied for a tracker + launch or relaunch, it uses the launching process's own explicit + environment snapshot, if any; otherwise, it inherits that process's + current environment. It does not retain an environment from an earlier + tracker launch. Processes created with ``env=None`` have no explicit + snapshot. + + .. versionchanged:: next + Added the *env* parameter. + By default, no arguments are passed to *target*. The *args* argument, which defaults to ``()``, can be used to specify a list or tuple of the arguments to pass to *target*. diff --git a/Lib/multiprocessing/popen_spawn_posix.py b/Lib/multiprocessing/popen_spawn_posix.py index cccd659ae776377..313cee358e9b900 100644 --- a/Lib/multiprocessing/popen_spawn_posix.py +++ b/Lib/multiprocessing/popen_spawn_posix.py @@ -37,9 +37,10 @@ def duplicate_for_child(self, fd): def _launch(self, process_obj): from . import resource_tracker - tracker_fd = resource_tracker.getfd() + tracker_fd = resource_tracker.getfd(env=process_obj._env) self._fds.append(tracker_fd) prep_data = spawn.get_preparation_data(process_obj._name) + prep_data['process_env'] = process_obj._env fp = io.BytesIO() set_spawning_popen(self) try: @@ -56,7 +57,7 @@ def _launch(self, process_obj): pipe_handle=child_r) self._fds.extend([child_r, child_w]) self.pid = util.spawnv_passfds(spawn.get_executable(), - cmd, self._fds) + cmd, self._fds, env=process_obj._env) os.close(child_r) child_r = None os.close(child_w) diff --git a/Lib/multiprocessing/process.py b/Lib/multiprocessing/process.py index 262513f295fde56..b0806b655149bd2 100644 --- a/Lib/multiprocessing/process.py +++ b/Lib/multiprocessing/process.py @@ -78,8 +78,14 @@ def _Popen(self): raise NotImplementedError def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, - *, daemon=None): + *, daemon=None, env=None): assert group is None, 'group argument must be None for now' + # Snapshot an explicit environment at construction time. + if env is None: + self._env = None + else: + from .util import _encode_spawn_env + self._env = _encode_spawn_env(env) count = next(_process_counter) self._identity = _current_process._identity + (count,) self._config = _current_process._config.copy() @@ -117,6 +123,11 @@ def start(self): 'can only start a process object created by current process' assert not _current_process._config.get('daemon'), \ 'daemonic processes are not allowed to have children' + if self._env is not None: + from . import get_start_method + method = getattr(self, '_start_method', None) or get_start_method() + if os.name != 'posix' or method != 'spawn': + raise ValueError('env is only supported with POSIX spawn') _cleanup() self._popen = self._Popen(self) self._sentinel = self._popen.sentinel diff --git a/Lib/multiprocessing/resource_tracker.py b/Lib/multiprocessing/resource_tracker.py index d3328a8c6170a66..a9596d6bdc2ad9f 100644 --- a/Lib/multiprocessing/resource_tracker.py +++ b/Lib/multiprocessing/resource_tracker.py @@ -26,6 +26,7 @@ import json +from . import process from . import spawn from . import util @@ -203,16 +204,16 @@ def _stop_locked( # os.waitstatus_to_exitcode may raise an exception for invalid values self._exitcode = None - def getfd(self): - self.ensure_running() + def getfd(self, *, env=None): + self.ensure_running(env=env) return self._fd - def ensure_running(self): + def ensure_running(self, *, env=None): '''Make sure that resource tracker process is running. 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() + return self._ensure_running_and_write(env=env) def _teardown_dead_process(self): os.close(self._fd) @@ -233,7 +234,11 @@ def _teardown_dead_process(self): warnings.warn('resource_tracker: process died unexpectedly, ' 'relaunching. Some resources might leak.') - def _launch(self): + def _launch(self, env=None): + # Prefer this launch's explicit environment, then the launching + # process's snapshot. None preserves normal environment inheritance. + if env is None: + env = getattr(process.current_process(), '_env', None) fds_to_pass = [] try: fds_to_pass.append(sys.stderr.fileno()) @@ -260,7 +265,8 @@ def _launch(self): try: if _HAVE_SIGMASK: prev_sigmask = signal.pthread_sigmask(signal.SIG_BLOCK, _IGNORED_SIGNALS) - pid = util.spawnv_passfds(exe, args, fds_to_pass) + pid = util.spawnv_passfds(exe, args, fds_to_pass, + env=env) finally: if prev_sigmask is not None: signal.pthread_sigmask(signal.SIG_SETMASK, prev_sigmask) @@ -286,7 +292,7 @@ def _make_probe_message(self): + "\n" ).encode("ascii") - def _ensure_running_and_write(self, msg=None): + def _ensure_running_and_write(self, msg=None, *, env=None): with self._lock: if self._lock._recursion_count() > 1: # The code below is certainly not reentrant-safe, so bail out @@ -304,11 +310,11 @@ def _ensure_running_and_write(self, msg=None): self._write(to_send) except OSError: self._teardown_dead_process() - self._launch() + self._launch(env) msg = None # message was sent in probe else: - self._launch() + self._launch(env) while True: try: diff --git a/Lib/multiprocessing/spawn.py b/Lib/multiprocessing/spawn.py index d43864c939cb63f..db8f09091af3c02 100644 --- a/Lib/multiprocessing/spawn.py +++ b/Lib/multiprocessing/spawn.py @@ -213,6 +213,11 @@ def prepare(data): ''' Try to get current process ready to unpickle process object ''' + if 'process_env' in data: + # Resource tracking can start while importing the main module or + # unpickling the process, before _bootstrap installs current_process. + process.current_process()._env = data['process_env'] + if 'name' in data: process.current_process().name = data['name'] diff --git a/Lib/multiprocessing/util.py b/Lib/multiprocessing/util.py index cf7e0b2990598b7..6709efec79004de 100644 --- a/Lib/multiprocessing/util.py +++ b/Lib/multiprocessing/util.py @@ -540,13 +540,28 @@ def _flush_std_streams(): # Start a program with only specified fds kept open # -def spawnv_passfds(path, args, passfds): +def _encode_spawn_env(env): + """Copy a mapping to immutable exec environment entries. + + The caller must not mutate the mapping while this copy is made. + """ + entries = [] + for key, value in env.items(): + key = os.fsencode(key) + value = os.fsencode(value) + if not key or b'=' in key or b'\0' in key or b'\0' in value: + raise ValueError('illegal environment variable name or value') + entries.append(key + b'=' + value) + return tuple(entries) + + +def spawnv_passfds(path, args, passfds, env=None): import _posixsubprocess passfds = tuple(sorted(map(int, passfds))) errpipe_read, errpipe_write = os.pipe() try: return _posixsubprocess.fork_exec( - args, [path], True, passfds, None, None, + args, [path], True, passfds, None, env, -1, -1, -1, -1, -1, -1, errpipe_read, errpipe_write, False, False, -1, None, None, None, -1, None) finally: diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index 46ed8843fcd0519..5ca71a06cf45634 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -362,6 +362,243 @@ def __call__(self, q, c): q.put(5) +@unittest.skipUnless(os.name == 'posix', 'env requires POSIX spawn') +class _TestProcessEnvironment(BaseTestCase): + + ALLOWED_TYPES = ('processes',) + START_METHODS = {'spawn'} + ENV_KEY = 'MP_TEST_CHILD_ENV' + + @staticmethod + def report_environment(conn): + with conn: + conn.send(dict(os.environ)) + + def make_process(self, env): + reader, writer = self.Pipe(duplex=False) + self.addCleanup(reader.close) + self.addCleanup(writer.close) + process = self.Process(target=self.report_environment, + args=(writer,), env=env) + self.addCleanup(self.cleanup_process, process) + return process, reader + + @staticmethod + def cleanup_process(process): + if process.pid is not None: + if process.is_alive(): + process.kill() + process.join() + process.close() + + def collect(self, process, reader): + process.start() + self.assertTrue(reader.poll(support.SHORT_TIMEOUT)) + result = reader.recv() + join_process(process) + self.assertEqual(process.exitcode, 0) + return result + + def test_environment_snapshot(self): + with os_helper.EnvironmentVarGuard() as parent_env: + parent_env[self.ENV_KEY] = 'parent' + env = dict(os.environ, **{self.ENV_KEY: 'child'}) + process, reader = self.make_process(env) + env[self.ENV_KEY] = 'changed mapping' + parent_env[self.ENV_KEY] = 'changed parent' + actual = self.collect(process, reader) + self.assertEqual(actual[self.ENV_KEY], 'child') + self.assertEqual(os.environ[self.ENV_KEY], 'changed parent') + + def test_none_inherits_environment(self): + with os_helper.EnvironmentVarGuard() as env: + env[self.ENV_KEY] = 'before construction' + process, reader = self.make_process(None) + env[self.ENV_KEY] = 'at start' + actual = self.collect(process, reader) + self.assertEqual(actual[self.ENV_KEY], 'at start') + + def test_environment_replacement(self): + env = {self.ENV_KEY: 'child'} + with os_helper.EnvironmentVarGuard() as parent_env: + parent_env['MP_TEST_PARENT_ONLY'] = 'parent' + actual = self.collect(*self.make_process(env)) + self.assertEqual(actual[self.ENV_KEY], 'child') + self.assertNotIn('MP_TEST_PARENT_ONLY', actual) + + def test_empty_environment(self): + with os_helper.EnvironmentVarGuard() as env: + env[self.ENV_KEY] = 'parent' + actual = self.collect(*self.make_process({})) + self.assertNotIn(self.ENV_KEY, actual) + + def test_unicode_environment(self): + env = dict(os.environ, **{self.ENV_KEY: '中文=value'}) + actual = self.collect(*self.make_process(env)) + self.assertEqual(actual[self.ENV_KEY], env[self.ENV_KEY]) + + def test_bytes_environment(self): + env = {os.fsencode(k): os.fsencode(v) for k, v in os.environ.items()} + env[os.fsencode(self.ENV_KEY)] = b'child=value' + actual = self.collect(*self.make_process(env)) + self.assertEqual(actual[self.ENV_KEY], 'child=value') + + def test_invalid_environment(self): + for env in ({'': 'x'}, {'a=b': 'c'}, {'a\0': 'b'}, {'a': 'b\0'}): + with self.subTest(env=env), self.assertRaises(ValueError): + self.Process(env=env) + for env in ({'a': 1}, {1: 'a'}): + with self.subTest(env=env), self.assertRaises(TypeError): + self.Process(env=env) + + def test_other_start_methods_rejected(self): + for method in multiprocessing.get_all_start_methods(): + if method == 'spawn': + continue + with self.subTest(method=method): + process = multiprocessing.get_context(method).Process(env={}) + try: + with self.assertRaisesRegex(ValueError, 'POSIX spawn'): + process.start() + self.assertIsNone(process.pid) + finally: + process.close() + + @unittest.skipUnless(sys.platform == 'linux', 'requires /proc/PID/environ') + def test_resource_tracker_environment(self): + # Use a fresh interpreter to control the shared tracker's first launch. + code = '''if True: + import multiprocessing as mp + from multiprocessing import resource_tracker as rt + import os, pathlib, signal, warnings + from test import support + + key = 'MP_TEST_CHILD_ENV' + + def start(value): + env = None if value is None else dict(os.environ, **{key: value}) + p = mp.get_context('spawn').Process(env=env) + try: + p.start() + p.join(support.SHORT_TIMEOUT) + assert p.exitcode == 0, p.exitcode + finally: + if p.pid is not None: + if p.is_alive(): + p.kill() + p.join() + p.close() + + def check(value): + pid = rt._resource_tracker._pid + data = pathlib.Path(f'/proc/{pid}/environ').read_bytes() + assert (key + '=' + value).encode() in data.split(b'\\0') + + def kill_tracker(): + pid = rt._resource_tracker._pid + os.kill(pid, signal.SIGKILL) + os.waitpid(pid, 0) + + try: + os.environ[key] = 'parent' + start('first') + check('first') + pid = rt._resource_tracker._pid + start('second') + assert rt._resource_tracker._pid == pid + check('first') + kill_tracker() + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + rt.ensure_running() + assert caught + check('parent') + kill_tracker() + with warnings.catch_warnings(): + warnings.simplefilter('ignore', UserWarning) + start('third') + check('third') + kill_tracker() + with warnings.catch_warnings(): + warnings.simplefilter('ignore', UserWarning) + start(None) + check('parent') + finally: + rt._resource_tracker._stop() + ''' + assert_python_ok('-c', code) + + @staticmethod + def relaunch_tracker(conn): + from multiprocessing import resource_tracker + + tracker = resource_tracker._resource_tracker + try: + conn.send('ready') + if not conn.poll(support.SHORT_TIMEOUT): + raise AssertionError('parent did not request tracker relaunch') + conn.recv() + os.environ[_TestProcessEnvironment.ENV_KEY] = 'changed in worker' + with warnings.catch_warnings(): + warnings.simplefilter('ignore', UserWarning) + resource_tracker.ensure_running() + data = pathlib.Path(f'/proc/{tracker._pid}/environ').read_bytes() + conn.send(data.split(b'\0')) + finally: + tracker._stop() + conn.close() + + @unittest.skipUnless(sys.platform == 'linux', 'requires /proc/PID/environ') + def test_resource_tracker_relaunch_in_worker(self): + # Isolate tracker termination from the test runner's shared tracker. + code = '''if True: + import multiprocessing as mp + from multiprocessing import resource_tracker as rt, util + import os, signal + from test import support + from test._test_multiprocessing import _TestProcessEnvironment + + key = _TestProcessEnvironment.ENV_KEY + ctx = mp.get_context('spawn') + for env, expected in (({key: 'worker'}, 'worker'), + ({}, None), + (None, 'changed in worker')): + rt.ensure_running(env=util._encode_spawn_env({key: 'tracker'})) + tracker_pid = rt._resource_tracker._pid + reader, writer = ctx.Pipe() + p = ctx.Process(target=_TestProcessEnvironment.relaunch_tracker, + args=(writer,), env=env) + try: + p.start() + writer.close() + assert reader.poll(support.SHORT_TIMEOUT) + assert reader.recv() == 'ready' + assert rt._resource_tracker._pid == tracker_pid + os.kill(tracker_pid, signal.SIGKILL) + os.waitpid(tracker_pid, 0) + reader.send('relaunch') + assert reader.poll(support.SHORT_TIMEOUT) + entries = reader.recv() + values = [e for e in entries if e.startswith(key.encode() + b'=')] + if expected is None: + assert not values, values + else: + assert values == [(key + '=' + expected).encode()], values + p.join(support.SHORT_TIMEOUT) + assert p.exitcode == 0, p.exitcode + finally: + if p.pid is not None: + if p.is_alive(): + p.kill() + p.join() + p.close() + reader.close() + writer.close() + rt._resource_tracker._stop() + ''' + assert_python_ok('-c', code) + + class _TestProcess(BaseTestCase): ALLOWED_TYPES = ('processes', 'threads') diff --git a/Misc/NEWS.d/next/Library/2026-09-17-00-00-00.gh-issue-157613.fR7x2a.rst b/Misc/NEWS.d/next/Library/2026-09-17-00-00-00.gh-issue-157613.fR7x2a.rst new file mode 100644 index 000000000000000..6353391fcbdd988 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-09-17-00-00-00.gh-issue-157613.fR7x2a.rst @@ -0,0 +1,3 @@ +Add an optional *env* mapping to :class:`multiprocessing.Process` for the +POSIX ``spawn`` start method, allowing callers to specify the child interpreter's +environment without modifying the parent process's environment.