process.py 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330
  1. #
  2. # Module providing the `Process` class which emulates `threading.Thread`
  3. #
  4. # processing/process.py
  5. #
  6. # Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
  7. #
  8. __all__ = [
  9. 'Process', 'currentProcess', 'activeChildren'
  10. ]
  11. #
  12. # Imports
  13. #
  14. import os
  15. import sys
  16. import time
  17. import signal
  18. import atexit
  19. import weakref
  20. import copy_reg
  21. import itertools
  22. #
  23. # Public functions
  24. #
  25. def currentProcess():
  26. '''
  27. Return process object representing the current process
  28. '''
  29. return _current_process
  30. def activeChildren():
  31. '''
  32. Return list of process objects corresponding to live child processes
  33. '''
  34. _cleanup()
  35. return list(_current_process._children)
  36. #
  37. #
  38. #
  39. def _cleanup():
  40. '''
  41. Purge `_children` of dead processes
  42. '''
  43. for p in list(_current_process._children):
  44. if p._popen.poll() is not None:
  45. _current_process._children.discard(p)
  46. #
  47. # The `Process` class
  48. #
  49. class Process(object):
  50. '''
  51. Process objects represent activity that is run in a separate process
  52. The class is analagous to `threading.Thread`
  53. '''
  54. def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
  55. counter = _current_process._counter.next()
  56. self._identity = _current_process._identity + (counter,)
  57. self._authkey = _current_process._authkey
  58. self._daemonic = _current_process._daemonic
  59. self._parent_pid = os.getpid()
  60. self._popen = None
  61. self._exiting = False
  62. self._target = target
  63. self._args = tuple(args)
  64. self._kwargs = kwargs.copy()
  65. self._name = name or 'Process-' + ':'.join(map(str, self._identity))
  66. def run(self):
  67. '''
  68. Method to be run in sub-process; can be overridden in sub-class
  69. '''
  70. if self._target:
  71. self._target(*self._args, **self._kwargs)
  72. def start(self):
  73. '''
  74. Start child process
  75. '''
  76. from processing.forking import Popen
  77. assert self._popen is None, 'cannot start a process twice'
  78. assert self._parent_pid == os.getpid(), \
  79. 'can only start a process object created by current process'
  80. _cleanup()
  81. self._popen = Popen(self)
  82. _current_process._children.add(self)
  83. def terminate(self):
  84. '''
  85. Terminate process; sends `SIGTERM` signal or uses `TerminateProcess()`
  86. '''
  87. self._popen.terminate()
  88. def join(self, timeout=None):
  89. '''
  90. Wait until child process terminates
  91. '''
  92. assert self._parent_pid == os.getpid(), 'can only join a child process'
  93. assert self._popen is not None, 'can only join a started process'
  94. if timeout == 0:
  95. res = self._popen.poll()
  96. elif timeout is None:
  97. res = self._popen.wait()
  98. else:
  99. res = self._popen.waitTimeout(timeout)
  100. if res is not None:
  101. _current_process._children.discard(self)
  102. def isAlive(self):
  103. '''
  104. Return whether child process is alive
  105. '''
  106. if self is _current_process:
  107. return True
  108. assert self._parent_pid == os.getpid(), 'can only test a child process'
  109. if self._popen is None:
  110. return False
  111. self._popen.poll()
  112. return self._popen.returncode is None
  113. def getName(self):
  114. '''
  115. Return name of process
  116. '''
  117. return self._name
  118. def setName(self, name):
  119. '''
  120. Set name of process
  121. '''
  122. assert type(name) is str, 'name must be a string'
  123. self._name = name
  124. def isDaemon(self):
  125. '''
  126. Return whether process is a daemon
  127. '''
  128. return self._daemonic
  129. def setDaemon(self, daemonic):
  130. '''
  131. Set whether process is a daemon
  132. '''
  133. assert self._popen is None, 'process has already started'
  134. self._daemonic = daemonic
  135. def getAuthKey(self):
  136. '''
  137. Return authorization key of process
  138. '''
  139. return self._authkey
  140. def setAuthKey(self, authkey):
  141. '''
  142. Set authorization key of process
  143. '''
  144. assert type(authkey) is str, 'value must be a string'
  145. self._authkey = authkey
  146. def getExitCode(self):
  147. '''
  148. Return exit code of process or `None` if it has yet to stop
  149. '''
  150. if self._popen is None:
  151. return self._popen
  152. return self._popen.poll()
  153. def getPid(self):
  154. '''
  155. Return PID of process or `None` if it has yet to start
  156. '''
  157. if self is _current_process:
  158. return os.getpid()
  159. else:
  160. assert self._parent_pid == os.getpid(), 'not a child process'
  161. return self._popen and self._popen.pid
  162. def __repr__(self):
  163. if self is _current_process:
  164. status = 'started'
  165. elif self._parent_pid != os.getpid():
  166. status = 'unknown'
  167. elif self._popen is None:
  168. status = 'initial'
  169. else:
  170. if self._popen.poll() is not None:
  171. status = self.getExitCode()
  172. else:
  173. status = 'started'
  174. if type(status) is int:
  175. if status == 0:
  176. status = 'stopped'
  177. else:
  178. status = 'stopped[%s]' % _exitcode_to_name.get(status, status)
  179. return '<%s(%s, %s%s)>' % (type(self).__name__, self._name,
  180. status, self._daemonic and ' daemon' or '')
  181. ##
  182. def _bootstrap(self):
  183. from processing.finalize import _registry
  184. from processing.logger import info
  185. global _current_process
  186. try:
  187. self._children = set()
  188. self._counter = itertools.count(1)
  189. sys.stdin.close()
  190. _registry.clear()
  191. _current_process = self
  192. _runAfterForkers()
  193. info('child process calling self.run()')
  194. try:
  195. self.run()
  196. exitcode = 0
  197. finally:
  198. _exitFunction()
  199. except SystemExit, e:
  200. if not e.args:
  201. exitcode = 1
  202. elif type(e.args[0]) is int:
  203. exitcode = e.args[0]
  204. else:
  205. print >>sys.stderr, e.args[0]
  206. exitcode = 1
  207. except:
  208. exitcode = 1
  209. import traceback
  210. print >>sys.stderr, 'Process %s:' % self.getName()
  211. traceback.print_exc()
  212. info('process exiting with exitcode %d' % exitcode)
  213. return exitcode
  214. #
  215. # Create object representing the main process
  216. #
  217. class _MainProcess(Process):
  218. def __init__(self):
  219. self._identity = ()
  220. self._daemonic = False
  221. self._name = 'MainProcess'
  222. self._parent_pid = None
  223. self._popen = None
  224. self._counter = itertools.count(1)
  225. self._children = set()
  226. self._authkey = ''.join('%02x' % ord(c) for c in os.urandom(16))
  227. _current_process = _MainProcess()
  228. del _MainProcess
  229. #
  230. # Give names to some return codes
  231. #
  232. _exitcode_to_name = {}
  233. for name, signum in signal.__dict__.items():
  234. if name[:3]=='SIG' and '_' not in name:
  235. _exitcode_to_name[-signum] = name
  236. #
  237. # Make bound and unbound instance methods and class methods picklable
  238. #
  239. def _reduceMethod(m):
  240. if m.im_self is None:
  241. return getattr, (m.im_class, m.im_func.func_name)
  242. else:
  243. return getattr, (m.im_self, m.im_func.func_name)
  244. copy_reg.pickle(type(_current_process.start), _reduceMethod)
  245. #
  246. # Support for reinitialization of objects when bootstrapping a child process
  247. #
  248. _afterfork_registry = weakref.WeakValueDictionary()
  249. _afterForkerId = itertools.count().next
  250. def _runAfterForkers():
  251. # execute in order of registration
  252. for (index, ident, func), obj in sorted(_afterfork_registry.items()):
  253. func(obj)
  254. def _registerAfterFork(obj, func):
  255. _afterfork_registry[(_afterForkerId(), id(obj), func)] = obj
  256. #
  257. # Clean up on exit
  258. #
  259. def _exitFunction():
  260. from processing.finalize import _runFinalizers
  261. from processing.logger import info
  262. _current_process._exiting = True
  263. info('running all "atexit" finalizers with priority >= 0')
  264. _runFinalizers(0)
  265. for p in activeChildren():
  266. if p._daemonic:
  267. info('calling `terminate()` for daemon %s', p.getName())
  268. p._popen.terminate()
  269. for p in activeChildren():
  270. info('calling `join()` for process %s', p.getName())
  271. p.join()
  272. info('running the remaining "atexit" finalizers')
  273. _runFinalizers()
  274. atexit.register(_exitFunction)