python-2.6.6-subprocess-timeout.patch

diff -up Python-2.6.6/Lib/subprocess.py.subprocess-timeout Python-2.6.6/Lib/subprocess.py
--- Python-2.6.6/Lib/subprocess.py.subprocess-timeout	2011-01-18 19:44:29.006987137 -0500
+++ Python-2.6.6/Lib/subprocess.py	2011-01-19 15:02:14.670776642 -0500
@@ -384,6 +384,7 @@ import types
 import traceback
 import gc
 import signal
+import time
 
 # Exception classes used by this module.
 class CalledProcessError(Exception):
@@ -397,6 +398,12 @@ class CalledProcessError(Exception):
         return "Command '%s' returned non-zero exit status %d" % (self.cmd, self.returncode)
 
 
+class TimeoutExpired(Exception):
+    """This exception is raised when the timeout expires while waiting for a
+    child process.
+    """
+
+
 if mswindows:
     from _subprocess import CREATE_NEW_CONSOLE
     import threading
@@ -460,14 +467,21 @@ def _eintr_retry_call(func, *args):
 
 
 def call(*popenargs, **kwargs):
-    """Run command with arguments.  Wait for command to complete, then
-    return the returncode attribute.
+    """Run command with arguments.  Wait for command to complete or
+    timeout, then return the returncode attribute.
 
     The arguments are the same as for the Popen constructor.  Example:
 
     retcode = call(["ls", "-l"])
     """
-    return Popen(*popenargs, **kwargs).wait()
+    timeout = kwargs.pop('timeout', None)
+    p = Popen(*popenargs, **kwargs)
+    try:
+        return p.wait(timeout=timeout)
+    except TimeoutExpired:
+        p.kill()
+        p.wait()
+        raise
 
 
 def check_call(*popenargs, **kwargs):
@@ -569,6 +583,8 @@ class Popen(object):
         _cleanup()
 
         self._child_created = False
+        self._input = None
+        self._communication_started = False
         if not isinstance(bufsize, (int, long)):
             raise TypeError("bufsize must be an integer")
 
@@ -661,18 +677,35 @@ class Popen(object):
             _active.append(self)
 
 
-    def communicate(self, input=None):
+    def communicate(self, input=None, timeout=None):
         """Interact with process: Send data to stdin.  Read data from
         stdout and stderr, until end-of-file is reached.  Wait for
         process to terminate.  The optional input argument should be a
         string to be sent to the child process, or None, if no data
         should be sent to the child.
 
-        communicate() returns a tuple (stdout, stderr)."""
+        communicate() returns a tuple (stdout, stderr).
+        
+        The optional "timeout" argument is a non-standard addition to
+        Python 2.6.  If supplied, it is a number of seconds, which can be an
+        integer or a float (though there are no precision guarantees).  A
+        TimeoutExpired exception will be raised after the given number of
+        seconds elapses, if the call has not yet returned.
+        """
+
+        if self._communication_started and input:
+            raise ValueError("Cannot send input after starting communication")
+
+        if timeout is not None:
+            endtime = time.time() + timeout
+        else:
+            endtime = None
 
-        # Optimization: If we are only using one pipe, or no pipe at
-        # all, using select() or threads is unnecessary.
-        if [self.stdin, self.stdout, self.stderr].count(None) >= 2:
+        # Optimization: If we are not worried about timeouts, we haven't
+        # started communicating, and we have one or zero pipes, using select()
+        # or threads is unnecessary.
+        if (endtime is None and not self._communication_started and
+            [self.stdin, self.stdout, self.stderr].count(None) >= 2):
             stdout = None
             stderr = None
             if self.stdin:
@@ -688,13 +721,52 @@ class Popen(object):
             self.wait()
             return (stdout, stderr)
 
-        return self._communicate(input)
+        try:
+            stdout, stderr = self._communicate(input, endtime)
+        finally:
+            self._communication_started = True
+
+        # All data exchanged.  Translate lists into strings.
+        if stdout is not None:
+            stdout = ''.join(stdout)
+        if stderr is not None:
+            stderr = ''.join(stderr)
+
+        # Translate newlines, if requested.  We cannot let the file
+        # object do the translation: It is based on stdio, which is
+        # impossible to combine with select (unless forcing no
+        # buffering).
+        if self.universal_newlines and hasattr(file, 'newlines'):
+            if stdout:
+                stdout = self._translate_newlines(stdout)
+            if stderr:
+                stderr = self._translate_newlines(stderr)
+
+        sts = self.wait(timeout=self._remaining_time(endtime))
+
+        return (stdout, stderr)
 
 
     def poll(self):
         return self._internal_poll()
 
 
+    def _remaining_time(self, endtime):
+        """Convenience for _communicate when computing timeouts."""
+        if endtime is None:
+            return None
+        else:
+            return endtime - time.time()
+
+
+    def _check_timeout(self, endtime):
+        """Convenience for checking if a timeout has expired."""
+        if endtime is None:
+            return
+        if time.time() > endtime:
+            raise TimeoutExpired
+
+
     if mswindows:
         #
         # Windows methods
@@ -875,12 +947,17 @@ class Popen(object):
             return self.returncode
 
 
-        def wait(self):
+        def wait(self, timeout=None):
             """Wait for child process to terminate.  Returns returncode
             attribute."""
+            if timeout is None:
+                timeout = _subprocess.INFINITE
+            else:
+                timeout = int(timeout * 1000)
             if self.returncode is None:
-                _subprocess.WaitForSingleObject(self._handle,
-                                                _subprocess.INFINITE)
+                result = _subprocess.WaitForSingleObject(self._handle, timeout)
+                if result == _subprocess.WAIT_TIMEOUT:
+                    raise TimeoutExpired
                 self.returncode = _subprocess.GetExitCodeProcess(self._handle)
             return self.returncode
 
@@ -889,50 +966,52 @@ class Popen(object):
             buffer.append(fh.read())
 
 
-        def _communicate(self, input):
-            stdout = None # Return
-            stderr = None # Return
-
-            if self.stdout:
-                stdout = []
-                stdout_thread = threading.Thread(target=self._readerthread,
-                                                 args=(self.stdout, stdout))
-                stdout_thread.setDaemon(True)
-                stdout_thread.start()
-            if self.stderr:
-                stderr = []
-                stderr_thread = threading.Thread(target=self._readerthread,
-                                                 args=(self.stderr, stderr))
-                stderr_thread.setDaemon(True)
-                stderr_thread.start()
+        def _communicate(self, input, endtime):
+            # Start reader threads feeding into a list hanging off of this
+            # object, unless they've already been started.
+            if self.stdout and not hasattr(self, "_stdout_buff"):
+                self._stdout_buff = []
+                self.stdout_thread = \
+                        threading.Thread(target=self._readerthread,
+                                         args=(self.stdout, self._stdout_buff))
+                self.stdout_thread.setDaemon(True)
+                self.stdout_thread.start()
+            if self.stderr and not hasattr(self, "_stderr_buff"):
+                self._stderr_buff = []
+                self.stderr_thread = \
+                        threading.Thread(target=self._readerthread,
+                                         args=(self.stderr, self._stderr_buff))
+                self.stderr_thread.setDaemon(True)
+                self.stderr_thread.start()
 
             if self.stdin:
                 if input is not None:
                     self.stdin.write(input)
                 self.stdin.close()
 
+            # Wait for the reader threads, or time out.  If we timeout, the
+            # threads remain reading and the fds left open in case the user
+            # calls communicate again.
             if self.stdout:
-                stdout_thread.join()
+                self.stdout_thread.join(self._remaining_time(endtime))
+                if self.stdout_thread.isAlive():
+                    raise TimeoutExpired
             if self.stderr:
-                stderr_thread.join()
+                self.stderr_thread.join(self._remaining_time(endtime))
+                if self.stderr_thread.isAlive():
+                    raise TimeoutExpired
 
-            # All data exchanged.  Translate lists into strings.
-            if stdout is not None:
-                stdout = stdout[0]
-            if stderr is not None:
-                stderr = stderr[0]
-
-            # Translate newlines, if requested.  We cannot let the file
-            # object do the translation: It is based on stdio, which is
-            # impossible to combine with select (unless forcing no
-            # buffering).
-            if self.universal_newlines and hasattr(file, 'newlines'):
-                if stdout:
-                    stdout = self._translate_newlines(stdout)
-                if stderr:
-                    stderr = self._translate_newlines(stderr)
+            # Collect the output from and close both pipes, now that we know
+            # both have been read successfully.
+            stdout = None
+            stderr = None
+            if self.stdout:
+                stdout = self._stdout_buff
+                self.stdout.close()
+            if self.stderr:
+                stderr = self._stderr_buff
+                self.stderr.close()
 
-            self.wait()
             return (stdout, stderr)
 
         def send_signal(self, sig):
@@ -1175,17 +1254,44 @@ class Popen(object):
             return self.returncode
 
 
-        def wait(self):
+        def wait(self, timeout=None, endtime=None):
             """Wait for child process to terminate.  Returns returncode
-            attribute."""
-            if self.returncode is None:
+            attribute.
+
+            The optional "timeout" argument is a non-standard addition to
+            Python 2.6.  If supplied, it is a number of seconds, which can be
+            an integer or a float (though there are no precision guarantees).
+            A TimeoutExpired exception will be raised after the given number of
+            seconds elapses, if the call has not yet returned.
+            """
+            if self.returncode is not None:
+                return self.returncode
+            if timeout is not None:
+                # Enter a busy loop if we have a timeout.  This busy
+                # loop was cribbed from Lib/threading.py in
+                # Thread.wait() at r71065.
+                endtime = time.time() + timeout
+                delay = 0.0005 # 500 us -> initial delay of 1 ms
+                while True:
+                    (pid, sts) = _eintr_retry_call(os.waitpid,
+                                                   self.pid, os.WNOHANG)
+                    assert pid == self.pid or pid == 0
+                    if pid == self.pid:
+                        self._handle_exitstatus(sts)
+                        break
+                    remaining = endtime - time.time()
+                    if remaining <= 0:
+                        raise TimeoutExpired
+                    delay = min(delay * 2, remaining, .05)
+                    time.sleep(delay)
+            elif self.returncode is None:
                 pid, sts = _eintr_retry_call(os.waitpid, self.pid, 0)
                 self._handle_exitstatus(sts)
             return self.returncode
 
 
-        def _communicate(self, input):
-            if self.stdin:
+        def _communicate(self, input, endtime):
+            if self.stdin and not self._communication_started:
                 # Flush stdio buffer.  This might block, if the user has
                 # been writing to .stdin in an uncontrolled fashion.
                 self.stdin.flush()
@@ -1193,9 +1299,9 @@ class Popen(object):
                     self.stdin.close()
 
             if _has_poll:
-                stdout, stderr = self._communicate_with_poll(input)
+                stdout, stderr = self._communicate_with_poll(input, endtime)
             else:
-                stdout, stderr = self._communicate_with_select(input)
+                stdout, stderr = self._communicate_with_select(input, endtime)
 
             # All data exchanged.  Translate lists into strings.
             if stdout is not None:
@@ -1213,57 +1319,73 @@ class Popen(object):
                 if stderr:
                     stderr = self._translate_newlines(stderr)
 
-            self.wait()
+            self.wait(timeout=self._remaining_time(endtime))
             return (stdout, stderr)
 
 
-        def _communicate_with_poll(self, input):
+        def _communicate_with_poll(self, input, endtime):
             stdout = None # Return
             stderr = None # Return
-            fd2file = {}
-            fd2output = {}
+
+            if not self._communication_started:
+                self._fd2file = {}
 
             poller = select.poll()
             def register_and_append(file_obj, eventmask):
                 poller.register(file_obj.fileno(), eventmask)
-                fd2file[file_obj.fileno()] = file_obj
+                self._fd2file[file_obj.fileno()] = file_obj
 
             def close_unregister_and_remove(fd):
                 poller.unregister(fd)
-                fd2file[fd].close()
-                fd2file.pop(fd)
+                self._fd2file[fd].close()
+                self._fd2file.pop(fd)
 
             if self.stdin and input:
                 register_and_append(self.stdin, select.POLLOUT)
 
+            # Only create this mapping if we haven't already.
+            if not self._communication_started:
+                self._fd2output = {}
+                if self.stdout:
+                    self._fd2output[self.stdout.fileno()] = []
+                if self.stderr:
+                    self._fd2output[self.stderr.fileno()] = []
+
             select_POLLIN_POLLPRI = select.POLLIN | select.POLLPRI
             if self.stdout:
                 register_and_append(self.stdout, select_POLLIN_POLLPRI)
-                fd2output[self.stdout.fileno()] = stdout = []
+                stdout = self._fd2output[self.stdout.fileno()]
             if self.stderr:
                 register_and_append(self.stderr, select_POLLIN_POLLPRI)
-                fd2output[self.stderr.fileno()] = stderr = []
+                stderr = self._fd2output[self.stderr.fileno()]
+
+            # Save the input here so that if we time out while communicating,
+            # we can continue sending input if we retry.
+            if not self._input:
+                self._input_offset = 0
+                self._input = input
 
-            input_offset = 0
-            while fd2file:
+            while self._fd2file:
                 try:
-                    ready = poller.poll()
+                    ready = poller.poll(self._remaining_time(endtime))
                 except select.error, e:
                     if e.args[0] == errno.EINTR:
                         continue
                     raise
+                self._check_timeout(endtime)
 
                 for fd, mode in ready:
                     if mode & select.POLLOUT:
-                        chunk = input[input_offset : input_offset + _PIPE_BUF]
-                        input_offset += os.write(fd, chunk)
-                        if input_offset >= len(input):
+                        chunk = self._input[self._input_offset :
+                                            self._input_offset + _PIPE_BUF]
+                        self._input_offset += os.write(fd, chunk)
+                        if self._input_offset >= len(self._input):
                             close_unregister_and_remove(fd)
                     elif mode & select_POLLIN_POLLPRI:
                         data = os.read(fd, 4096)
                         if not data:
                             close_unregister_and_remove(fd)
-                        fd2output[fd].append(data)
+                        self._fd2output[fd].append(data)
                     else:
                         # Ignore hang up or errors.
                         close_unregister_and_remove(fd)
@@ -1271,50 +1393,71 @@ class Popen(object):
             return (stdout, stderr)
 
 
-        def _communicate_with_select(self, input):
-            read_set = []
-            write_set = []
+        def _communicate_with_select(self, input, endtime):
+            if not self._communication_started:
+                self._read_set = []
+                self._write_set = []
+                if self.stdin and input:
+                    self._write_set.append(self.stdin)
+                if self.stdout:
+                    self._read_set.append(self.stdout)
+                if self.stderr:
+                    self._read_set.append(self.stderr)
+
+            if not self._input:
+                self._input = input
+                self._input_offset = 0
+
             stdout = None # Return
             stderr = None # Return
 
-            if self.stdin and input:
-                write_set.append(self.stdin)
             if self.stdout:
-                read_set.append(self.stdout)
-                stdout = []
+                if not self._communication_started:
+                    self._stdout_buff = []
+                stdout = self._stdout_buff
             if self.stderr:
-                read_set.append(self.stderr)
-                stderr = []
+                if not self._communication_started:
+                    self._stderr_buff = []
+                stderr = self._stderr_buff
 
-            input_offset = 0
-            while read_set or write_set:
+            while self._read_set or self._write_set:
+                remaining = self._remaining_time(endtime)
                 try:
-                    rlist, wlist, xlist = select.select(read_set, write_set, [])
+                    rlist, wlist, xlist = select.select(self._read_set,
+                                                        self._write_set, [],
+                                                        remaining)
                 except select.error, e:
                     if e.args[0] == errno.EINTR:
                         continue
                     raise
+                self._check_timeout(endtime)
+
+                if not (rlist or wlist or xlist):
+                    # According to the docs, returning three empty lists
+                    # indicates that the timeout expired.
+                    raise TimeoutExpired
 
                 if self.stdin in wlist:
-                    chunk = input[input_offset : input_offset + _PIPE_BUF]
+                    chunk = self._input[self._input_offset :
+                                  self._input_offset + _PIPE_BUF]
                     bytes_written = os.write(self.stdin.fileno(), chunk)
-                    input_offset += bytes_written
-                    if input_offset >= len(input):
+                    self._input_offset += bytes_written
+                    if self._input_offset >= len(self._input):
                         self.stdin.close()
-                        write_set.remove(self.stdin)
+                        self._write_set.remove(self.stdin)
 
                 if self.stdout in rlist:
                     data = os.read(self.stdout.fileno(), 1024)
                     if data == "":
                         self.stdout.close()
-                        read_set.remove(self.stdout)
+                        self._read_set.remove(self.stdout)
                     stdout.append(data)
 
                 if self.stderr in rlist:
                     data = os.read(self.stderr.fileno(), 1024)
                     if data == "":
                         self.stderr.close()
-                        read_set.remove(self.stderr)
+                        self._read_set.remove(self.stderr)
                     stderr.append(data)
 
             return (stdout, stderr)
diff -up Python-2.6.6/Lib/test/test_subprocess.py.subprocess-timeout Python-2.6.6/Lib/test/test_subprocess.py
--- Python-2.6.6/Lib/test/test_subprocess.py.subprocess-timeout	2011-01-18 19:44:29.007985793 -0500
+++ Python-2.6.6/Lib/test/test_subprocess.py	2011-01-18 19:54:26.132986067 -0500
@@ -48,6 +48,13 @@ class ProcessTestCase(unittest.TestCase)
             fname = tempfile.mktemp()
             return os.open(fname, os.O_RDWR|os.O_CREAT), fname
 
+    def assertStderrEqual(self, stderr, expected, msg=None):
+        # In a debug build, stuff like "[6580 refs]" is printed to stderr at
+        # shutdown time.  That frustrates tests trying to check stderr produced
+        # from a spawned Python process.
+        actual = re.sub(br"\[\d+ refs\]\r?\n?$", b"", stderr)
+        self.assertEqual(actual, expected, msg)
+
     #
     # Generic tests
     #
@@ -57,6 +64,15 @@ class ProcessTestCase(unittest.TestCase)
                               "import sys; sys.exit(47)"])
         self.assertEqual(rc, 47)
 
+    def test_call_timeout(self):
+        # call() function with timeout argument; we want to test that the child
+        # process gets killed when the timeout expires.  If the child isn't
+        # killed, this call will deadlock since subprocess.call waits for the
+        # child.
+        self.assertRaises(subprocess.TimeoutExpired, subprocess.call,
+                          [sys.executable, "-c", "while True: pass"],
+                          timeout=0.1)
+
     def test_check_call_zero(self):
         # check_call() function with zero return code
         rc = subprocess.check_call([sys.executable, "-c",
@@ -303,6 +319,41 @@ class ProcessTestCase(unittest.TestCase)
         self.assertEqual(remove_stderr_debug_decorations(stderr),
                          "pineapple")
 
+    def test_communicate_timeout(self):
+        p = subprocess.Popen([sys.executable, "-c",
+                              'import sys,os,time;'
+                              'sys.stderr.write("pineapple\\n");'
+                              'time.sleep(2);'
+                              'sys.stderr.write("pear\\n");'
+                              'sys.stdout.write(sys.stdin.read())'],
+                             universal_newlines=True,
+                             stdin=subprocess.PIPE,
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+        self.assertRaises(subprocess.TimeoutExpired, p.communicate, "banana",
+                          timeout=1)
+        # Make sure we can keep waiting for it, and that we get the whole output
+        # after it completes.
+        (stdout, stderr) = p.communicate()
+        self.assertEqual(stdout, "banana")
+        self.assertStderrEqual(stderr, "pineapple\npear\n")
+
+    def test_communicate_timeout_large_ouput(self):
+        # Test a expring timeout while the child is outputting lots of data.
+        p = subprocess.Popen([sys.executable, "-c",
+                              'import sys,os,time;'
+                              'sys.stdout.write("a" * (64 * 1024));'
+                              'time.sleep(0.2);'
+                              'sys.stdout.write("a" * (64 * 1024));'
+                              'time.sleep(0.2);'
+                              'sys.stdout.write("a" * (64 * 1024));'
+                              'time.sleep(0.2);'
+                              'sys.stdout.write("a" * (64 * 1024));'],
+                             stdout=subprocess.PIPE)
+        self.assertRaises(subprocess.TimeoutExpired, p.communicate, timeout=0.4)
+        (stdout, _) = p.communicate()
+        self.assertEqual(len(stdout), 4 * 64 * 1024)
+
     # This test is Linux specific for simplicity to at least have
     # some coverage.  It is not a platform specific bug.
     if os.path.isdir('/proc/%d/fd' % os.getpid()):
@@ -475,6 +526,13 @@ class ProcessTestCase(unittest.TestCase)
         self.assertEqual(p.wait(), 0)
 
 
+    def test_wait_timeout(self):
+        p = subprocess.Popen([sys.executable,
+                              "-c", "import time; time.sleep(1)"])
+        self.assertRaises(subprocess.TimeoutExpired, p.wait, timeout=0.1)
+        self.assertEqual(p.wait(timeout=2), 0)
+
+
     def test_invalid_bufsize(self):
         # an invalid type of the bufsize argument should raise
         # TypeError.
diff -up Python-2.6.6/PC/_subprocess.c.subprocess-timeout Python-2.6.6/PC/_subprocess.c
--- Python-2.6.6/PC/_subprocess.c.subprocess-timeout	2010-05-09 11:15:40.000000000 -0400
+++ Python-2.6.6/PC/_subprocess.c	2011-01-18 19:44:29.054989261 -0500
@@ -668,5 +668,6 @@ init_subprocess()
     defint(d, "SW_HIDE", SW_HIDE);
     defint(d, "INFINITE", INFINITE);
     defint(d, "WAIT_OBJECT_0", WAIT_OBJECT_0);
+    defint(d, "WAIT_TIMEOUT", WAIT_TIMEOUT);
     defint(d, "CREATE_NEW_CONSOLE", CREATE_NEW_CONSOLE);
 }