processes.py 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163
  1. import warnings
  2. warnings.warn("eventlet.processes is deprecated in favor of "
  3. "eventlet.green.subprocess, which is API-compatible with the standard "
  4. " library subprocess module.",
  5. DeprecationWarning, stacklevel=2)
  6. import errno
  7. import os
  8. import popen2
  9. import signal
  10. from eventlet import api
  11. from eventlet import pools
  12. from eventlet import greenio
  13. class DeadProcess(RuntimeError):
  14. pass
  15. def cooperative_wait(pobj, check_interval=0.01):
  16. """ Waits for a child process to exit, returning the status
  17. code.
  18. Unlike ``os.wait``, :func:`cooperative_wait` does not block the entire
  19. process, only the calling coroutine. If the child process does not die,
  20. :func:`cooperative_wait` could wait forever.
  21. The argument *check_interval* is the amount of time, in seconds, that
  22. :func:`cooperative_wait` will sleep between calls to ``os.waitpid``.
  23. """
  24. try:
  25. while True:
  26. status = pobj.poll()
  27. if status >= 0:
  28. return status
  29. api.sleep(check_interval)
  30. except OSError, e:
  31. if e.errno == errno.ECHILD:
  32. # no child process, this happens if the child process
  33. # already died and has been cleaned up, or if you just
  34. # called with a random pid value
  35. return -1
  36. else:
  37. raise
  38. class Process(object):
  39. """Construct Process objects, then call read, and write on them."""
  40. process_number = 0
  41. def __init__(self, command, args, dead_callback=lambda:None):
  42. self.process_number = self.process_number + 1
  43. Process.process_number = self.process_number
  44. self.command = command
  45. self.args = args
  46. self._dead_callback = dead_callback
  47. self.run()
  48. def run(self):
  49. self.dead = False
  50. self.started = False
  51. self.popen4 = None
  52. ## We use popen4 so that read() will read from either stdout or stderr
  53. self.popen4 = popen2.Popen4([self.command] + self.args)
  54. child_stdout_stderr = self.popen4.fromchild
  55. child_stdin = self.popen4.tochild
  56. self.child_stdout_stderr = greenio.GreenPipe(child_stdout_stderr, child_stdout_stderr.mode, 0)
  57. self.child_stdin = greenio.GreenPipe(child_stdin, child_stdin.mode, 0)
  58. self.sendall = self.child_stdin.write
  59. self.send = self.child_stdin.write
  60. self.recv = self.child_stdout_stderr.read
  61. self.readline = self.child_stdout_stderr.readline
  62. self._read_first_result = False
  63. def wait(self):
  64. return cooperative_wait(self.popen4)
  65. def dead_callback(self):
  66. self.wait()
  67. self.dead = True
  68. if self._dead_callback:
  69. self._dead_callback()
  70. def makefile(self, mode, *arg):
  71. if mode.startswith('r'):
  72. return self.child_stdout_stderr
  73. if mode.startswith('w'):
  74. return self.child_stdin
  75. raise RuntimeError("Unknown mode", mode)
  76. def read(self, amount=None):
  77. """Reads from the stdout and stderr of the child process.
  78. The first call to read() will return a string; subsequent
  79. calls may raise a DeadProcess when EOF occurs on the pipe.
  80. """
  81. result = self.child_stdout_stderr.read(amount)
  82. if result == '' and self._read_first_result:
  83. # This process is dead.
  84. self.dead_callback()
  85. raise DeadProcess
  86. else:
  87. self._read_first_result = True
  88. return result
  89. def write(self, stuff):
  90. written = 0
  91. try:
  92. written = self.child_stdin.write(stuff)
  93. self.child_stdin.flush()
  94. except ValueError, e:
  95. ## File was closed
  96. assert str(e) == 'I/O operation on closed file'
  97. if written == 0:
  98. self.dead_callback()
  99. raise DeadProcess
  100. def flush(self):
  101. self.child_stdin.flush()
  102. def close(self):
  103. self.child_stdout_stderr.close()
  104. self.child_stdin.close()
  105. self.dead_callback()
  106. def close_stdin(self):
  107. self.child_stdin.close()
  108. def kill(self, sig=None):
  109. if sig == None:
  110. sig = signal.SIGTERM
  111. pid = self.getpid()
  112. os.kill(pid, sig)
  113. def getpid(self):
  114. return self.popen4.pid
  115. class ProcessPool(pools.Pool):
  116. def __init__(self, command, args=None, min_size=0, max_size=4):
  117. """*command*
  118. the command to run
  119. """
  120. self.command = command
  121. if args is None:
  122. args = []
  123. self.args = args
  124. pools.Pool.__init__(self, min_size, max_size)
  125. def create(self):
  126. """Generate a process
  127. """
  128. def dead_callback():
  129. self.current_size -= 1
  130. return Process(self.command, self.args, dead_callback)
  131. def put(self, item):
  132. if not item.dead:
  133. if item.popen4.poll() != -1:
  134. item.dead_callback()
  135. else:
  136. pools.Pool.put(self, item)