[09/21] git commit: SLIDER-86: slider.py non-blocking streaming on all platforms; debugging

[sm...@apache.org]

SLIDER-86: slider.py non-blocking streaming on all platforms; debugging


Project: http://git-wip-us.apache.org/repos/asf/incubator-slider/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-slider/commit/d6e73cd8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-slider/tree/d6e73cd8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-slider/diff/d6e73cd8

Branch: refs/heads/feature/slider_install_package
Commit: d6e73cd8d06d70cba5da9b19c9b00330161cb3ea
Parents: 1efc358
Author: Steve Loughran <st...@apache.org>
Authored: Wed Sep 10 12:12:42 2014 +0100
Committer: Steve Loughran <st...@apache.org>
Committed: Wed Sep 10 12:12:42 2014 +0100

----------------------------------------------------------------------
 slider-assembly/src/main/scripts/slider.py | 107 ++++++++++++++++--------
 1 file changed, 74 insertions(+), 33 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/d6e73cd8/slider-assembly/src/main/scripts/slider.py
----------------------------------------------------------------------
diff --git a/slider-assembly/src/main/scripts/slider.py b/slider-assembly/src/main/scripts/slider.py
index cbc0faf..4b078e7 100644
--- a/slider-assembly/src/main/scripts/slider.py
+++ b/slider-assembly/src/main/scripts/slider.py
@@ -18,6 +18,8 @@
 import sys
 import os
 import subprocess
+import time
+from threading import Thread
 
 CONF = "conf"
 
@@ -31,9 +33,18 @@ SLIDER_CLASSPATH_EXTRA = "SLIDER_CLASSPATH_EXTRA"
 SLIDER_CLASSNAME = "org.apache.slider.Slider"
 DEFAULT_JVM__OPTS = "-Djava.net.preferIPv4Stack=true -Djava.awt.headless=true -Xmx256m -Djava.confdir=%s"
 
+ON_POSIX = 'posix' in sys.builtin_module_names
+
+finished = False
+DEBUG = True
+
 """
 Launches slider
 
+Nonblocking IO on windows is "tricky" ... see
+http://stackoverflow.com/questions/375427/non-blocking-read-on-a-subprocess-pipe-in-python
+to explain the code here
+
 
 """
 
@@ -66,6 +77,20 @@ def dirMustExist(dirname):
     raise Exception("Directory does not exist: %s " % dirname)
   return dirname
 
+
+def debug(text):
+  if DEBUG: print '[DEBUG] ' + text
+
+
+def error(text):
+  print '[ERROR] ' + text
+  sys.stdout.flush()
+
+def info(text):
+  print text
+  sys.stdout.flush()
+
+
 def read(pipe, line):
   """
   read a char, append to the listing if there is a char that is not \n
@@ -86,55 +111,71 @@ def read(pipe, line):
     return line, False
 
 
+def print_output(name, src):
+  """
+  Relay the output stream to stdout line by line 
+  :param name: 
+  :param src: source stream
+  :return:
+  """
+
+  debug ("starting printer for %s" % name )
+  line = ""
+  while not finished:
+    (line, done) = read(src, line)
+    if done:
+      info(name +': ' + line)
+      line = ""
+  src.close()
+
+
 def runProcess(commandline):
   """
   Run a process
   :param commandline: command line 
   :return:the return code
   """
-  print "ready to exec : %s" % commandline
+  debug ("Executing : %s" % commandline)
   exe = subprocess.Popen(commandline,
                          stdin=None,
                          stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE,
-                         shell=False)
-  stdout = exe.stdout
-  stderr = exe.stderr
-  outline = ""
-  errline = ""
+                         shell=False,
+                         bufsize=1, 
+                         close_fds=ON_POSIX)
+
+  t = Thread(target=print_output, args=("stdout", exe.stdout))
+  t.daemon = True 
+  t.start()
+  t2 = Thread(target=print_output, args=("stderr", exe.stderr,))
+  t2.daemon = True 
+  t2.start()
+
+  debug("Waiting for completion")
   while exe.poll() is None:
     # process is running; grab output and echo every line
-    outline, done = read(stdout, outline)
-    if done:
-      print outline
-      outline = ""
-    errline, done = read(stderr, errline)
-    if done:
-      print errline
-      errline = ""
-
-  # get tail
-  out, err = exe.communicate()
-  print outline + out.decode()
-  print errline + err.decode()
+    time.sleep(1)
+  debug("completed with exit code : %d" % exe.returncode)
   return exe.returncode
 
-def which(program):
-    def is_exe(fpath):
-        return os.path.isfile(fpath) and os.access(fpath, os.X_OK)
 
-    fpath, fname = os.path.split(program)
-    if fpath:
-        if is_exe(program):
-            return program
-    else:
-        for path in os.environ["PATH"].split(os.pathsep):
-            path = path.strip('"')
-            exe_file = os.path.join(path, program)
-            if is_exe(exe_file):
-                return exe_file
+def is_exe(fpath):
+  return os.path.isfile(fpath) and os.access(fpath, os.X_OK)
+
+def which(program):
+  
+  fpath, fname = os.path.split(program)
+  if fpath:
+    if is_exe(program):
+      return program
+  else:
+    for path in os.environ["PATH"].split(os.pathsep):
+      path = path.strip('"')
+      exe_file = os.path.join(path, program)
+      if is_exe(exe_file):
+        return exe_file
 
-    return None
+  return None
 
 def java(classname, args, classpath, jvm_opts_list):
   """