Merge github.com:grpc/grpc into pid_ctl

9 files changed, 223 insertions, 30 deletions

diff --git a/test/core/util/mock_endpoint.c b/test/core/util/mock_endpoint.c
index 13e0e918fb..2b041a4484 100644
--- a/test/core/util/mock_endpoint.c
+++ b/test/core/util/mock_endpoint.c
@@ -33,16 +33,20 @@
 
 #include "test/core/util/mock_endpoint.h"
 
+#include <inttypes.h>
+
 #include <grpc/support/alloc.h>
 #include <grpc/support/string_util.h>
 
 typedef struct grpc_mock_endpoint {
   grpc_endpoint base;
   gpr_mu mu;
+  int refs;
   void (*on_write)(gpr_slice slice);
   gpr_slice_buffer read_buffer;
   gpr_slice_buffer *on_read_out;
   grpc_closure *on_read;
+  grpc_resource_user resource_user;
 } grpc_mock_endpoint;
 
 static void me_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
@@ -74,6 +78,24 @@ static void me_add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
 static void me_add_to_pollset_set(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
                                   grpc_pollset_set *pollset) {}
 
+static void unref(grpc_exec_ctx *exec_ctx, grpc_mock_endpoint *m) {
+  gpr_mu_lock(&m->mu);
+  if (0 == --m->refs) {
+    gpr_mu_unlock(&m->mu);
+    gpr_slice_buffer_destroy(&m->read_buffer);
+    grpc_resource_user_destroy(exec_ctx, &m->resource_user);
+    gpr_free(m);
+  } else {
+    gpr_mu_unlock(&m->mu);
+  }
+}
+
+static void me_finish_shutdown(grpc_exec_ctx *exec_ctx, void *me,
+                               grpc_error *error) {
+  grpc_mock_endpoint *m = me;
+  unref(exec_ctx, m);
+}
+
 static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   grpc_mock_endpoint *m = (grpc_mock_endpoint *)ep;
   gpr_mu_lock(&m->mu);
@@ -82,19 +104,25 @@ static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
                         GRPC_ERROR_CREATE("Endpoint Shutdown"), NULL);
     m->on_read = NULL;
   }
+  grpc_resource_user_shutdown(exec_ctx, &m->resource_user,
+                              grpc_closure_create(me_finish_shutdown, m));
   gpr_mu_unlock(&m->mu);
 }
 
 static void me_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   grpc_mock_endpoint *m = (grpc_mock_endpoint *)ep;
-  gpr_slice_buffer_destroy(&m->read_buffer);
-  gpr_free(m);
+  unref(exec_ctx, m);
 }
 
 static char *me_get_peer(grpc_endpoint *ep) {
   return gpr_strdup("fake:mock_endpoint");
 }
 
+static grpc_resource_user *me_get_resource_user(grpc_endpoint *ep) {
+  grpc_mock_endpoint *m = (grpc_mock_endpoint *)ep;
+  return &m->resource_user;
+}
+
 static grpc_workqueue *me_get_workqueue(grpc_endpoint *ep) { return NULL; }
 
 static const grpc_endpoint_vtable vtable = {
@@ -105,12 +133,19 @@ static const grpc_endpoint_vtable vtable = {
     me_add_to_pollset_set,
     me_shutdown,
     me_destroy,
+    me_get_resource_user,
     me_get_peer,
 };
 
-grpc_endpoint *grpc_mock_endpoint_create(void (*on_write)(gpr_slice slice)) {
+grpc_endpoint *grpc_mock_endpoint_create(void (*on_write)(gpr_slice slice),
+                                         grpc_resource_quota *resource_quota) {
   grpc_mock_endpoint *m = gpr_malloc(sizeof(*m));
   m->base.vtable = &vtable;
+  m->refs = 2;
+  char *name;
+  gpr_asprintf(&name, "mock_endpoint_%" PRIxPTR, (intptr_t)m);
+  grpc_resource_user_init(&m->resource_user, resource_quota, name);
+  gpr_free(name);
   gpr_slice_buffer_init(&m->read_buffer);
   gpr_mu_init(&m->mu);
   m->on_write = on_write;
diff --git a/test/core/util/mock_endpoint.h b/test/core/util/mock_endpoint.h
index 051af9866b..b3a464ca01 100644
--- a/test/core/util/mock_endpoint.h
+++ b/test/core/util/mock_endpoint.h
@@ -36,7 +36,8 @@
 
 #include "src/core/lib/iomgr/endpoint.h"
 
-grpc_endpoint *grpc_mock_endpoint_create(void (*on_write)(gpr_slice slice));
+grpc_endpoint *grpc_mock_endpoint_create(void (*on_write)(gpr_slice slice),
+                                         grpc_resource_quota *resource_quota);
 void grpc_mock_endpoint_put_read(grpc_exec_ctx *exec_ctx,
                                  grpc_endpoint *mock_endpoint, gpr_slice slice);
 
diff --git a/test/core/util/passthru_endpoint.c b/test/core/util/passthru_endpoint.c
index 7ed9e97bd6..ee6ef7da60 100644
--- a/test/core/util/passthru_endpoint.c
+++ b/test/core/util/passthru_endpoint.c
@@ -33,6 +33,8 @@
 
 #include "test/core/util/passthru_endpoint.h"
 
+#include <inttypes.h>
+
 #include <grpc/support/alloc.h>
 #include <grpc/support/string_util.h>
 
@@ -44,6 +46,7 @@ typedef struct {
   gpr_slice_buffer read_buffer;
   gpr_slice_buffer *on_read_out;
   grpc_closure *on_read;
+  grpc_resource_user resource_user;
 } half;
 
 struct passthru_endpoint {
@@ -122,7 +125,8 @@ static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   gpr_mu_unlock(&m->parent->mu);
 }
 
-static void me_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
+static void me_really_destroy(grpc_exec_ctx *exec_ctx, void *ep,
+                              grpc_error *error) {
   passthru_endpoint *p = ((half *)ep)->parent;
   gpr_mu_lock(&p->mu);
   if (0 == --p->halves) {
@@ -136,12 +140,23 @@ static void me_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   }
 }
 
+static void me_destroy(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
+  half *m = (half *)ep;
+  grpc_resource_user_shutdown(exec_ctx, &m->resource_user,
+                              grpc_closure_create(me_really_destroy, m));
+}
+
 static char *me_get_peer(grpc_endpoint *ep) {
   return gpr_strdup("fake:mock_endpoint");
 }
 
 static grpc_workqueue *me_get_workqueue(grpc_endpoint *ep) { return NULL; }
 
+static grpc_resource_user *me_get_resource_user(grpc_endpoint *ep) {
+  half *m = (half *)ep;
+  return &m->resource_user;
+}
+
 static const grpc_endpoint_vtable vtable = {
     me_read,
     me_write,
@@ -150,23 +165,32 @@ static const grpc_endpoint_vtable vtable = {
     me_add_to_pollset_set,
     me_shutdown,
     me_destroy,
+    me_get_resource_user,
     me_get_peer,
 };
 
-static void half_init(half *m, passthru_endpoint *parent) {
+static void half_init(half *m, passthru_endpoint *parent,
+                      grpc_resource_quota *resource_quota,
+                      const char *half_name) {
   m->base.vtable = &vtable;
   m->parent = parent;
   gpr_slice_buffer_init(&m->read_buffer);
   m->on_read = NULL;
+  char *name;
+  gpr_asprintf(&name, "passthru_endpoint_%s_%" PRIxPTR, half_name,
+               (intptr_t)parent);
+  grpc_resource_user_init(&m->resource_user, resource_quota, name);
+  gpr_free(name);
 }
 
 void grpc_passthru_endpoint_create(grpc_endpoint **client,
-                                   grpc_endpoint **server) {
+                                   grpc_endpoint **server,
+                                   grpc_resource_quota *resource_quota) {
   passthru_endpoint *m = gpr_malloc(sizeof(*m));
   m->halves = 2;
   m->shutdown = 0;
-  half_init(&m->client, m);
-  half_init(&m->server, m);
+  half_init(&m->client, m, resource_quota, "client");
+  half_init(&m->server, m, resource_quota, "server");
   gpr_mu_init(&m->mu);
   *client = &m->client.base;
   *server = &m->server.base;
diff --git a/test/core/util/passthru_endpoint.h b/test/core/util/passthru_endpoint.h
index aa1d3a1763..b81ac5571c 100644
--- a/test/core/util/passthru_endpoint.h
+++ b/test/core/util/passthru_endpoint.h
@@ -37,6 +37,7 @@
 #include "src/core/lib/iomgr/endpoint.h"
 
 void grpc_passthru_endpoint_create(grpc_endpoint **client,
-                                   grpc_endpoint **server);
+                                   grpc_endpoint **server,
+                                   grpc_resource_quota *resource_quota);
 
 #endif
diff --git a/test/core/util/port_posix.c b/test/core/util/port_posix.c
index 265e0acee1..60537b4946 100644
--- a/test/core/util/port_posix.c
+++ b/test/core/util/port_posix.c
@@ -31,9 +31,9 @@
  *
  */
 
-#include <grpc/support/port_platform.h>
+#include "src/core/lib/iomgr/port.h"
 #include "test/core/util/test_config.h"
-#if defined(GPR_POSIX_SOCKET) && defined(GRPC_TEST_PICK_PORT)
+#if defined(GRPC_POSIX_SOCKET) && defined(GRPC_TEST_PICK_PORT)
 
 #include "test/core/util/port.h"
 
@@ -237,4 +237,4 @@ int grpc_pick_unused_port_or_die(void) {
 
 void grpc_recycle_unused_port(int port) { GPR_ASSERT(free_chosen_port(port)); }
 
-#endif /* GPR_POSIX_SOCKET && GRPC_TEST_PICK_PORT */
+#endif /* GRPC_POSIX_SOCKET && GRPC_TEST_PICK_PORT */
diff --git a/test/core/util/port_server_client.c b/test/core/util/port_server_client.c
index a5c8c49650..b2342feeb4 100644
--- a/test/core/util/port_server_client.c
+++ b/test/core/util/port_server_client.c
@@ -99,9 +99,12 @@ void grpc_free_port_using_server(char *server, int port) {
   req.http.path = path;
 
   grpc_httpcli_context_init(&context);
-  grpc_httpcli_get(&exec_ctx, &context, &pr.pops, &req,
+  grpc_resource_quota *resource_quota =
+      grpc_resource_quota_create("port_server_client/free");
+  grpc_httpcli_get(&exec_ctx, &context, &pr.pops, resource_quota, &req,
                    GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
                    grpc_closure_create(freed_port_from_server, &pr), &rsp);
+  grpc_resource_quota_internal_unref(&exec_ctx, resource_quota);
   gpr_mu_lock(pr.mu);
   while (!pr.done) {
     grpc_pollset_worker *worker = NULL;
@@ -167,10 +170,13 @@ static void got_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,
     req.http.path = "/get";
     grpc_http_response_destroy(&pr->response);
     memset(&pr->response, 0, sizeof(pr->response));
-    grpc_httpcli_get(exec_ctx, pr->ctx, &pr->pops, &req,
+    grpc_resource_quota *resource_quota =
+        grpc_resource_quota_create("port_server_client/pick_retry");
+    grpc_httpcli_get(exec_ctx, pr->ctx, &pr->pops, resource_quota, &req,
                      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
                      grpc_closure_create(got_port_from_server, pr),
                      &pr->response);
+    grpc_resource_quota_internal_unref(exec_ctx, resource_quota);
     return;
   }
   GPR_ASSERT(response);
@@ -211,9 +217,13 @@ int grpc_pick_port_using_server(char *server) {
   req.http.path = "/get";
 
   grpc_httpcli_context_init(&context);
-  grpc_httpcli_get(
-      &exec_ctx, &context, &pr.pops, &req, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
-      grpc_closure_create(got_port_from_server, &pr), &pr.response);
+  grpc_resource_quota *resource_quota =
+      grpc_resource_quota_create("port_server_client/pick");
+  grpc_httpcli_get(&exec_ctx, &context, &pr.pops, resource_quota, &req,
+                   GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
+                   grpc_closure_create(got_port_from_server, &pr),
+                   &pr.response);
+  grpc_resource_quota_internal_unref(&exec_ctx, resource_quota);
   grpc_exec_ctx_finish(&exec_ctx);
   gpr_mu_lock(pr.mu);
   while (pr.port == -1) {
diff --git a/test/core/util/port_uv.c b/test/core/util/port_uv.c
new file mode 100644
index 0000000000..0c9c0d87d6
--- /dev/null
+++ b/test/core/util/port_uv.c
@@ -0,0 +1,117 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#include "src/core/lib/iomgr/port.h"
+#include "test/core/util/test_config.h"
+#if defined(GRPC_UV) && defined(GRPC_TEST_PICK_PORT)
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+
+#include "src/core/lib/support/env.h"
+#include "test/core/util/port.h"
+#include "test/core/util/port_server_client.h"
+
+// Almost everything in this file has been copied from port_posix.c
+
+static int *chosen_ports = NULL;
+static size_t num_chosen_ports = 0;
+
+static int free_chosen_port(int port) {
+  size_t i;
+  int found = 0;
+  size_t found_at = 0;
+  char *env = gpr_getenv("GRPC_TEST_PORT_SERVER");
+  /* Find the port and erase it from the list, then tell the server it can be
+     freed. */
+  for (i = 0; i < num_chosen_ports; i++) {
+    if (chosen_ports[i] == port) {
+      GPR_ASSERT(found == 0);
+      found = 1;
+      found_at = i;
+    }
+  }
+  if (found) {
+    chosen_ports[found_at] = chosen_ports[num_chosen_ports - 1];
+    num_chosen_ports--;
+    if (env) {
+      grpc_free_port_using_server(env, port);
+    }
+  }
+  gpr_free(env);
+  return found;
+}
+
+static void free_chosen_ports(void) {
+  char *env = gpr_getenv("GRPC_TEST_PORT_SERVER");
+  if (env != NULL) {
+    size_t i;
+    for (i = 0; i < num_chosen_ports; i++) {
+      grpc_free_port_using_server(env, chosen_ports[i]);
+    }
+    gpr_free(env);
+  }
+
+  gpr_free(chosen_ports);
+}
+
+static void chose_port(int port) {
+  if (chosen_ports == NULL) {
+    atexit(free_chosen_ports);
+  }
+  num_chosen_ports++;
+  chosen_ports = gpr_realloc(chosen_ports, sizeof(int) * num_chosen_ports);
+  chosen_ports[num_chosen_ports - 1] = port;
+}
+
+int grpc_pick_unused_port(void) {
+  // Currently only works with the port server
+  char *env = gpr_getenv("GRPC_TEST_PORT_SERVER");
+  GPR_ASSERT(env);
+  int port = grpc_pick_port_using_server(env);
+  gpr_free(env);
+  if (port != 0) {
+    chose_port(port);
+  }
+  return port;
+}
+
+int grpc_pick_unused_port_or_die(void) {
+  int port = grpc_pick_unused_port();
+  GPR_ASSERT(port > 0);
+  return port;
+}
+
+void grpc_recycle_unused_port(int port) { GPR_ASSERT(free_chosen_port(port)); }
+
+#endif /* GRPC_UV && GRPC_TEST_PICK_PORT */
diff --git a/test/core/util/port_windows.c b/test/core/util/port_windows.c
index 9023719675..0c50a46644 100644
--- a/test/core/util/port_windows.c
+++ b/test/core/util/port_windows.c
@@ -31,9 +31,11 @@
  *
  */
 
-#include <grpc/support/port_platform.h>
+#include "src/core/lib/iomgr/port.h"
 #include "test/core/util/test_config.h"
-#if defined(GPR_WINSOCK_SOCKET) && defined(GRPC_TEST_PICK_PORT)
+#if defined(GRPC_WINSOCK_SOCKET) && defined(GRPC_TEST_PICK_PORT)
+
+#include "src/core/lib/iomgr/sockaddr.h"
 
 #include "test/core/util/port.h"
 
@@ -242,4 +244,4 @@ int grpc_pick_unused_port_or_die(void) {
 
 void grpc_recycle_unused_port(int port) { GPR_ASSERT(free_chosen_port(port)); }
 
-#endif /* GPR_WINSOCK_SOCKET && GRPC_TEST_PICK_PORT */
+#endif /* GRPC_WINSOCK_SOCKET && GRPC_TEST_PICK_PORT */
diff --git a/test/core/util/test_tcp_server.c b/test/core/util/test_tcp_server.c
index 8a0b3932d8..16df91d968 100644
--- a/test/core/util/test_tcp_server.c
+++ b/test/core/util/test_tcp_server.c
@@ -31,6 +31,8 @@
  *
  */
 
+#include "src/core/lib/iomgr/sockaddr.h"
+
 #include "test/core/util/test_tcp_server.h"
 
 #include <grpc/grpc.h>
@@ -41,7 +43,7 @@
 #include <grpc/support/time.h>
 #include <string.h>
 #include "src/core/lib/iomgr/endpoint.h"
-#include "src/core/lib/iomgr/sockaddr.h"
+#include "src/core/lib/iomgr/resolve_address.h"
 #include "src/core/lib/iomgr/tcp_server.h"
 #include "test/core/util/port.h"
 
@@ -64,19 +66,20 @@ void test_tcp_server_init(test_tcp_server *server,
 }
 
 void test_tcp_server_start(test_tcp_server *server, int port) {
-  struct sockaddr_in addr;
+  grpc_resolved_address resolved_addr;
+  struct sockaddr_in *addr = (struct sockaddr_in *)resolved_addr.addr;
   int port_added;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
-  addr.sin_family = AF_INET;
-  addr.sin_port = htons((uint16_t)port);
-  memset(&addr.sin_addr, 0, sizeof(addr.sin_addr));
+  addr->sin_family = AF_INET;
+  addr->sin_port = htons((uint16_t)port);
+  memset(&addr->sin_addr, 0, sizeof(addr->sin_addr));
 
-  grpc_error *error = grpc_tcp_server_create(&server->shutdown_complete, NULL,
-                                             &server->tcp_server);
+  grpc_error *error = grpc_tcp_server_create(
+      &exec_ctx, &server->shutdown_complete, NULL, &server->tcp_server);
   GPR_ASSERT(error == GRPC_ERROR_NONE);
-  error = grpc_tcp_server_add_port(server->tcp_server, &addr, sizeof(addr),
-                                   &port_added);
+  error =
+      grpc_tcp_server_add_port(server->tcp_server, &resolved_addr, &port_added);
   GPR_ASSERT(error == GRPC_ERROR_NONE);
   GPR_ASSERT(port_added == port);