Add nag_timer.

subprocess2.py

 # coding=utf8
 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 """Collection of subprocess wrapper functions.
 
 In theory you shouldn't need anything else in subprocess, or this module failed.
 """
 
 import cStringIO
@@ skipping 157 matching lines @@
       raise CalledProcessError(None, args, kwargs.get('cwd'), None, None)
     if kwargs.get('cwd', None):
       tmp_str += ';  cwd=%s' % kwargs['cwd']
     logging.debug(tmp_str)
 
     self.stdout_cb = None
     self.stderr_cb = None
     self.stdin_is_void = False
     self.stdout_is_void = False
     self.stderr_is_void = False
+    self.cmd_str = tmp_str
 
     if kwargs.get('stdin') is VOID:
       kwargs['stdin'] = open(os.devnull, 'r')
       self.stdin_is_void = True
 
     for stream in ('stdout', 'stderr'):
       if kwargs.get(stream) in (VOID, os.devnull):
         kwargs[stream] = open(os.devnull, 'w')
         setattr(self, stream + '_is_void', True)
       if callable(kwargs.get(stream)):
         setattr(self, stream + '_cb', kwargs[stream])
         kwargs[stream] = PIPE
 
     self.start = time.time()
     self.timeout = None
+    self.nag_timer = None
     self.shell = kwargs.get('shell', None)
     # Silence pylint on MacOSX
     self.returncode = None
 
     try:
       super(Popen, self).__init__(args, **kwargs)
     except OSError, e:
       if e.errno == errno.EAGAIN and sys.platform == 'cygwin':
         # Convert fork() emulation failure into a CygwinRebaseError().
         raise CygwinRebaseError(
@@ skipping 18 matching lines @@
     effectiveness will be delayed accordingly.
     """
     # Queue of either of <threadname> when done or (<threadname>, data).  In
     # theory we would like to limit to ~64kb items to not cause large memory
     # usage when the callback blocks. It is not done because it slows down
     # processing on OSX10.6 by a factor of 2x, making it even slower than
     # Windows!  Revisit this decision if it becomes a problem, e.g. crash
     # because of memory exhaustion.
     queue = Queue.Queue()
     done = threading.Event()
+    timer = []
+    last_output = [time.time()] * 2
 
     def write_stdin():
       try:
         stdin_io = cStringIO.StringIO(input)
         while True:
           data = stdin_io.read(1024)
           if data:
             self.stdin.write(data)
           else:
             self.stdin.close()
             break
       finally:
         queue.put('stdin')
 
     def _queue_pipe_read(pipe, name):
       """Queues characters read from a pipe into a queue."""
       try:
         while True:
           data = pipe.read(1)
           if not data:
             break
+          last_output[0] = time.time()
           queue.put((name, data))
       finally:
         queue.put(name)
 
+    def nag_fn():
+      now = time.time()
+      if done.is_set():
+        return
+      if last_output[0] == last_output[1]:
+        logging.warn('  No output for %.0f seconds from command:' % (
+            now - last_output[1]))
+        logging.warn('    %s' % self.cmd_str)
+      # Use 0.1 fudge factor in case:
+      #   now ~= last_output[0] + self.nag_timer
+      sleep_time = self.nag_timer + last_output[0] - now - 0.1
+      while sleep_time < 0:
+        sleep_time += self.nag_timer
+      last_output[1] = last_output[0]
+      timer[0] = threading.Timer(sleep_time, nag_fn)
+      timer[0].start()
+
     def timeout_fn():
       try:
         done.wait(self.timeout)
       finally:
         queue.put('timeout')
 
     def wait_fn():
       try:
         self.wait()
       finally:
@@ skipping 17 matching lines @@
       threads['stderr'] = threading.Thread(
         target=_queue_pipe_read, args=(self.stderr, 'stderr'))
     if input:
       threads['stdin'] = threading.Thread(target=write_stdin)
     elif self.stdin:
       # Pipe but no input, make sure it's closed.
       self.stdin.close()
     for t in threads.itervalues():
       t.start()
 
+    if self.nag_timer:
+      timer.append(threading.Timer(self.nag_timer, nag_fn))
+      timer[0].start()
+
     timed_out = False
     try:
       # This thread needs to be optimized for speed.
       while threads:
         item = queue.get()
         if item[0] == 'stdout':
           self.stdout_cb(item[1])
         elif item[0] == 'stderr':
           self.stderr_cb(item[1])
         else:
           # A thread terminated.
           threads[item].join()
           del threads[item]
           if item == 'wait':
             # Terminate the timeout thread if necessary.
             done.set()
           elif item == 'timeout' and not timed_out and self.poll() is None:
             logging.debug('Timed out after %fs: killing' % self.timeout)
             self.kill()
             timed_out = True
     finally:
       # Stop the threads.
       done.set()
+      if timer:
+        timer[0].cancel()
       if 'wait' in threads:
         # Accelerate things, otherwise it would hang until the child process is
         # done.
         logging.debug('Killing child because of an exception')
         self.kill()
       # Join threads.
       for thread in threads.itervalues():
         thread.join()
       if timed_out:
         self.returncode = TIMED_OUT
 
-  def communicate(self, input=None, timeout=None): # pylint: disable=W0221,W0622
+  # pylint: disable=W0221,W0622
+  def communicate(self, input=None, timeout=None, nag_timer=None):
     """Adds timeout and callbacks support.
 
     Returns (stdout, stderr) like subprocess.Popen().communicate().
 
     - The process will be killed after |timeout| seconds and returncode set to
       TIMED_OUT.
+    - If the subprocess runs for |nag_timer| seconds without producing terminal
+      output, print a warning to stderr.
     """
     self.timeout = timeout
-    if not self.timeout and not self.stdout_cb and not self.stderr_cb:
+    self.nag_timer = nag_timer
+    if (not self.timeout and not self.nag_timer and
+        not self.stdout_cb and not self.stderr_cb):
       return super(Popen, self).communicate(input)
 
     if self.timeout and self.shell:
       raise TypeError(
           'Using timeout and shell simultaneously will cause a process leak '
           'since the shell will be killed instead of the child process.')
 
     stdout = None
     stderr = None
     # Convert to a lambda to workaround python's deadlock.
     # http://docs.python.org/library/subprocess.html#subprocess.Popen.wait
     # When the pipe fills up, it would deadlock this process.
     if self.stdout and not self.stdout_cb and not self.stdout_is_void:
       stdout = []
       self.stdout_cb = stdout.append
     if self.stderr and not self.stderr_cb and not self.stderr_is_void:
       stderr = []
       self.stderr_cb = stderr.append
     self._tee_threads(input)
     if stdout is not None:
       stdout = ''.join(stdout)
     if stderr is not None:
       stderr = ''.join(stderr)
     return (stdout, stderr)
 
 
-def communicate(args, timeout=None, **kwargs):
+def communicate(args, timeout=None, nag_timer=None, **kwargs):
   """Wraps subprocess.Popen().communicate() and add timeout support.
 
   Returns ((stdout, stderr), returncode).
 
   - The process will be killed after |timeout| seconds and returncode set to
     TIMED_OUT.
+  - If the subprocess runs for |nag_timer| seconds without producing terminal
+    output, print a warning to stderr.
   - Automatically passes stdin content as input so do not specify stdin=PIPE.
   """
   stdin = kwargs.pop('stdin', None)
   if stdin is not None:
     if isinstance(stdin, basestring):
       # When stdin is passed as an argument, use it as the actual input data and
       # set the Popen() parameter accordingly.
       kwargs['stdin'] = PIPE
     else:
       kwargs['stdin'] = stdin
       stdin = None
 
   proc = Popen(args, **kwargs)
   if stdin:
-    return proc.communicate(stdin, timeout), proc.returncode
+    return proc.communicate(stdin, timeout, nag_timer), proc.returncode
   else:
-    return proc.communicate(None, timeout), proc.returncode
+    return proc.communicate(None, timeout, nag_timer), proc.returncode
 
 
 def call(args, **kwargs):
   """Emulates subprocess.call().
 
   Automatically convert stdout=PIPE or stderr=PIPE to VOID.
   In no case they can be returned since no code path raises
   subprocess2.CalledProcessError.
   """
   if kwargs.get('stdout') == PIPE:
@@ skipping 42 matching lines @@
 
   - Throws if return code is not 0.
   - Works even prior to python 2.7.
   - Blocks stdin by default if not specified since no output will be visible.
   - As per doc, "The stdout argument is not allowed as it is used internally."
   """
   kwargs.setdefault('stdin', VOID)
   if 'stdout' in kwargs:
     raise ValueError('stdout argument not allowed, it will be overridden.')
   return check_call_out(args, stdout=PIPE, **kwargs)[0]