Merge pull request #2788 from ctiller/kinetic-lopsided-poetry

Testing port server

4 files changed, 183 insertions, 18 deletions

diff --git a/tools/run_tests/jobset.py b/tools/run_tests/jobset.py
index e5e778a3f1..b7e0089269 100755
--- a/tools/run_tests/jobset.py
+++ b/tools/run_tests/jobset.py
@@ -164,13 +164,15 @@ class JobSpec(object):
 class Job(object):
   """Manages one job."""
 
-  def __init__(self, spec, bin_hash, newline_on_success, travis, xml_report):
+  def __init__(self, spec, bin_hash, newline_on_success, travis, add_env, xml_report):
     self._spec = spec
     self._bin_hash = bin_hash
     self._tempfile = tempfile.TemporaryFile()
     env = os.environ.copy()
     for k, v in spec.environ.iteritems():
       env[k] = v
+    for k, v in add_env.iteritems():
+      env[k] = v
     self._start = time.time()
     self._process = subprocess.Popen(args=spec.cmdline,
                                      stderr=subprocess.STDOUT,
@@ -229,7 +231,7 @@ class Jobset(object):
   """Manages one run of jobs."""
 
   def __init__(self, check_cancelled, maxjobs, newline_on_success, travis,
-               stop_on_failure, cache, xml_report):
+               stop_on_failure, add_env, cache, xml_report):
     self._running = set()
     self._check_cancelled = check_cancelled
     self._cancelled = False
@@ -242,6 +244,7 @@ class Jobset(object):
     self._stop_on_failure = stop_on_failure
     self._hashes = {}
     self._xml_report = xml_report
+    self._add_env = add_env
 
   def start(self, spec):
     """Start a job. Return True on success, False on failure."""
@@ -264,16 +267,12 @@ class Jobset(object):
       bin_hash = None
       should_run = True
     if should_run:
-      try:
-        self._running.add(Job(spec,
-                              bin_hash,
-                              self._newline_on_success,
-                              self._travis,
-                              self._xml_report))
-      except:
-        message('FAILED', spec.shortname)
-        self._cancelled = True
-        return False
+      self._running.add(Job(spec,
+                            bin_hash,
+                            self._newline_on_success,
+                            self._travis,
+                            self._add_env,
+                            self._xml_report))
     return True
 
   def reap(self):
@@ -344,10 +343,11 @@ def run(cmdlines,
         infinite_runs=False,
         stop_on_failure=False,
         cache=None,
-        xml_report=None):
+        xml_report=None,
+        add_env={}):
   js = Jobset(check_cancelled,
               maxjobs if maxjobs is not None else _DEFAULT_MAX_JOBS,
-              newline_on_success, travis, stop_on_failure,
+              newline_on_success, travis, stop_on_failure, add_env,
               cache if cache is not None else NoCache(),
               xml_report)
   for cmdline in cmdlines:
diff --git a/tools/run_tests/port_server.py b/tools/run_tests/port_server.py
new file mode 100755
index 0000000000..0f81470d28
--- /dev/null
+++ b/tools/run_tests/port_server.py
@@ -0,0 +1,117 @@
+#!/usr/bin/env python
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+"""Manage TCP ports for unit tests; started by run_tests.py"""
+
+import argparse
+import BaseHTTPServer
+import hashlib
+import os
+import socket
+import sys
+import time
+
+argp = argparse.ArgumentParser(description='Server for httpcli_test')
+argp.add_argument('-p', '--port', default=12345, type=int)
+args = argp.parse_args()
+
+print 'port server running on port %d' % args.port
+
+pool = []
+in_use = {}
+
+with open(__file__) as f:
+  _MY_VERSION = hashlib.sha1(f.read()).hexdigest()
+
+
+def refill_pool():
+  """Scan for ports not marked for being in use"""
+  for i in range(10000, 65000):
+    if len(pool) > 100: break
+    if i in in_use:
+      age = time.time() - in_use[i]
+      if age < 600:
+        continue
+      del in_use[i]
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+      s.bind(('localhost', i))
+      pool.append(i)
+    except:
+      pass # we really don't care about failures
+    finally:
+      s.close()
+
+
+def allocate_port():
+  global pool
+  global in_use
+  if not pool:
+    refill_pool()
+  port = pool[0]
+  pool = pool[1:]
+  in_use[port] = time.time()
+  return port
+
+
+keep_running = True
+
+
+class Handler(BaseHTTPServer.BaseHTTPRequestHandler):
+
+  def do_GET(self):
+    global keep_running
+    if self.path == '/get':
+      # allocate a new port, it will stay bound for ten minutes and until
+      # it's unused
+      self.send_response(200)
+      self.send_header('Content-Type', 'text/plain')
+      self.end_headers()
+      p = allocate_port()
+      self.log_message('allocated port %d' % p)
+      self.wfile.write('%d' % p)
+    elif self.path == '/version':
+      # fetch a version string and the current process pid
+      self.send_response(200)
+      self.send_header('Content-Type', 'text/plain')
+      self.end_headers()
+      self.wfile.write(_MY_VERSION)
+    elif self.path == '/quit':
+      self.send_response(200)
+      self.end_headers()
+      keep_running = False
+
+
+httpd = BaseHTTPServer.HTTPServer(('', args.port), Handler)
+while keep_running:
+  httpd.handle_request()
+
+print 'done'
+
diff --git a/tools/run_tests/run_tests.py b/tools/run_tests/run_tests.py
index 0d4caa66aa..47d2b6183e 100755
--- a/tools/run_tests/run_tests.py
+++ b/tools/run_tests/run_tests.py
@@ -32,6 +32,7 @@
 
 import argparse
 import glob
+import hashlib
 import itertools
 import json
 import multiprocessing
@@ -43,6 +44,7 @@ import subprocess
 import sys
 import time
 import xml.etree.cElementTree as ET
+import urllib2
 
 import jobset
 import watch_dirs
@@ -530,7 +532,43 @@ class TestCache(object):
         self.parse(json.loads(f.read()))
 
 
-def _build_and_run(check_cancelled, newline_on_success, travis, cache, xml_report=None):
+def _start_port_server(port_server_port):
+  # check if a compatible port server is running
+  # if incompatible (version mismatch) ==> start a new one
+  # if not running ==> start a new one
+  # otherwise, leave it up
+  try:
+    version = urllib2.urlopen('http://localhost:%d/version' % port_server_port).read()
+    running = True
+  except Exception:
+    running = False
+  if running:
+    with open('tools/run_tests/port_server.py') as f:
+      current_version = hashlib.sha1(f.read()).hexdigest()
+      running = (version == current_version)
+      if not running:
+        urllib2.urlopen('http://localhost:%d/quit' % port_server_port).read()
+        time.sleep(1)
+  if not running:
+    port_log = open('portlog.txt', 'w')
+    port_server = subprocess.Popen(
+        ['python', 'tools/run_tests/port_server.py', '-p', '%d' % port_server_port],
+        stderr=subprocess.STDOUT,
+        stdout=port_log)
+    # ensure port server is up
+    while True:
+      try:
+        urllib2.urlopen('http://localhost:%d/get' % port_server_port).read()
+        break
+      except urllib2.URLError:
+        time.sleep(0.5)
+      except:
+        port_server.kill()
+        raise
+
+
+def _build_and_run(
+    check_cancelled, newline_on_success, travis, cache, xml_report=None):
   """Do one pass of building & running tests."""
   # build latest sequentially
   if not jobset.run(build_steps, maxjobs=1,
@@ -540,6 +578,8 @@ def _build_and_run(check_cancelled, newline_on_success, travis, cache, xml_repor
   # start antagonists
   antagonists = [subprocess.Popen(['tools/run_tests/antagonist.py'])
                  for _ in range(0, args.antagonists)]
+  port_server_port = 9999
+  _start_port_server(port_server_port)
   try:
     infinite_runs = runs_per_test == 0
     # When running on travis, we want out test runs to be as similar as possible
@@ -566,7 +606,8 @@ def _build_and_run(check_cancelled, newline_on_success, travis, cache, xml_repor
                       maxjobs=args.jobs,
                       stop_on_failure=args.stop_on_failure,
                       cache=cache if not xml_report else None,
-                      xml_report=testsuite):
+                      xml_report=testsuite,
+                      add_env={'GRPC_TEST_PORT_SERVER': 'localhost:%d' % port_server_port}):
       return 2
   finally:
     for antagonist in antagonists:
diff --git a/tools/run_tests/sources_and_headers.json b/tools/run_tests/sources_and_headers.json
index e3490c93af..5d23bf9e88 100644
--- a/tools/run_tests/sources_and_headers.json
+++ b/tools/run_tests/sources_and_headers.json
@@ -12252,7 +12252,6 @@
       "src/core/debug/trace.h", 
       "src/core/httpcli/format_request.h", 
       "src/core/httpcli/httpcli.h", 
-      "src/core/httpcli/httpcli_security_connector.h", 
       "src/core/httpcli/parser.h", 
       "src/core/iomgr/alarm.h", 
       "src/core/iomgr/alarm_heap.h", 
@@ -12410,7 +12409,6 @@
       "src/core/httpcli/httpcli.c", 
       "src/core/httpcli/httpcli.h", 
       "src/core/httpcli/httpcli_security_connector.c", 
-      "src/core/httpcli/httpcli_security_connector.h", 
       "src/core/httpcli/parser.c", 
       "src/core/httpcli/parser.h", 
       "src/core/iomgr/alarm.c", 
@@ -12723,6 +12721,9 @@
       "src/core/client_config/uri_parser.h", 
       "src/core/compression/message_compress.h", 
       "src/core/debug/trace.h", 
+      "src/core/httpcli/format_request.h", 
+      "src/core/httpcli/httpcli.h", 
+      "src/core/httpcli/parser.h", 
       "src/core/iomgr/alarm.h", 
       "src/core/iomgr/alarm_heap.h", 
       "src/core/iomgr/alarm_internal.h", 
@@ -12860,6 +12861,12 @@
       "src/core/compression/message_compress.h", 
       "src/core/debug/trace.c", 
       "src/core/debug/trace.h", 
+      "src/core/httpcli/format_request.c", 
+      "src/core/httpcli/format_request.h", 
+      "src/core/httpcli/httpcli.c", 
+      "src/core/httpcli/httpcli.h", 
+      "src/core/httpcli/parser.c", 
+      "src/core/httpcli/parser.h", 
       "src/core/iomgr/alarm.c", 
       "src/core/iomgr/alarm.h", 
       "src/core/iomgr/alarm_heap.c",