Testing port server

run_tests.py will start a server (if it's not running, or if the running
port server mismatches the 'current' one) that serves ports to use for
tests. The server is left running after run_tests.py finishes, so that
in environments such as Mac and Windows where tests run unshielded from
each other, we don't start jumping on already used ports.

5 files changed, 178 insertions, 25 deletions

diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal
index cce0b6fed0..abe1f72d36 100644
--- a/tools/doxygen/Doxyfile.core.internal
+++ b/tools/doxygen/Doxyfile.core.internal
@@ -767,10 +767,6 @@ include/grpc/compression.h \
 include/grpc/grpc.h \
 include/grpc/status.h \
 include/grpc/census.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/security/auth_filters.h \
 src/core/security/base64.h \
 src/core/security/credentials.h \
@@ -810,6 +806,9 @@ src/core/client_config/subchannel_factory_decorators/merge_channel_args.h \
 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 \
@@ -883,10 +882,7 @@ src/core/transport/transport.h \
 src/core/transport/transport_impl.h \
 src/core/census/context.h \
 src/core/census/rpc_stat_id.h \
-src/core/httpcli/format_request.c \
-src/core/httpcli/httpcli.c \
 src/core/httpcli/httpcli_security_connector.c \
-src/core/httpcli/parser.c \
 src/core/security/base64.c \
 src/core/security/client_auth_filter.c \
 src/core/security/credentials.c \
@@ -933,6 +929,9 @@ src/core/client_config/uri_parser.c \
 src/core/compression/algorithm.c \
 src/core/compression/message_compress.c \
 src/core/debug/trace.c \
+src/core/httpcli/format_request.c \
+src/core/httpcli/httpcli.c \
+src/core/httpcli/parser.c \
 src/core/iomgr/alarm.c \
 src/core/iomgr/alarm_heap.c \
 src/core/iomgr/endpoint.c \
diff --git a/tools/run_tests/jobset.py b/tools/run_tests/jobset.py
index ec25b47610..0318e357b8 100755
--- a/tools/run_tests/jobset.py
+++ b/tools/run_tests/jobset.py
@@ -162,13 +162,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,
@@ -227,7 +229,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
@@ -240,6 +242,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."""
@@ -262,16 +265,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):
@@ -342,10 +341,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..41f862ad88
--- /dev/null
+++ b/tools/run_tests/port_server.py
@@ -0,0 +1,105 @@
+#!/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(sys.argv[0]) 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
+
+
+class Handler(BaseHTTPServer.BaseHTTPRequestHandler):
+
+  def do_GET(self):
+    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_and_pid':
+      # 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('%s+%d' % (_MY_VERSION, os.getpid()))
+
+
+BaseHTTPServer.HTTPServer(('', args.port), Handler).serve_forever()
+
diff --git a/tools/run_tests/run_tests.py b/tools/run_tests/run_tests.py
index fa749498d2..653b98d57d 100755
--- a/tools/run_tests/run_tests.py
+++ b/tools/run_tests/run_tests.py
@@ -32,17 +32,20 @@
 
 import argparse
 import glob
+import hashlib
 import itertools
 import json
 import multiprocessing
 import os
 import platform
+import psutil
 import random
 import re
 import subprocess
 import sys
 import time
 import xml.etree.cElementTree as ET
+import urllib2
 
 import jobset
 import watch_dirs
@@ -522,7 +525,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, _, pid = urllib2.urlopen(
+        'http://localhost:%d/version_and_pid' % port_server_port).read().partition('+')
+    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:
+        psutil.Process(int(pid)).terminate()
+  if not running:
+    port_log = open('portlog.txt', 'w')
+    port_server = subprocess.Popen(
+        ['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,
@@ -532,6 +571,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
@@ -558,7 +599,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 d2cf07f197..a8bc1a615c 100644
--- a/tools/run_tests/sources_and_headers.json
+++ b/tools/run_tests/sources_and_headers.json
@@ -10955,7 +10955,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", 
@@ -11114,7 +11113,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", 
@@ -11423,6 +11421,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", 
@@ -11561,6 +11562,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",