process.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404
  1. #
  2. # Module providing the `Process` class which emulates `threading.Thread`
  3. #
  4. # multiprocessing/process.py
  5. #
  6. # Copyright (c) 2006-2008, R Oudkerk
  7. # Licensed to PSF under a Contributor Agreement.
  8. #
  9. from __future__ import absolute_import
  10. #
  11. # Imports
  12. #
  13. import os
  14. import sys
  15. import signal
  16. import itertools
  17. import logging
  18. import threading
  19. from _weakrefset import WeakSet
  20. from multiprocessing import process as _mproc
  21. from .five import items, string_t
  22. try:
  23. ORIGINAL_DIR = os.path.abspath(os.getcwd())
  24. except OSError:
  25. ORIGINAL_DIR = None
  26. __all__ = ['BaseProcess', 'Process', 'current_process', 'active_children']
  27. #
  28. # Public functions
  29. #
  30. def current_process():
  31. '''
  32. Return process object representing the current process
  33. '''
  34. return _current_process
  35. def _set_current_process(process):
  36. global _current_process
  37. _current_process = _mproc._current_process = process
  38. def _cleanup():
  39. # check for processes which have finished
  40. for p in list(_children):
  41. if p._popen.poll() is not None:
  42. _children.discard(p)
  43. def _maybe_flush(f):
  44. try:
  45. f.flush()
  46. except (AttributeError, EnvironmentError, NotImplementedError):
  47. pass
  48. def active_children(_cleanup=_cleanup):
  49. '''
  50. Return list of process objects corresponding to live child processes
  51. '''
  52. try:
  53. _cleanup()
  54. except TypeError:
  55. # called after gc collect so _cleanup does not exist anymore
  56. return []
  57. return list(_children)
  58. class BaseProcess(object):
  59. '''
  60. Process objects represent activity that is run in a separate process
  61. The class is analagous to `threading.Thread`
  62. '''
  63. def _Popen(self):
  64. raise NotImplementedError()
  65. def __init__(self, group=None, target=None, name=None,
  66. args=(), kwargs={}, daemon=None, **_kw):
  67. assert group is None, 'group argument must be None for now'
  68. count = next(_process_counter)
  69. self._identity = _current_process._identity + (count, )
  70. self._config = _current_process._config.copy()
  71. self._parent_pid = os.getpid()
  72. self._popen = None
  73. self._target = target
  74. self._args = tuple(args)
  75. self._kwargs = dict(kwargs)
  76. self._name = (
  77. name or type(self).__name__ + '-' +
  78. ':'.join(str(i) for i in self._identity)
  79. )
  80. if daemon is not None:
  81. self.daemon = daemon
  82. if _dangling is not None:
  83. _dangling.add(self)
  84. self._controlled_termination = False
  85. def run(self):
  86. '''
  87. Method to be run in sub-process; can be overridden in sub-class
  88. '''
  89. if self._target:
  90. self._target(*self._args, **self._kwargs)
  91. def start(self):
  92. '''
  93. Start child process
  94. '''
  95. assert self._popen is None, 'cannot start a process twice'
  96. assert self._parent_pid == os.getpid(), \
  97. 'can only start a process object created by current process'
  98. _cleanup()
  99. self._popen = self._Popen(self)
  100. self._sentinel = self._popen.sentinel
  101. _children.add(self)
  102. def close(self):
  103. if self._popen is not None:
  104. self._popen.close()
  105. def terminate(self):
  106. '''
  107. Terminate process; sends SIGTERM signal or uses TerminateProcess()
  108. '''
  109. self._popen.terminate()
  110. def terminate_controlled(self):
  111. self._controlled_termination = True
  112. self.terminate()
  113. def join(self, timeout=None):
  114. '''
  115. Wait until child process terminates
  116. '''
  117. assert self._parent_pid == os.getpid(), 'can only join a child process'
  118. assert self._popen is not None, 'can only join a started process'
  119. res = self._popen.wait(timeout)
  120. if res is not None:
  121. _children.discard(self)
  122. self.close()
  123. def is_alive(self):
  124. '''
  125. Return whether process is alive
  126. '''
  127. if self is _current_process:
  128. return True
  129. assert self._parent_pid == os.getpid(), 'can only test a child process'
  130. if self._popen is None:
  131. return False
  132. self._popen.poll()
  133. return self._popen.returncode is None
  134. def _is_alive(self):
  135. if self._popen is None:
  136. return False
  137. return self._popen.poll() is None
  138. @property
  139. def name(self):
  140. return self._name
  141. @name.setter
  142. def name(self, name): # noqa
  143. assert isinstance(name, string_t), 'name must be a string'
  144. self._name = name
  145. @property
  146. def daemon(self):
  147. '''
  148. Return whether process is a daemon
  149. '''
  150. return self._config.get('daemon', False)
  151. @daemon.setter # noqa
  152. def daemon(self, daemonic):
  153. '''
  154. Set whether process is a daemon
  155. '''
  156. assert self._popen is None, 'process has already started'
  157. self._config['daemon'] = daemonic
  158. @property
  159. def authkey(self):
  160. return self._config['authkey']
  161. @authkey.setter # noqa
  162. def authkey(self, authkey):
  163. '''
  164. Set authorization key of process
  165. '''
  166. self._config['authkey'] = AuthenticationString(authkey)
  167. @property
  168. def exitcode(self):
  169. '''
  170. Return exit code of process or `None` if it has yet to stop
  171. '''
  172. if self._popen is None:
  173. return self._popen
  174. return self._popen.poll()
  175. @property
  176. def ident(self):
  177. '''
  178. Return identifier (PID) of process or `None` if it has yet to start
  179. '''
  180. if self is _current_process:
  181. return os.getpid()
  182. else:
  183. return self._popen and self._popen.pid
  184. pid = ident
  185. @property
  186. def sentinel(self):
  187. '''
  188. Return a file descriptor (Unix) or handle (Windows) suitable for
  189. waiting for process termination.
  190. '''
  191. try:
  192. return self._sentinel
  193. except AttributeError:
  194. raise ValueError("process not started")
  195. @property
  196. def _counter(self):
  197. # compat for 2.7
  198. return _process_counter
  199. @property
  200. def _children(self):
  201. # compat for 2.7
  202. return _children
  203. @property
  204. def _authkey(self):
  205. # compat for 2.7
  206. return self.authkey
  207. @property
  208. def _daemonic(self):
  209. # compat for 2.7
  210. return self.daemon
  211. @property
  212. def _tempdir(self):
  213. # compat for 2.7
  214. return self._config.get('tempdir')
  215. def __repr__(self):
  216. if self is _current_process:
  217. status = 'started'
  218. elif self._parent_pid != os.getpid():
  219. status = 'unknown'
  220. elif self._popen is None:
  221. status = 'initial'
  222. else:
  223. if self._popen.poll() is not None:
  224. status = self.exitcode
  225. else:
  226. status = 'started'
  227. if type(status) is int:
  228. if status == 0:
  229. status = 'stopped'
  230. else:
  231. status = 'stopped[%s]' % _exitcode_to_name.get(status, status)
  232. return '<%s(%s, %s%s)>' % (type(self).__name__, self._name,
  233. status, self.daemon and ' daemon' or '')
  234. ##
  235. def _bootstrap(self):
  236. from . import util, context
  237. global _current_process, _process_counter, _children
  238. try:
  239. if self._start_method is not None:
  240. context._force_start_method(self._start_method)
  241. _process_counter = itertools.count(1)
  242. _children = set()
  243. if sys.stdin is not None:
  244. try:
  245. sys.stdin.close()
  246. sys.stdin = open(os.devnull)
  247. except (OSError, ValueError):
  248. pass
  249. old_process = _current_process
  250. _set_current_process(self)
  251. # Re-init logging system.
  252. # Workaround for https://bugs.python.org/issue6721/#msg140215
  253. # Python logging module uses RLock() objects which are broken
  254. # after fork. This can result in a deadlock (Celery Issue #496).
  255. loggerDict = logging.Logger.manager.loggerDict
  256. logger_names = list(loggerDict.keys())
  257. logger_names.append(None) # for root logger
  258. for name in logger_names:
  259. if not name or not isinstance(loggerDict[name],
  260. logging.PlaceHolder):
  261. for handler in logging.getLogger(name).handlers:
  262. handler.createLock()
  263. logging._lock = threading.RLock()
  264. try:
  265. util._finalizer_registry.clear()
  266. util._run_after_forkers()
  267. finally:
  268. # delay finalization of the old process object until after
  269. # _run_after_forkers() is executed
  270. del old_process
  271. util.info('child process %s calling self.run()', self.pid)
  272. try:
  273. self.run()
  274. exitcode = 0
  275. finally:
  276. util._exit_function()
  277. except SystemExit as exc:
  278. if not exc.args:
  279. exitcode = 1
  280. elif isinstance(exc.args[0], int):
  281. exitcode = exc.args[0]
  282. else:
  283. sys.stderr.write(str(exc.args[0]) + '\n')
  284. _maybe_flush(sys.stderr)
  285. exitcode = 0 if isinstance(exc.args[0], str) else 1
  286. except:
  287. exitcode = 1
  288. if not util.error('Process %s', self.name, exc_info=True):
  289. import traceback
  290. sys.stderr.write('Process %s:\n' % self.name)
  291. traceback.print_exc()
  292. finally:
  293. util.info('process %s exiting with exitcode %d',
  294. self.pid, exitcode)
  295. _maybe_flush(sys.stdout)
  296. _maybe_flush(sys.stderr)
  297. return exitcode
  298. #
  299. # We subclass bytes to avoid accidental transmission of auth keys over network
  300. #
  301. class AuthenticationString(bytes):
  302. def __reduce__(self):
  303. from .context import get_spawning_popen
  304. if get_spawning_popen() is None:
  305. raise TypeError(
  306. 'Pickling an AuthenticationString object is '
  307. 'disallowed for security reasons')
  308. return AuthenticationString, (bytes(self),)
  309. #
  310. # Create object representing the main process
  311. #
  312. class _MainProcess(BaseProcess):
  313. def __init__(self):
  314. self._identity = ()
  315. self._name = 'MainProcess'
  316. self._parent_pid = None
  317. self._popen = None
  318. self._config = {'authkey': AuthenticationString(os.urandom(32)),
  319. 'semprefix': '/mp'}
  320. _current_process = _MainProcess()
  321. _process_counter = itertools.count(1)
  322. _children = set()
  323. del _MainProcess
  324. Process = BaseProcess
  325. #
  326. # Give names to some return codes
  327. #
  328. _exitcode_to_name = {}
  329. for name, signum in items(signal.__dict__):
  330. if name[:3] == 'SIG' and '_' not in name:
  331. _exitcode_to_name[-signum] = name
  332. # For debug and leak testing
  333. _dangling = WeakSet()