Mercurial > repos > rliterman > csp2

comparison CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/multiprocessing/resource_tracker.py @ 68:5028fdace37b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit 2e9511a184a1ca667c7be0c6321a36dc4e3d116d
author jpayne
date Tue, 18 Mar 2025 16:23:26 -0400
parents
children
comparison
equal deleted inserted replaced
67:0e9998148a16 68:5028fdace37b
1 ###############################################################################
2 # Server process to keep track of unlinked resources (like shared memory
3 # segments, semaphores etc.) and clean them.
4 #
5 # On Unix we run a server process which keeps track of unlinked
6 # resources. The server ignores SIGINT and SIGTERM and reads from a
7 # pipe. Every other process of the program has a copy of the writable
8 # end of the pipe, so we get EOF when all other processes have exited.
9 # Then the server process unlinks any remaining resource names.
10 #
11 # This is important because there may be system limits for such resources: for
12 # instance, the system only supports a limited number of named semaphores, and
13 # shared-memory segments live in the RAM. If a python process leaks such a
14 # resource, this resource will not be removed till the next reboot. Without
15 # this resource tracker process, "killall python" would probably leave unlinked
16 # resources.
17
18 import os
19 import signal
20 import sys
21 import threading
22 import warnings
23
24 from . import spawn
25 from . import util
26
27 __all__ = ['ensure_running', 'register', 'unregister']
28
29 _HAVE_SIGMASK = hasattr(signal, 'pthread_sigmask')
30 _IGNORED_SIGNALS = (signal.SIGINT, signal.SIGTERM)
31
32 _CLEANUP_FUNCS = {
33 'noop': lambda: None,
34 }
35
36 if os.name == 'posix':
37 import _multiprocessing
38 import _posixshmem
39
40 _CLEANUP_FUNCS.update({
41 'semaphore': _multiprocessing.sem_unlink,
42 'shared_memory': _posixshmem.shm_unlink,
43 })
44
45
46 class ResourceTracker(object):
47
48 def __init__(self):
49 self._lock = threading.Lock()
50 self._fd = None
51 self._pid = None
52
53 def _stop(self):
54 with self._lock:
55 if self._fd is None:
56 # not running
57 return
58
59 # closing the "alive" file descriptor stops main()
60 os.close(self._fd)
61 self._fd = None
62
63 os.waitpid(self._pid, 0)
64 self._pid = None
65
66 def getfd(self):
67 self.ensure_running()
68 return self._fd
69
70 def ensure_running(self):
71 '''Make sure that resource tracker process is running.
72
73 This can be run from any process. Usually a child process will use
74 the resource created by its parent.'''
75 with self._lock:
76 if self._fd is not None:
77 # resource tracker was launched before, is it still running?
78 if self._check_alive():
79 # => still alive
80 return
81 # => dead, launch it again
82 os.close(self._fd)
83
84 # Clean-up to avoid dangling processes.
85 try:
86 # _pid can be None if this process is a child from another
87 # python process, which has started the resource_tracker.
88 if self._pid is not None:
89 os.waitpid(self._pid, 0)
90 except ChildProcessError:
91 # The resource_tracker has already been terminated.
92 pass
93 self._fd = None
94 self._pid = None
95
96 warnings.warn('resource_tracker: process died unexpectedly, '
97 'relaunching. Some resources might leak.')
98
99 fds_to_pass = []
100 try:
101 fds_to_pass.append(sys.stderr.fileno())
102 except Exception:
103 pass
104 cmd = 'from multiprocessing.resource_tracker import main;main(%d)'
105 r, w = os.pipe()
106 try:
107 fds_to_pass.append(r)
108 # process will out live us, so no need to wait on pid
109 exe = spawn.get_executable()
110 args = [exe] + util._args_from_interpreter_flags()
111 args += ['-c', cmd % r]
112 # bpo-33613: Register a signal mask that will block the signals.
113 # This signal mask will be inherited by the child that is going
114 # to be spawned and will protect the child from a race condition
115 # that can make the child die before it registers signal handlers
116 # for SIGINT and SIGTERM. The mask is unregistered after spawning
117 # the child.
118 try:
119 if _HAVE_SIGMASK:
120 signal.pthread_sigmask(signal.SIG_BLOCK, _IGNORED_SIGNALS)
121 pid = util.spawnv_passfds(exe, args, fds_to_pass)
122 finally:
123 if _HAVE_SIGMASK:
124 signal.pthread_sigmask(signal.SIG_UNBLOCK, _IGNORED_SIGNALS)
125 except:
126 os.close(w)
127 raise
128 else:
129 self._fd = w
130 self._pid = pid
131 finally:
132 os.close(r)
133
134 def _check_alive(self):
135 '''Check that the pipe has not been closed by sending a probe.'''
136 try:
137 # We cannot use send here as it calls ensure_running, creating
138 # a cycle.
139 os.write(self._fd, b'PROBE:0:noop\n')
140 except OSError:
141 return False
142 else:
143 return True
144
145 def register(self, name, rtype):
146 '''Register name of resource with resource tracker.'''
147 self._send('REGISTER', name, rtype)
148
149 def unregister(self, name, rtype):
150 '''Unregister name of resource with resource tracker.'''
151 self._send('UNREGISTER', name, rtype)
152
153 def _send(self, cmd, name, rtype):
154 self.ensure_running()
155 msg = '{0}:{1}:{2}\n'.format(cmd, name, rtype).encode('ascii')
156 if len(name) > 512:
157 # posix guarantees that writes to a pipe of less than PIPE_BUF
158 # bytes are atomic, and that PIPE_BUF >= 512
159 raise ValueError('name too long')
160 nbytes = os.write(self._fd, msg)
161 assert nbytes == len(msg), "nbytes {0:n} but len(msg) {1:n}".format(
162 nbytes, len(msg))
163
164
165 _resource_tracker = ResourceTracker()
166 ensure_running = _resource_tracker.ensure_running
167 register = _resource_tracker.register
168 unregister = _resource_tracker.unregister
169 getfd = _resource_tracker.getfd
170
171 def main(fd):
172 '''Run resource tracker.'''
173 # protect the process from ^C and "killall python" etc
174 signal.signal(signal.SIGINT, signal.SIG_IGN)
175 signal.signal(signal.SIGTERM, signal.SIG_IGN)
176 if _HAVE_SIGMASK:
177 signal.pthread_sigmask(signal.SIG_UNBLOCK, _IGNORED_SIGNALS)
178
179 for f in (sys.stdin, sys.stdout):
180 try:
181 f.close()
182 except Exception:
183 pass
184
185 cache = {rtype: set() for rtype in _CLEANUP_FUNCS.keys()}
186 try:
187 # keep track of registered/unregistered resources
188 with open(fd, 'rb') as f:
189 for line in f:
190 try:
191 cmd, name, rtype = line.strip().decode('ascii').split(':')
192 cleanup_func = _CLEANUP_FUNCS.get(rtype, None)
193 if cleanup_func is None:
194 raise ValueError(
195 f'Cannot register {name} for automatic cleanup: '
196 f'unknown resource type {rtype}')
197
198 if cmd == 'REGISTER':
199 cache[rtype].add(name)
200 elif cmd == 'UNREGISTER':
201 cache[rtype].remove(name)
202 elif cmd == 'PROBE':
203 pass
204 else:
205 raise RuntimeError('unrecognized command %r' % cmd)
206 except Exception:
207 try:
208 sys.excepthook(*sys.exc_info())
209 except:
210 pass
211 finally:
212 # all processes have terminated; cleanup any remaining resources
213 for rtype, rtype_cache in cache.items():
214 if rtype_cache:
215 try:
216 warnings.warn('resource_tracker: There appear to be %d '
217 'leaked %s objects to clean up at shutdown' %
218 (len(rtype_cache), rtype))
219 except Exception:
220 pass
221 for name in rtype_cache:
222 # For some reason the process which created and registered this
223 # resource has failed to unregister it. Presumably it has
224 # died. We therefore unlink it.
225 try:
226 try:
227 _CLEANUP_FUNCS[rtype](name)
228 except Exception as e:
229 warnings.warn('resource_tracker: %r: %s' % (name, e))
230 finally:
231 pass