Better port selection

- avoid IANA and Linux ephemeral port ranges
- support dropping allocated ports
- aggressively try to reclaim ports if we reach exhaustion
- set SO_REUSEADDR on test port binds

test/core/util/port_posix.c

@@ -47,6 +47,7 @@
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
 
 #include "src/core/httpcli/httpcli.h"
 #include "src/core/support/env.h"
@@ -66,7 +67,71 @@ static int has_port_been_chosen(int port) {
   return 0;
 }
 
-static void free_chosen_ports() { gpr_free(chosen_ports); }
+typedef struct freereq {
+  grpc_pollset pollset;
+  int done;
+} freereq;
+
+static void destroy_pollset_and_shutdown(void *p) {
+  grpc_pollset_destroy(p);
+  grpc_shutdown();
+}
+
+static void freed_port_from_server(void *arg,
+                                   const grpc_httpcli_response *response) {
+  freereq *pr = arg;
+  GPR_ASSERT(response);
+  GPR_ASSERT(response->status == 200);
+  gpr_mu_lock(GRPC_POLLSET_MU(&pr->pollset));
+  pr->done = 1;
+  grpc_pollset_kick(&pr->pollset, NULL);
+  gpr_mu_unlock(GRPC_POLLSET_MU(&pr->pollset));
+}
+
+static void free_port_using_server(char *server, int port) {
+  grpc_httpcli_context context;
+  grpc_httpcli_request req;
+  freereq pr;
+  char *path;
+
+  grpc_init();
+
+  memset(&pr, 0, sizeof(pr));
+  memset(&req, 0, sizeof(req));
+  grpc_pollset_init(&pr.pollset);
+
+  req.host = server;
+  gpr_asprintf(&path, "/drop/%d", port);
+  req.path = path;
+
+  grpc_httpcli_context_init(&context);
+  grpc_httpcli_get(&context, &pr.pollset, &req,
+                   GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), freed_port_from_server,
+                   &pr);
+  gpr_mu_lock(GRPC_POLLSET_MU(&pr.pollset));
+  while (!pr.done) {
+    grpc_pollset_worker worker;
+    grpc_pollset_work(&pr.pollset, &worker, gpr_now(GPR_CLOCK_MONOTONIC),
+                      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1));
+  }
+  gpr_mu_unlock(GRPC_POLLSET_MU(&pr.pollset));
+
+  grpc_httpcli_context_destroy(&context);
+  grpc_pollset_shutdown(&pr.pollset, destroy_pollset_and_shutdown, &pr.pollset);
+  gpr_free(path);
+}
+
+static void free_chosen_ports() { 
+  char *env = gpr_getenv("GRPC_TEST_PORT_SERVER");
+  if (env != NULL) {
+    size_t i;
+    for (i = 0; i < num_chosen_ports; i++) {
+      free_port_using_server(env, chosen_ports[i]);
+    }
+  }
+
+  gpr_free(chosen_ports); 
+}
 
 static void chose_port(int port) {
   if (chosen_ports == NULL) {
@@ -151,11 +216,6 @@ static void got_port_from_server(void *arg,
   gpr_mu_unlock(GRPC_POLLSET_MU(&pr->pollset));
 }
 
-static void destroy_pollset_and_shutdown(void *p) {
-  grpc_pollset_destroy(p);
-  grpc_shutdown();
-}
-
 static int pick_port_using_server(char *server) {
   grpc_httpcli_context context;
   grpc_httpcli_request req;
@@ -211,8 +271,9 @@ int grpc_pick_unused_port(void) {
     int port = pick_port_using_server(env);
     gpr_free(env);
     if (port != 0) {
-      return port;
+      chose_port(port);
     }
+    return port;
   }
 
   for (;;) {

tools/run_tests/port_server.py

@@ -37,6 +37,7 @@ import os
 import socket
 import sys
 import time
+import yaml
 
 argp = argparse.ArgumentParser(description='Server for httpcli_test')
 argp.add_argument('-p', '--port', default=12345, type=int)
@@ -51,16 +52,17 @@ with open(__file__) as f:
   _MY_VERSION = hashlib.sha1(f.read()).hexdigest()
 
 
-def refill_pool():
+def refill_pool(max_timeout):
   """Scan for ports not marked for being in use"""
-  for i in range(10000, 65000):
+  for i in range(1025, 32767):
     if len(pool) > 100: break
     if i in in_use:
       age = time.time() - in_use[i]
-      if age < 600:
+      if age < max_timeout:
         continue
       del in_use[i]
     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     try:
       s.bind(('localhost', i))
       pool.append(i)
@@ -73,8 +75,12 @@ def refill_pool():
 def allocate_port():
   global pool
   global in_use
-  if not pool:
-    refill_pool()
+  max_timeout = 600
+  while not pool:
+    refill_pool(max_timeout)
+    if not pool:
+      time.sleep(1)
+      max_timeout /= 2
   port = pool[0]
   pool = pool[1:]
   in_use[port] = time.time()
@@ -97,12 +103,26 @@ class Handler(BaseHTTPServer.BaseHTTPRequestHandler):
       p = allocate_port()
       self.log_message('allocated port %d' % p)
       self.wfile.write('%d' % p)
+    elif self.path[0:6] == '/drop/':
+      self.send_response(200)
+      self.send_header('Content-Type', 'text/plain')
+      self.end_headers()
+      p = int(self.path[6:])
+      del in_use[p]
+      pool.append(p)
+      self.log_message('drop port %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 == '/dump':
+      self.send_response(200)
+      self.send_header('Content-Type', 'text/plain')
+      self.end_headers()
+      now = time.time()
+      self.wfile.write(yaml.dump({'pool': pool, 'in_use': dict((k, now - v) for k, v in in_use.iteritems())}))
     elif self.path == '/quit':
       self.send_response(200)
       self.end_headers()