import warnings

warnings.warn("eventlet.processes is deprecated in favor of "

"eventlet.green.subprocess, which is API-compatible with the standard "

" library subprocess module.",

DeprecationWarning, stacklevel=2)

import errno

import os

import signal

import eventlet

from eventlet import greenio, pools

from eventlet.green import subprocess

class DeadProcess(RuntimeError):

pass

def cooperative_wait(pobj, check_interval=0.01):

""" Waits for a child process to exit, returning the status

code.

Unlike ``os.wait``, :func:`cooperative_wait` does not block the entire

process, only the calling coroutine. If the child process does not die,

:func:`cooperative_wait` could wait forever.

The argument *check_interval* is the amount of time, in seconds, that

:func:`cooperative_wait` will sleep between calls to ``os.waitpid``.

"""

try:

while True:

status = pobj.poll()

if status >= 0:

return status

eventlet.sleep(check_interval)

except OSError as e:

if e.errno == errno.ECHILD:

# no child process, this happens if the child process

# already died and has been cleaned up, or if you just

# called with a random pid value

return -1

else:

raise

class Process(object):

"""Construct Process objects, then call read, and write on them."""

process_number = 0

def __init__(self, command, args, dead_callback=None):

self.process_number = self.process_number + 1

Process.process_number = self.process_number

self.command = command

self.args = args

self._dead_callback = dead_callback

self.run()

def run(self):

self.dead = False

self.started = False

self.proc = None

args = [self.command]

args.extend(self.args)

self.proc = subprocess.Popen(

args=args,

shell=False,

stdin=subprocess.PIPE,

stdout=subprocess.PIPE,

stderr=subprocess.STDOUT,

close_fds=True,

)

self.child_stdout_stderr = self.proc.stdout

self.child_stdin = self.proc.stdin

self.sendall = self.child_stdin.write

self.send = self.child_stdin.write

self.recv = self.child_stdout_stderr.read

self.readline = self.child_stdout_stderr.readline

self._read_first_result = False

def wait(self):

return cooperative_wait(self.proc)

def dead_callback(self):

self.wait()

self.dead = True

if self._dead_callback:

self._dead_callback()

def makefile(self, mode, *arg):

if mode.startswith('r'):

return self.child_stdout_stderr

if mode.startswith('w'):

return self.child_stdin

raise RuntimeError("Unknown mode", mode)

def read(self, amount=None):

"""Reads from the stdout and stderr of the child process.

The first call to read() will return a string; subsequent

calls may raise a DeadProcess when EOF occurs on the pipe.

"""

result = self.child_stdout_stderr.read(amount)

if result == '' and self._read_first_result:

# This process is dead.

self.dead_callback()

raise DeadProcess

else:

self._read_first_result = True

return result

def write(self, stuff):

written = 0

try:

written = self.child_stdin.write(stuff)

self.child_stdin.flush()

except ValueError as e:

## File was closed

assert str(e) == 'I/O operation on closed file'

if written == 0:

self.dead_callback()

raise DeadProcess

def flush(self):

self.child_stdin.flush()

def close(self):

self.child_stdout_stderr.close()

self.child_stdin.close()

self.dead_callback()

def close_stdin(self):

self.child_stdin.close()

def kill(self, sig=None):

if sig is None:

sig = signal.SIGTERM

pid = self.getpid()

os.kill(pid, sig)

def getpid(self):

return self.proc.pid

class ProcessPool(pools.Pool):

def __init__(self, command, args=None, min_size=0, max_size=4):

"""*command*

the command to run

"""

self.command = command

if args is None:

args = []

self.args = args

pools.Pool.__init__(self, min_size, max_size)

def create(self):

"""Generate a process

"""

def dead_callback():

self.current_size -= 1

return Process(self.command, self.args, dead_callback)

def put(self, item):

if not item.dead:

if item.proc.poll() != -1:

item.dead_callback()

else:

pools.Pool.put(self, item)