ยปCore Development>Code coverage>Lib/multiprocessing/popen_spawn_posix.py

Python code coverage for Lib/multiprocessing/popen_spawn_posix.py

#countcontent
1n/aimport io
2n/aimport os
3n/a
4n/afrom .context import reduction, set_spawning_popen
5n/afrom . import popen_fork
6n/afrom . import spawn
7n/afrom . import util
8n/a
9n/a__all__ = ['Popen']
10n/a
11n/a
12n/a#
13n/a# Wrapper for an fd used while launching a process
14n/a#
15n/a
16n/aclass _DupFd(object):
17n/a def __init__(self, fd):
18n/a self.fd = fd
19n/a def detach(self):
20n/a return self.fd
21n/a
22n/a#
23n/a# Start child process using a fresh interpreter
24n/a#
25n/a
26n/aclass Popen(popen_fork.Popen):
27n/a method = 'spawn'
28n/a DupFd = _DupFd
29n/a
30n/a def __init__(self, process_obj):
31n/a self._fds = []
32n/a super().__init__(process_obj)
33n/a
34n/a def duplicate_for_child(self, fd):
35n/a self._fds.append(fd)
36n/a return fd
37n/a
38n/a def _launch(self, process_obj):
39n/a from . import semaphore_tracker
40n/a tracker_fd = semaphore_tracker.getfd()
41n/a self._fds.append(tracker_fd)
42n/a prep_data = spawn.get_preparation_data(process_obj._name)
43n/a fp = io.BytesIO()
44n/a set_spawning_popen(self)
45n/a try:
46n/a reduction.dump(prep_data, fp)
47n/a reduction.dump(process_obj, fp)
48n/a finally:
49n/a set_spawning_popen(None)
50n/a
51n/a parent_r = child_w = child_r = parent_w = None
52n/a try:
53n/a parent_r, child_w = os.pipe()
54n/a child_r, parent_w = os.pipe()
55n/a cmd = spawn.get_command_line(tracker_fd=tracker_fd,
56n/a pipe_handle=child_r)
57n/a self._fds.extend([child_r, child_w])
58n/a self.pid = util.spawnv_passfds(spawn.get_executable(),
59n/a cmd, self._fds)
60n/a self.sentinel = parent_r
61n/a with open(parent_w, 'wb', closefd=False) as f:
62n/a f.write(fp.getbuffer())
63n/a finally:
64n/a if parent_r is not None:
65n/a util.Finalize(self, os.close, (parent_r,))
66n/a for fd in (child_r, child_w, parent_w):
67n/a if fd is not None:
68n/a os.close(fd)