Adding a server side, non-blocking subprocess mechanism.

BUG=453679

Review URL: https://codereview.chromium.org/870103005

Cr-Commit-Position: refs/heads/master@{#316594}

testing/legion/client_lib.py

@@ -18,10 +18,8 @@
 #pylint: disable=relative-import
 import common_lib
 
-THIS_DIR = os.path.dirname(os.path.abspath(__file__))
-SWARMING_DIR = os.path.join(THIS_DIR, '..', '..', 'tools/swarming_client')
-ISOLATE_PY = os.path.join(SWARMING_DIR, 'isolate.py')
-SWARMING_PY = os.path.join(SWARMING_DIR, 'swarming.py')
+ISOLATE_PY = os.path.join(common_lib.SWARMING_DIR, 'isolate.py')
+SWARMING_PY = os.path.join(common_lib.SWARMING_DIR, 'swarming.py')
 
 
 class Error(Exception):

testing/legion/client_rpc_methods.py

@@ -4,39 +4,148 @@
 
 """Defines the client RPC methods."""
 
+import os
+import sys
 import logging
-import subprocess
 import threading
 
+#pylint: disable=relative-import
+import common_lib
+
+# Map swarming_client to use subprocess42
+sys.path.append(common_lib.SWARMING_DIR)
+
+from utils import subprocess42
+
 
 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)
 
   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 a key returned to
+  the caller at the time of creation.
+  """
+
+  _processes = {}
+  _process_next_id = 0
+  _creation_lock = threading.Lock()
+
+  def __init__(self, cmd):
+    self.proc = subprocess42.Popen(cmd, stdout=subprocess42.PIPE,
+                                   stderr=subprocess42.PIPE)
+    self.stdout = ''
+    self.stderr = ''
+    self.data_lock = threading.Lock()
+    threading.Thread(target=self._run).start()
+
+  def _run(self):
+    for pipe, data in self.proc.yield_any():
+      with self.data_lock:
+        if pipe == 'stdout':
+          self.stdout += data
+        else:
+          self.stderr += data
+
+  @classmethod
+  def Popen(cls, cmd):
+    with cls._creation_lock:
+      key = 'Process%d' % cls._process_next_id
+      cls._process_next_id += 1
+    logging.debug('Creating process %s', key)
+    process = cls(cmd)
+    cls._processes[key] = process
+    return key
+
+  @classmethod
+  def Terminate(cls, key):
+    logging.debug('Terminating and deleting process %s', key)
+    return cls._processes.pop(key).proc.terminate()
+
+  @classmethod
+  def Kill(cls, key):
+    logging.debug('Killing and deleting process %s', key)
+    return cls._processes.pop(key).proc.kill()
+
+  @classmethod
+  def Delete(cls, key):
+    logging.debug('Deleting process %s', key)
+    cls._processes.pop(key)
+
+  @classmethod
+  def GetReturncode(cls, key):
+    return cls._processes[key].proc.returncode
+
+  @classmethod
+  def ReadStdout(cls, key):
+    """Returns all stdout since the last call to ReadStdout.
+
+    This call allows the user to read stdout while the process is running.
+    However each call will flush the local stdout buffer. In order to make
+    multiple calls to ReadStdout and to retain the entire output the results
+    of this call will need to be buffered in the calling code.
+    """
+    proc = cls._processes[key]
+    with proc.data_lock:
+      # Perform a "read" on the stdout data
+      stdout = proc.stdout
+      proc.stdout = ''
+    return stdout
+
+  @classmethod
+  def ReadStderr(cls, key):
+    """Returns all stderr read since the last call to ReadStderr.
+
+    See ReadStdout for additional details.
+    """
+    proc = cls._processes[key]
+    with proc.data_lock:
+      # Perform a "read" on the stderr data
+      stderr = proc.stderr
+      proc.stderr = ''
+    return stderr
+
+  @classmethod
+  def Wait(cls, key):
+    return cls._processes[key].proc.wait()
+
+  @classmethod
+  def Poll(cls, key):
+    return cls._processes[key].proc.poll()
+
+  @classmethod
+  def GetPid(cls, key):
+    return cls._processes[key].proc.pid

testing/legion/common_lib.py

@@ -6,6 +6,7 @@
 
 import argparse
 import logging
+import os
 import socket
 import xmlrpclib
 
@@ -14,6 +15,8 @@
 SERVER_ADDRESS = ''
 SERVER_PORT = 31710
 DEFAULT_TIMEOUT_SECS = 20 * 60  # 30 minutes
+THIS_DIR = os.path.dirname(os.path.abspath(__file__))
+SWARMING_DIR = os.path.join(THIS_DIR, '..', '..', 'tools', 'swarming_client')
 
 
 def InitLogging():

testing/legion/examples/hello_world/host_test.py

@@ -60,13 +60,16 @@ def Task(self):
   def CallEcho(self, client):
     """Call rpc.Echo on a client."""
     logging.info('Calling Echo on %s', client.name)
-    logging.info(self.client1.rpc.Echo(client.name))
+    logging.info(client.rpc.Echo(client.name))
 
   def CallClientTest(self, client):
     """Call client_test.py name on a client."""
     logging.info('Calling Subprocess to run "./client_test.py %s"', client.name)
-    retcode, stdout, stderr = client.rpc.Subprocess(
-        ['./client_test.py', client.name])
+    proc = client.rpc.subprocess.Popen(['./client_test.py', client.name])
+    client.rpc.subprocess.Wait(proc)
+    retcode = client.rpc.subprocess.GetReturncode(proc)
+    stdout = client.rpc.subprocess.ReadStdout(proc)
+    stderr = client.rpc.subprocess.ReadStderr(proc)
     logging.info('retcode: %s, stdout: %s, stderr: %s', retcode, stdout, stderr)
 
 

testing/legion/examples/subprocess/client.isolate

@@ -0,0 +1,22 @@
+# Copyright 2015 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.
+
+{
+  'includes': [
+    '../../legion.isolate'
+  ],
+  'conditions': [
+    ['multi_machine == 1', {
+      'variables': {
+        'command': [
+          'python',
+          '../../client_controller.py',
+        ],
+        'files': [
+          'client.isolate'
+        ],
+      },
+    }],
+  ],
+}

testing/legion/examples/subprocess/subprocess_test.isolate

@@ -0,0 +1,22 @@
+# Copyright 2015 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.
+
+{
+  'includes': [
+    '../../legion.isolate',
+    'client.isolate'
+  ],
+  'conditions': [
+    ['multi_machine == 1', {
+      'variables': {
+        'command': [
+          'subprocess_test.py',
+        ],
+        'files': [
+          'subprocess_test.py',
+        ],
+      },
+    }],
+  ]
+}

testing/legion/examples/subprocess/subprocess_test.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python
+# Copyright 2015 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.
+
+"""A host test module demonstrating interacting with remote subprocesses."""
+
+# Map the legion directory so we can import the host controller.
+import sys
+sys.path.append('../../')
+
+import logging
+import time
+import xmlrpclib
+
+import host_controller
+
+
+class ExampleController(host_controller.HostController):
+  """An example controller using the remote subprocess functions."""
+
+  def __init__(self):
+    super(ExampleController, self).__init__()
+    self.client = None
+
+  def SetUp(self):
+    """Creates the client machine and waits until it connects."""
+    self.client = self.NewClient(
+        isolate_file='client.isolate',
+        config_vars={'multi_machine': '1'},
+        dimensions={'os': 'legion-linux'},
+        idle_timeout_secs=90, connection_timeout_secs=90,
+        verbosity=logging.DEBUG)
+    self.client.Create()
+    self.client.WaitForConnection()
+
+  def Task(self):
+    """Main method to run the task code."""
+    self.TestLs()
+    self.TestTerminate()
+    self.TestMultipleProcesses()
+
+  def TestMultipleProcesses(self):
+    start = time.time()
+
+    sleep20 = self.client.rpc.subprocess.Popen(['sleep', '20'])
+    sleep10 = self.client.rpc.subprocess.Popen(['sleep', '10'])
+
+    self.client.rpc.subprocess.Wait(sleep10)
+    elapsed = time.time() - start
+    assert elapsed >= 10 and elapsed < 11
+
+    self.client.rpc.subprocess.Wait(sleep20)
+    elapsed = time.time() - start
+    assert elapsed >= 20
+
+    self.client.rpc.subprocess.Delete(sleep20)
+    self.client.rpc.subprocess.Delete(sleep10)
+
+  def TestTerminate(self):
+    start = time.time()
+    proc = self.client.rpc.subprocess.Popen(['sleep', '20'])
+    self.client.rpc.subprocess.Terminate(proc)  # Implicitly deleted
+    try:
+      self.client.rpc.subprocess.Wait(proc)
+    except xmlrpclib.Fault:
+      pass
+    assert time.time() - start < 20
+
+  def TestLs(self):
+    proc = self.client.rpc.subprocess.Popen(['ls'])
+    self.client.rpc.subprocess.Wait(proc)
+    assert self.client.rpc.subprocess.GetReturncode(proc) == 0
+    assert 'client.isolate' in self.client.rpc.subprocess.ReadStdout(proc)
+    self.client.rpc.subprocess.Delete(proc)
+
+
+if __name__ == '__main__':
+  ExampleController().RunController()