eventlet/eventlet/processes.py

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