Adding a server side, non-blocking subprocess mechanism.

testing/legion/client_rpc_methods.py

Index: testing/legion/client_rpc_methods.py
diff --git a/testing/legion/client_rpc_methods.py b/testing/legion/client_rpc_methods.py
index 24a552ecb5d600cb5a41d7621d2957bffe78623e..7ed6fa20698c7c7e62a12f3b124cb50077d080e5 100644
--- a/testing/legion/client_rpc_methods.py
+++ b/testing/legion/client_rpc_methods.py
@@ -12,31 +12,91 @@ import threading
 class RPCMethods(object):
   """Class exposing RPC methods."""
 
+  _dotted_whitelist = ['subprocess']
+
   def __init__(self, server):
-    self.server = server
+    self._server = server
+    self.subprocess = Subprocess
+
+  def _dispatch(self, method, params):
+    obj = self
+    if '.' in method:
+      # Allow only white listed dotted names
+      name, method = method.split('.')
+      assert name in self._dotted_whitelist
+      obj = getattr(self, name)
+    return getattr(obj, method)(*params)

[M-A Ruel, 2015/02/10 19:52:19]

I generally prefer to prefix all the rpc methods with something like "rpc_" or
anything similar. This makes it clear which methods are RPCs and secures the
interface.

[Mike Meade, 2015/02/11 00:25:34]

With SimpleXMLRPCServer any public method is considered an RPC method, and any
"protected" method is not exposed. In the actual client code these are accessed
via an rpc property, effectively making the calls client.rpc.Echo,
client.rpc.subprocess.Popen, etc. This seems to be very similar to what you are
proposing. Feel free to correct me if I misunderstood your comment.

 
   def Echo(self, message):
     """Simple RPC method to print and return a message."""
     logging.info('Echoing %s', message)
     return 'echo %s' % str(message)
 
-  def Subprocess(self, cmd):
-    """Run the commands in a subprocess.
-
-    Returns:
-      (returncode, stdout, stderr).
-    """
-    p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
-                         stderr=subprocess.PIPE)
-    stdout, stderr = p.communicate()
-    return (p.returncode, stdout, stderr)
-
   def Quit(self):
-    """Call server.shutdown in another thread.
+    """Call _server.shutdown in another thread.
 
     This is needed because server.shutdown waits for the server to actually
     quit. However the server cannot shutdown until it completes handling this
     call. Calling this in the same thread results in a deadlock.
     """
-    t = threading.Thread(target=self.server.shutdown)
+    t = threading.Thread(target=self._server.shutdown)
     t.start()
+
+
+class Subprocess(object):
+  """Implements a server-based non-blocking subprocess.
+
+  This non-blocking subprocess allows the caller to continue operating while
+  also able to interact with this subprocess based on an index returned to
+  the caller at the time of creation.
+  """
+
+  _processes = []
+
+  def __init__(self, cmd):
+    self.proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+                                 stderr=subprocess.PIPE)
+    self.stdout = ''
+    self.stderr = ''
+    threading.Thread(target=self._run).start()
+
+  def _run(self):
+    self.stdout, self.stderr = self.proc.communicate()

[M-A Ruel, 2015/02/10 19:52:19]

Note that you may like subprocess42.py, which supports streaming.
https://code.google.com/p/swarming/source/browse/utils/subprocess42.py?repo=c...

[Mike Meade, 2015/02/11 00:25:34]

I ended up using subprocess42, but mainly for the added Windows support. The
problem is that the host controller is expected to run multiple subprocesses at
a time, and using the streaming means that they would need to implement the
reader threads on the host controller side. However if it proves to be a feature
request I can always add it since I am using subprocess42.

[Mike Meade, 2015/02/12 21:46:59]

The need for streaming reads actually came up pretty quickly. I added the
ability to do streaming reads and writes from the host controller without the
client controller blocking due to full pipes.

+
+  @classmethod
+  def Popen(cls, cmd):
+    process = cls(cmd)
+    cls._processes.append(process)
+    return len(cls._processes) - 1
+
+  @classmethod
+  def Terminate(cls, index):

[M-A Ruel, 2015/02/10 19:52:19]

This index based access is surprising, especially that it means it's a memory
leak long term.

[Mike Meade, 2015/02/11 00:25:34]

Changed this to a dictionary, added a Delete method, and made the Terminate and
Kill methods delete the object. There is still a chance for a memory leak, but
at least the test case author is given multiple tools to manage this.

+    return cls._processes[index].proc.terminate()
+
+  @classmethod
+  def Kill(cls, index):
+    return cls._processes[index].proc.kill()
+
+  @classmethod
+  def GetReturncode(cls, index):
+    return cls._processes[index].proc.returncode
+
+  @classmethod
+  def GetStdout(cls, index):
+    return cls._processes[index].stdout
+
+  @classmethod
+  def GetStderr(cls, index):
+    return cls._processes[index].stderr
+
+  @classmethod
+  def Wait(cls, index):
+    return cls._processes[index].proc.wait()
+
+  @classmethod
+  def Poll(cls, index):
+    return cls._processes[index].proc.poll()
+
+  @classmethod
+  def GetPid(cls, index):
+    return cls._processes[index].proc.pid