Refactor nag functionality in to NagTimer class.

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 114 matching lines @@
   # Requires modifications.
   env = env.copy()
   def fix_lang(name):
     if not is_english(name):
       env[name] = 'en_US.UTF-8'
   fix_lang('LANG')
   fix_lang('LANGUAGE')
   return env
 
 
+class NagTimer(object):
+  """
+  Triggers a callback when a time interval passes without an event being fired.
+
+  For example, the event could be receiving terminal output from a subprocess;
+  and the callback could print a warning to stderr that the subprocess appeared
+  to be hung.
+  """
+  def __init__(self, interval, cb):
+    self.interval = interval
+    self.cb = cb
+    self.timer = threading.Timer(self.interval, self.fn)
+    self.last_output = self.previous_last_output = 0
+
+  def start(self):
+    self.last_output = self.previous_last_output = time.time()
+    self.timer.start()
+
+  def event(self):
+    self.last_output = time.time()
+
+  def fn(self):
+    now = time.time()
+    if self.last_output == self.previous_last_output:
+      self.cb(now - self.previous_last_output)
+    # Use 0.1 fudge factor, just in case
+    #   (self.last_output - now) is very close to zero.
+    sleep_time = (self.last_output - now - 0.1) % self.interval

[M-A Ruel, 2013/05/03 16:42:58]

% with float? You should use an explicit cast in that case and also in
self.internal = int(interval)

[szager1, 2013/05/03 18:16:58]

Can you explain why?  I can't think of a good reason to forbid floats here.  The
float type in python fully supports __mod__.

[M-A Ruel, 2013/05/03 18:32:34]

Ah ok I hadn't realized that.

+    self.previous_last_output = self.last_output
+    self.timer = threading.Timer(sleep_time + 0.1, self.fn)
+    self.timer.start()
+
+  def cancel(self):
+    self.timer.cancel()
+
+
 class Popen(subprocess.Popen):
   """Wraps subprocess.Popen() with various workarounds.
 
   - Forces English output since it's easier to parse the stdout if it is always
     in English.
   - Sets shell=True on windows by default. You can override this by forcing
     shell parameter to a value.
   - Adds support for VOID to not buffer when not needed.
   - Adds self.start property.
 
@@ skipping 78 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
+    nag = None
 
     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()
+          if nag:
+            nag.event()
           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 18 matching lines @@
         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()
+      def _nag_cb(elapsed):
+        logging.warn('  No output for %.0f seconds from command:' % elapsed)
+        logging.warn('    %s' % self.cmd_str)
+      nag = NagTimer(self.nag_timer, _nag_cb)
+      nag.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 nag:
+        nag.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
@@ skipping 120 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]