Index: Lib/test/test_subprocess.py =================================================================== --- Lib/test/test_subprocess.py (revision 82429) +++ Lib/test/test_subprocess.py (working copy) @@ -61,6 +61,15 @@ "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", @@ -338,6 +347,40 @@ self.assertEqual(stdout, "banana") self.assertStderrEqual(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())'], + 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. @unittest.skipUnless(os.path.isdir('/proc/%d/fd' % os.getpid()), @@ -512,6 +555,13 @@ 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. Index: Lib/subprocess.py =================================================================== --- Lib/subprocess.py (revision 82429) +++ Lib/subprocess.py (working copy) @@ -396,6 +396,7 @@ import traceback import gc import signal +import time # Exception classes used by this module. class CalledProcessError(Exception): @@ -412,6 +413,12 @@ 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: import threading import msvcrt @@ -476,14 +483,21 @@ 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): @@ -618,6 +632,8 @@ _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") @@ -710,7 +726,7 @@ _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 @@ -719,9 +735,19 @@ communicate() returns a tuple (stdout, stderr).""" - # 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: + 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 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: @@ -737,13 +763,52 @@ 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 @@ -924,12 +989,16 @@ 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) + # TODO(rnk): Test this on Windows. if self.returncode is None: - _subprocess.WaitForSingleObject(self._handle, - _subprocess.INFINITE) + _subprocess.WaitForSingleObject(self._handle, timeout) self.returncode = _subprocess.GetExitCodeProcess(self._handle) return self.returncode @@ -938,20 +1007,21 @@ buffer.append(fh.read()) - def _communicate(self, input): - stdout = None # Return - stderr = None # Return - - if self.stdout: - stdout = [] + 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 = [] stdout_thread = threading.Thread(target=self._readerthread, - args=(self.stdout, stdout)) + args=(self.stdout, + self._stdout_buff)) stdout_thread.setDaemon(True) stdout_thread.start() - if self.stderr: - stderr = [] + if self.stderr and not hasattr(self, "_stderr_buff"): + self._stderr_buff = [] stderr_thread = threading.Thread(target=self._readerthread, - args=(self.stderr, stderr)) + args=(self.stderr, + self._stderr_buff)) stderr_thread.setDaemon(True) stderr_thread.start() @@ -960,28 +1030,32 @@ self.stdin.write(input) self.stdin.close() + # Wait for the reader threads, or time out. if self.stdout: - stdout_thread.join() + stdout_thread.join(self._remaining_time(endtime)) + if stdout_thread.isAlive(): + raise TimeoutExpired if self.stderr: - stderr_thread.join() + stderr_thread.join(self._remaining_time(endtime)) + if stderr_thread.isAlive(): + raise TimeoutExpired + # TODO: Somebody needs to research what happens to those + # threads if they are still running. Also, what happens if + # you close a file descriptor on Windows in one thread? + # Will it interrupt the other, or does the other keep its + # own handle? - # All data exchanged. Translate lists into strings. - if stdout is not None: - stdout = stdout[0] - if stderr is not None: - stderr = stderr[0] + # 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() - # 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) - - self.wait() return (stdout, stderr) def send_signal(self, sig): @@ -1235,17 +1309,37 @@ 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: + 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() @@ -1253,9 +1347,9 @@ 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: @@ -1273,57 +1367,73 @@ 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()] - input_offset = 0 - while fd2file: + # 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 + + 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) @@ -1331,50 +1441,71 @@ 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)