Merge github.com:grpc/grpc into you-complete-me

Conflicts:
	src/core/iomgr/resolve_address_posix.c
	src/core/surface/server.c

19 files changed, 258 insertions, 144 deletions

diff --git a/src/core/iomgr/endpoint_pair.h b/src/core/iomgr/endpoint_pair.h
index dffbd36d4c..25087be0c7 100644
--- a/src/core/iomgr/endpoint_pair.h
+++ b/src/core/iomgr/endpoint_pair.h
@@ -41,6 +41,7 @@ typedef struct {
   grpc_endpoint *server;
 } grpc_endpoint_pair;
 
-grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(size_t read_slice_size);
+grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name,
+                                                   size_t read_slice_size);
 
 #endif  /* GRPC_INTERNAL_CORE_IOMGR_ENDPOINT_PAIR_H */
diff --git a/src/core/iomgr/endpoint_pair_posix.c b/src/core/iomgr/endpoint_pair_posix.c
index ac511b97b2..9b3b63f1e7 100644
--- a/src/core/iomgr/endpoint_pair_posix.c
+++ b/src/core/iomgr/endpoint_pair_posix.c
@@ -44,6 +44,8 @@
 #include <sys/socket.h>
 
 #include "src/core/iomgr/tcp_posix.h"
+#include "src/core/support/string.h"
+#include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 
 static void create_sockets(int sv[2]) {
@@ -55,12 +57,21 @@ static void create_sockets(int sv[2]) {
   GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
 }
 
-grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(size_t read_slice_size) {
+grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name,
+                                                   size_t read_slice_size) {
   int sv[2];
   grpc_endpoint_pair p;
+  char *final_name;
   create_sockets(sv);
-  p.client = grpc_tcp_create(grpc_fd_create(sv[1]), read_slice_size);
-  p.server = grpc_tcp_create(grpc_fd_create(sv[0]), read_slice_size);
+
+  gpr_asprintf(&final_name, "%s:client", name);
+  p.client =
+      grpc_tcp_create(grpc_fd_create(sv[1], final_name), read_slice_size);
+  gpr_free(final_name);
+  gpr_asprintf(&final_name, "%s:server", name);
+  p.server =
+      grpc_tcp_create(grpc_fd_create(sv[0], final_name), read_slice_size);
+  gpr_free(final_name);
   return p;
 }
 
diff --git a/src/core/iomgr/endpoint_pair_windows.c b/src/core/iomgr/endpoint_pair_windows.c
index 988d622d01..c6790b2937 100644
--- a/src/core/iomgr/endpoint_pair_windows.c
+++ b/src/core/iomgr/endpoint_pair_windows.c
@@ -77,12 +77,12 @@ static void create_sockets(SOCKET sv[2]) {
   sv[0] = svr_sock;
 }
 
-grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(size_t read_slice_size) {
+grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name, size_t read_slice_size) {
   SOCKET sv[2];
   grpc_endpoint_pair p;
   create_sockets(sv);
-  p.client = grpc_tcp_create(grpc_winsocket_create(sv[1]));
-  p.server = grpc_tcp_create(grpc_winsocket_create(sv[0]));
+  p.client = grpc_tcp_create(grpc_winsocket_create(sv[1], "endpoint:client"));
+  p.server = grpc_tcp_create(grpc_winsocket_create(sv[0], "endpoint:server"));
   return p;
 }
 
diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c
index b697fcc64a..28ed7708f7 100644
--- a/src/core/iomgr/fd_posix.c
+++ b/src/core/iomgr/fd_posix.c
@@ -41,7 +41,6 @@
 #include <sys/socket.h>
 #include <unistd.h>
 
-#include "src/core/iomgr/iomgr_internal.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/useful.h>
@@ -91,6 +90,7 @@ static grpc_fd *alloc_fd(int fd) {
     gpr_mu_init(&r->set_state_mu);
     gpr_mu_init(&r->watcher_mu);
   }
+
   gpr_atm_rel_store(&r->refst, 1);
   gpr_atm_rel_store(&r->readst, NOT_READY);
   gpr_atm_rel_store(&r->writest, NOT_READY);
@@ -116,10 +116,9 @@ static void ref_by(grpc_fd *fd, int n) {
 static void unref_by(grpc_fd *fd, int n) {
   gpr_atm old = gpr_atm_full_fetch_add(&fd->refst, -n);
   if (old == n) {
-    close(fd->fd);
-    grpc_iomgr_add_callback(fd->on_done, fd->on_done_user_data);
+    grpc_iomgr_add_callback(&fd->on_done_closure);
     freelist_fd(fd);
-    grpc_iomgr_unref();
+    grpc_iomgr_unregister_object(&fd->iomgr_object);
   } else {
     GPR_ASSERT(old > n);
   }
@@ -138,9 +137,9 @@ void grpc_fd_global_shutdown(void) {
 
 static void do_nothing(void *ignored, int success) {}
 
-grpc_fd *grpc_fd_create(int fd) {
+grpc_fd *grpc_fd_create(int fd, const char *name) {
   grpc_fd *r = alloc_fd(fd);
-  grpc_iomgr_ref();
+  grpc_iomgr_register_object(&r->iomgr_object, name);
   grpc_pollset_add_fd(grpc_backup_pollset(), r);
   return r;
 }
@@ -180,8 +179,8 @@ static void wake_all_watchers_locked(grpc_fd *fd) {
 }
 
 void grpc_fd_orphan(grpc_fd *fd, grpc_iomgr_cb_func on_done, void *user_data) {
-  fd->on_done = on_done ? on_done : do_nothing;
-  fd->on_done_user_data = user_data;
+  grpc_iomgr_closure_init(&fd->on_done_closure, on_done ? on_done : do_nothing,
+                          user_data);
   shutdown(fd->fd, SHUT_RDWR);
   ref_by(fd, 1); /* remove active status, but keep referenced */
   gpr_mu_lock(&fd->watcher_mu);
@@ -195,21 +194,20 @@ void grpc_fd_ref(grpc_fd *fd) { ref_by(fd, 2); }
 
 void grpc_fd_unref(grpc_fd *fd) { unref_by(fd, 2); }
 
-static void make_callback(grpc_iomgr_cb_func cb, void *arg, int success,
+static void process_callback(grpc_iomgr_closure *closure, int success,
                           int allow_synchronous_callback) {
   if (allow_synchronous_callback) {
-    cb(arg, success);
+    closure->cb(closure->cb_arg, success);
   } else {
-    grpc_iomgr_add_delayed_callback(cb, arg, success);
+    grpc_iomgr_add_delayed_callback(closure, success);
   }
 }
 
-static void make_callbacks(grpc_iomgr_closure *callbacks, size_t n, int success,
-                           int allow_synchronous_callback) {
+static void process_callbacks(grpc_iomgr_closure *callbacks, size_t n,
+                              int success, int allow_synchronous_callback) {
   size_t i;
   for (i = 0; i < n; i++) {
-    make_callback(callbacks[i].cb, callbacks[i].cb_arg, success,
-                  allow_synchronous_callback);
+    process_callback(callbacks + i, success, allow_synchronous_callback);
   }
 }
 
@@ -234,10 +232,9 @@ static void notify_on(grpc_fd *fd, gpr_atm *st, grpc_iomgr_closure *closure,
     /* swap was unsuccessful due to an intervening set_ready call.
        Fall through to the READY code below */
     case READY:
-      assert(gpr_atm_no_barrier_load(st) == READY);
+      GPR_ASSERT(gpr_atm_no_barrier_load(st) == READY);
       gpr_atm_rel_store(st, NOT_READY);
-      make_callback(closure->cb, closure->cb_arg,
-                    !gpr_atm_acq_load(&fd->shutdown),
+      process_callback(closure, !gpr_atm_acq_load(&fd->shutdown),
                     allow_synchronous_callback);
       return;
     default: /* WAITING */
@@ -251,7 +248,7 @@ static void notify_on(grpc_fd *fd, gpr_atm *st, grpc_iomgr_closure *closure,
   abort();
 }
 
-static void set_ready_locked(gpr_atm *st, grpc_iomgr_closure *callbacks,
+static void set_ready_locked(gpr_atm *st, grpc_iomgr_closure **callbacks,
                              size_t *ncallbacks) {
   gpr_intptr state = gpr_atm_acq_load(st);
 
@@ -269,9 +266,9 @@ static void set_ready_locked(gpr_atm *st, grpc_iomgr_closure *callbacks,
          Fall through to the WAITING code below */
       state = gpr_atm_acq_load(st);
     default: /* waiting */
-      assert(gpr_atm_no_barrier_load(st) != READY &&
-             gpr_atm_no_barrier_load(st) != NOT_READY);
-      callbacks[(*ncallbacks)++] = *(grpc_iomgr_closure *)state;
+      GPR_ASSERT(gpr_atm_no_barrier_load(st) != READY &&
+                 gpr_atm_no_barrier_load(st) != NOT_READY);
+      callbacks[(*ncallbacks)++] = (grpc_iomgr_closure *)state;
       gpr_atm_rel_store(st, NOT_READY);
       return;
   }
@@ -282,25 +279,30 @@ static void set_ready(grpc_fd *fd, gpr_atm *st,
   /* only one set_ready can be active at once (but there may be a racing
      notify_on) */
   int success;
-  grpc_iomgr_closure cb;
+  grpc_iomgr_closure* closure;
   size_t ncb = 0;
+
   gpr_mu_lock(&fd->set_state_mu);
-  set_ready_locked(st, &cb, &ncb);
+  set_ready_locked(st, &closure, &ncb);
   gpr_mu_unlock(&fd->set_state_mu);
   success = !gpr_atm_acq_load(&fd->shutdown);
-  make_callbacks(&cb, ncb, success, allow_synchronous_callback);
+  GPR_ASSERT(ncb <= 1);
+  if (ncb > 0) {
+    process_callbacks(closure, ncb, success, allow_synchronous_callback);
+  }
 }
 
 void grpc_fd_shutdown(grpc_fd *fd) {
-  grpc_iomgr_closure cb[2];
   size_t ncb = 0;
   gpr_mu_lock(&fd->set_state_mu);
   GPR_ASSERT(!gpr_atm_no_barrier_load(&fd->shutdown));
   gpr_atm_rel_store(&fd->shutdown, 1);
-  set_ready_locked(&fd->readst, cb, &ncb);
-  set_ready_locked(&fd->writest, cb, &ncb);
+  set_ready_locked(&fd->readst, &fd->shutdown_closures[0], &ncb);
+  set_ready_locked(&fd->writest, &fd->shutdown_closures[0], &ncb);
   gpr_mu_unlock(&fd->set_state_mu);
-  make_callbacks(cb, ncb, 0, 0);
+  GPR_ASSERT(ncb <= 2);
+  process_callbacks(fd->shutdown_closures[0], ncb, 0 /* GPR_FALSE */,
+                    0 /* GPR_FALSE */);
 }
 
 void grpc_fd_notify_on_read(grpc_fd *fd, grpc_iomgr_closure *closure) {
diff --git a/src/core/iomgr/fd_posix.h b/src/core/iomgr/fd_posix.h
index cfc533b7f5..0fa71850e3 100644
--- a/src/core/iomgr/fd_posix.h
+++ b/src/core/iomgr/fd_posix.h
@@ -34,17 +34,12 @@
 #ifndef GRPC_INTERNAL_CORE_IOMGR_FD_POSIX_H
 #define GRPC_INTERNAL_CORE_IOMGR_FD_POSIX_H
 
-#include "src/core/iomgr/iomgr.h"
+#include "src/core/iomgr/iomgr_internal.h"
 #include "src/core/iomgr/pollset.h"
 #include <grpc/support/atm.h>
 #include <grpc/support/sync.h>
 #include <grpc/support/time.h>
 
-typedef struct {
-  grpc_iomgr_cb_func cb;
-  void *cb_arg;
-} grpc_iomgr_closure;
-
 typedef struct grpc_fd grpc_fd;
 
 typedef struct grpc_fd_watcher {
@@ -96,15 +91,18 @@ struct grpc_fd {
   gpr_atm readst;
   gpr_atm writest;
 
-  grpc_iomgr_cb_func on_done;
-  void *on_done_user_data;
   struct grpc_fd *freelist_next;
+
+  grpc_iomgr_closure on_done_closure;
+  grpc_iomgr_closure *shutdown_closures[2];
+
+  grpc_iomgr_object iomgr_object;
 };
 
 /* Create a wrapped file descriptor.
    Requires fd is a non-blocking file descriptor.
    This takes ownership of closing fd. */
-grpc_fd *grpc_fd_create(int fd);
+grpc_fd *grpc_fd_create(int fd, const char *name);
 
 /* Releases fd to be asynchronously destroyed.
    on_done is called when the underlying file descriptor is definitely close()d.
diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c
index d22542fc91..249228a214 100644
--- a/src/core/iomgr/iomgr.c
+++ b/src/core/iomgr/iomgr.c
@@ -37,25 +37,19 @@
 
 #include "src/core/iomgr/iomgr_internal.h"
 #include "src/core/iomgr/alarm_internal.h"
+#include "src/core/support/string.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/thd.h>
 #include <grpc/support/sync.h>
 
-typedef struct delayed_callback {
-  grpc_iomgr_cb_func cb;
-  void *cb_arg;
-  int success;
-  struct delayed_callback *next;
-} delayed_callback;
-
 static gpr_mu g_mu;
 static gpr_cv g_rcv;
-static delayed_callback *g_cbs_head = NULL;
-static delayed_callback *g_cbs_tail = NULL;
+static grpc_iomgr_closure *g_cbs_head = NULL;
+static grpc_iomgr_closure *g_cbs_tail = NULL;
 static int g_shutdown;
-static int g_refs;
 static gpr_event g_background_callback_executor_done;
+static grpc_iomgr_object g_root_object;
 
 /* Execute followup callbacks continuously.
    Other threads may check in and help during pollset_work() */
@@ -66,12 +60,11 @@ static void background_callback_executor(void *ignored) {
     gpr_timespec short_deadline =
         gpr_time_add(gpr_now(), gpr_time_from_millis(100));
     if (g_cbs_head) {
-      delayed_callback *cb = g_cbs_head;
-      g_cbs_head = cb->next;
+      grpc_iomgr_closure *closure = g_cbs_head;
+      g_cbs_head = closure->next;
       if (!g_cbs_head) g_cbs_tail = NULL;
       gpr_mu_unlock(&g_mu);
-      cb->cb(cb->cb_arg, cb->success);
-      gpr_free(cb);
+      closure->cb(closure->cb_arg, closure->success);
       gpr_mu_lock(&g_mu);
     } else if (grpc_alarm_check(&g_mu, gpr_now(), &deadline)) {
     } else {
@@ -96,34 +89,48 @@ void grpc_iomgr_init(void) {
   gpr_mu_init(&g_mu);
   gpr_cv_init(&g_rcv);
   grpc_alarm_list_init(gpr_now());
-  g_refs = 0;
+  g_root_object.next = g_root_object.prev = &g_root_object;
+  g_root_object.name = "root";
   grpc_iomgr_platform_init();
   gpr_event_init(&g_background_callback_executor_done);
   gpr_thd_new(&id, background_callback_executor, NULL, NULL);
 }
 
+static size_t count_objects(void) {
+  grpc_iomgr_object *obj;
+  size_t n = 0;
+  for (obj = g_root_object.next; obj != &g_root_object; obj = obj->next) {
+    n++;
+  }
+  return n;
+}
+
 void grpc_iomgr_shutdown(void) {
-  delayed_callback *cb;
+  grpc_iomgr_object *obj;
+  grpc_iomgr_closure *closure;
   gpr_timespec shutdown_deadline =
       gpr_time_add(gpr_now(), gpr_time_from_seconds(10));
 
 
   gpr_mu_lock(&g_mu);
   g_shutdown = 1;
-  while (g_cbs_head || g_refs) {
-    gpr_log(GPR_DEBUG, "Waiting for %d iomgr objects to be destroyed%s", g_refs,
+  while (g_cbs_head || g_root_object.next != &g_root_object) {
+    size_t nobjs = count_objects();
+    gpr_log(GPR_DEBUG, "Waiting for %d iomgr objects to be destroyed%s", nobjs,
             g_cbs_head ? " and executing final callbacks" : "");
-    while (g_cbs_head) {
-      cb = g_cbs_head;
-      g_cbs_head = cb->next;
-      if (!g_cbs_head) g_cbs_tail = NULL;
-      gpr_mu_unlock(&g_mu);
-
-      cb->cb(cb->cb_arg, 0);
-      gpr_free(cb);
-      gpr_mu_lock(&g_mu);
+    if (g_cbs_head) {
+      do {
+        closure = g_cbs_head;
+        g_cbs_head = closure->next;
+        if (!g_cbs_head) g_cbs_tail = NULL;
+        gpr_mu_unlock(&g_mu);
+
+        closure->cb(closure->cb_arg, 0);
+        gpr_mu_lock(&g_mu);
+      } while (g_cbs_head);
+      continue;
     }
-    if (g_refs) {
+    if (nobjs > 0) {
       int timeout = 0;
       gpr_timespec short_deadline = gpr_time_add(gpr_now(),
                                                  gpr_time_from_millis(100));
@@ -137,7 +144,10 @@ void grpc_iomgr_shutdown(void) {
         gpr_log(GPR_DEBUG,
                 "Failed to free %d iomgr objects before shutdown deadline: "
                 "memory leaks are likely",
-                g_refs);
+                count_objects());
+        for (obj = g_root_object.next; obj != &g_root_object; obj = obj->next) {
+          gpr_log(GPR_DEBUG, "LEAKED OBJECT: %s", obj->name);
+        }
         break;
       }
     }
@@ -153,56 +163,66 @@ void grpc_iomgr_shutdown(void) {
   gpr_cv_destroy(&g_rcv);
 }
 
-void grpc_iomgr_ref(void) {
+void grpc_iomgr_register_object(grpc_iomgr_object *obj, const char *name) {
+  obj->name = gpr_strdup(name);
   gpr_mu_lock(&g_mu);
-  ++g_refs;
+  obj->next = &g_root_object;
+  obj->prev = obj->next->prev;
+  obj->next->prev = obj->prev->next = obj;
   gpr_mu_unlock(&g_mu);
 }
 
-void grpc_iomgr_unref(void) {
+void grpc_iomgr_unregister_object(grpc_iomgr_object *obj) {
+  gpr_free(obj->name);
   gpr_mu_lock(&g_mu);
-  if (0 == --g_refs) {
-    gpr_cv_signal(&g_rcv);
-  }
+  obj->next->prev = obj->prev;
+  obj->prev->next = obj->next;
+  gpr_cv_signal(&g_rcv);
   gpr_mu_unlock(&g_mu);
 }
 
-void grpc_iomgr_add_delayed_callback(grpc_iomgr_cb_func cb, void *cb_arg,
-                                     int success) {
-  delayed_callback *dcb = gpr_malloc(sizeof(delayed_callback));
-  dcb->cb = cb;
-  dcb->cb_arg = cb_arg;
-  dcb->success = success;
+
+void grpc_iomgr_closure_init(grpc_iomgr_closure *closure, grpc_iomgr_cb_func cb,
+                             void *cb_arg) {
+  closure->cb = cb;
+  closure->cb_arg = cb_arg;
+  closure->next = NULL;
+}
+
+void grpc_iomgr_add_delayed_callback(grpc_iomgr_closure *closure, int success) {
+  closure->success = success;
   gpr_mu_lock(&g_mu);
-  dcb->next = NULL;
+  closure->next = NULL;
   if (!g_cbs_tail) {
-    g_cbs_head = g_cbs_tail = dcb;
+    g_cbs_head = g_cbs_tail = closure;
   } else {
-    g_cbs_tail->next = dcb;
-    g_cbs_tail = dcb;
+    g_cbs_tail->next = closure;
+    g_cbs_tail = closure;
   }
   gpr_mu_unlock(&g_mu);
 }
 
-void grpc_iomgr_add_callback(grpc_iomgr_cb_func cb, void *cb_arg) {
-  grpc_iomgr_add_delayed_callback(cb, cb_arg, 1);
+
+void grpc_iomgr_add_callback(grpc_iomgr_closure *closure) {
+  grpc_iomgr_add_delayed_callback(closure, 1 /* GPR_TRUE */);
 }
 
+
 int grpc_maybe_call_delayed_callbacks(gpr_mu *drop_mu, int success) {
   int n = 0;
   gpr_mu *retake_mu = NULL;
-  delayed_callback *cb;
+  grpc_iomgr_closure *closure;
   for (;;) {
     /* check for new work */
     if (!gpr_mu_trylock(&g_mu)) {
       break;
     }
-    cb = g_cbs_head;
-    if (!cb) {
+    closure = g_cbs_head;
+    if (!closure) {
       gpr_mu_unlock(&g_mu);
       break;
     }
-    g_cbs_head = cb->next;
+    g_cbs_head = closure->next;
     if (!g_cbs_head) g_cbs_tail = NULL;
     gpr_mu_unlock(&g_mu);
     /* if we have a mutex to drop, do so before executing work */
@@ -211,8 +231,7 @@ int grpc_maybe_call_delayed_callbacks(gpr_mu *drop_mu, int success) {
       retake_mu = drop_mu;
       drop_mu = NULL;
     }
-    cb->cb(cb->cb_arg, success && cb->success);
-    gpr_free(cb);
+    closure->cb(closure->cb_arg, success && closure->success);
     n++;
   }
   if (retake_mu) {
diff --git a/src/core/iomgr/iomgr.h b/src/core/iomgr/iomgr.h
index 1f5d23fdda..a10e481e48 100644
--- a/src/core/iomgr/iomgr.h
+++ b/src/core/iomgr/iomgr.h
@@ -34,14 +34,43 @@
 #ifndef GRPC_INTERNAL_CORE_IOMGR_IOMGR_H
 #define GRPC_INTERNAL_CORE_IOMGR_IOMGR_H
 
-/* gRPC Callback definition */
+/** gRPC Callback definition.
+ *
+ * \param arg Arbitrary input.
+ * \param success An indication on the state of the iomgr. On false, cleanup
+ * actions should be taken (eg, shutdown). */
 typedef void (*grpc_iomgr_cb_func)(void *arg, int success);
 
+/** A closure over a grpc_iomgr_cb_func. */
+typedef struct grpc_iomgr_closure {
+  /** Bound callback. */
+  grpc_iomgr_cb_func cb;
+
+  /** Arguments to be passed to "cb". */
+  void *cb_arg;
+
+  /** Internal. A boolean indication to "cb" on the state of the iomgr.
+   * For instance, closures created during a shutdown would have this field set
+   * to false. */
+  int success;
+
+  /**< Internal. Do not touch */
+  struct grpc_iomgr_closure *next;
+} grpc_iomgr_closure;
+
+/** Initializes \a closure with \a cb and \a cb_arg. */
+void grpc_iomgr_closure_init(grpc_iomgr_closure *closure, grpc_iomgr_cb_func cb,
+                             void *cb_arg);
+
+/** Initializes the iomgr. */
 void grpc_iomgr_init(void);
+
+/** Signals the intention to shutdown the iomgr. */
 void grpc_iomgr_shutdown(void);
 
-/* This function is called from within a callback or from anywhere else
-   and causes the invocation of a callback at some point in the future */
-void grpc_iomgr_add_callback(grpc_iomgr_cb_func cb, void *cb_arg);
+/** Registers a closure to be invoked at some point in the future.
+ *
+ * Can be called from within a callback or from anywhere else */
+void grpc_iomgr_add_callback(grpc_iomgr_closure *closure);
 
 #endif  /* GRPC_INTERNAL_CORE_IOMGR_IOMGR_H */
diff --git a/src/core/iomgr/iomgr_internal.h b/src/core/iomgr/iomgr_internal.h
index 07923258b9..6c1e0e1799 100644
--- a/src/core/iomgr/iomgr_internal.h
+++ b/src/core/iomgr/iomgr_internal.h
@@ -35,15 +35,19 @@
 #define GRPC_INTERNAL_CORE_IOMGR_IOMGR_INTERNAL_H
 
 #include "src/core/iomgr/iomgr.h"
-#include "src/core/iomgr/iomgr_internal.h"
 #include <grpc/support/sync.h>
 
+typedef struct grpc_iomgr_object {
+  char *name;
+  struct grpc_iomgr_object *next;
+  struct grpc_iomgr_object *prev;
+} grpc_iomgr_object;
+
 int grpc_maybe_call_delayed_callbacks(gpr_mu *drop_mu, int success);
-void grpc_iomgr_add_delayed_callback(grpc_iomgr_cb_func cb, void *cb_arg,
-                                     int success);
+void grpc_iomgr_add_delayed_callback(grpc_iomgr_closure *iocb, int success);
 
-void grpc_iomgr_ref(void);
-void grpc_iomgr_unref(void);
+void grpc_iomgr_register_object(grpc_iomgr_object *obj, const char *name);
+void grpc_iomgr_unregister_object(grpc_iomgr_object *obj);
 
 void grpc_iomgr_platform_init(void);
 void grpc_iomgr_platform_shutdown(void);
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index 6da7daabc8..d2f615271e 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -272,6 +272,7 @@ typedef struct grpc_unary_promote_args {
   const grpc_pollset_vtable *original_vtable;
   grpc_pollset *pollset;
   grpc_fd *fd;
+  grpc_iomgr_closure promotion_closure;
 } grpc_unary_promote_args;
 
 static void unary_poll_do_promote(void *args, int success) {
@@ -294,7 +295,7 @@ static void unary_poll_do_promote(void *args, int success) {
   /* First we need to ensure that nobody is polling concurrently */
   while (pollset->counter != 0) {
     grpc_pollset_kick(pollset);
-    grpc_iomgr_add_callback(unary_poll_do_promote, up_args);
+    grpc_iomgr_add_callback(&up_args->promotion_closure);
     gpr_mu_unlock(&pollset->mu);
     return;
   }
@@ -378,7 +379,9 @@ static void unary_poll_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) {
   up_args->pollset = pollset;
   up_args->fd = fd;
   up_args->original_vtable = pollset->vtable;
-  grpc_iomgr_add_callback(unary_poll_do_promote, up_args);
+  up_args->promotion_closure.cb = unary_poll_do_promote;
+  up_args->promotion_closure.cb_arg = up_args;
+  grpc_iomgr_add_callback(&up_args->promotion_closure);
 
   grpc_pollset_kick(pollset);
 }
diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c
index 5af0685f9d..b1f4c09a2c 100644
--- a/src/core/iomgr/pollset_windows.c
+++ b/src/core/iomgr/pollset_windows.c
@@ -66,15 +66,15 @@ int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
   gpr_timespec now;
   now = gpr_now();
   if (gpr_time_cmp(now, deadline) > 0) {
-    return 0;
+    return 0 /* GPR_FALSE */;
   }
-  if (grpc_maybe_call_delayed_callbacks(NULL, 1)) {
-    return 1;
+  if (grpc_maybe_call_delayed_callbacks(NULL, 1 /* GPR_TRUE */)) {
+    return 1 /* GPR_TRUE */;
   }
   if (grpc_alarm_check(NULL, now, &deadline)) {
-    return 1;
+    return 1 /* GPR_TRUE */;
   }
-  return 0;
+  return 0 /* GPR_FALSE */;
 }
 
 void grpc_pollset_kick(grpc_pollset *p) { }
diff --git a/src/core/iomgr/resolve_address_posix.c b/src/core/iomgr/resolve_address_posix.c
index 43fd704a6d..fcf48fe0d7 100644
--- a/src/core/iomgr/resolve_address_posix.c
+++ b/src/core/iomgr/resolve_address_posix.c
@@ -55,6 +55,7 @@ typedef struct {
   char *default_port;
   grpc_resolve_cb cb;
   void *arg;
+  grpc_iomgr_object iomgr_object;
 } request;
 
 grpc_resolved_addresses *grpc_blocking_resolve_address(
@@ -153,9 +154,9 @@ static void do_request(void *rp) {
   grpc_resolve_cb cb = r->cb;
   gpr_free(r->name);
   gpr_free(r->default_port);
+  grpc_iomgr_unregister_object(&r->iomgr_object);
   gpr_free(r);
   cb(arg, resolved);
-  grpc_iomgr_unref();
 }
 
 void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) {
@@ -166,14 +167,17 @@ void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) {
 void grpc_resolve_address(const char *name, const char *default_port,
                           grpc_resolve_cb cb, void *arg) {
   request *r = gpr_malloc(sizeof(request));
-  /*gpr_thd_id id;*/
-  grpc_iomgr_ref();
+  gpr_thd_id id;
+  char *tmp;
+  gpr_asprintf(&tmp, "resolve_address:name='%s':default_port='%s'", name,
+               default_port);
+  grpc_iomgr_register_object(&r->iomgr_object, tmp);
+  gpr_free(tmp);
   r->name = gpr_strdup(name);
   r->default_port = gpr_strdup(default_port);
   r->cb = cb;
   r->arg = arg;
-  /*gpr_thd_new(&id, do_request, r, NULL);*/
-  do_request(r);
+  gpr_thd_new(&id, do_request, r, NULL);
 }
 
 #endif
diff --git a/src/core/iomgr/resolve_address_windows.c b/src/core/iomgr/resolve_address_windows.c
index 9b416dfe8a..7d0d2f9e7a 100644
--- a/src/core/iomgr/resolve_address_windows.c
+++ b/src/core/iomgr/resolve_address_windows.c
@@ -54,6 +54,7 @@ typedef struct {
   char *default_port;
   grpc_resolve_cb cb;
   void *arg;
+  grpc_iomgr_object iomgr_object;
 } request;
 
 grpc_resolved_addresses *grpc_blocking_resolve_address(
@@ -135,7 +136,7 @@ static void do_request(void *rp) {
   gpr_free(r->default_port);
   gpr_free(r);
   cb(arg, resolved);
-  grpc_iomgr_unref();
+  grpc_iomgr_unregister_object(&r->iomgr_object);
 }
 
 void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) {
@@ -147,7 +148,10 @@ void grpc_resolve_address(const char *name, const char *default_port,
                           grpc_resolve_cb cb, void *arg) {
   request *r = gpr_malloc(sizeof(request));
   gpr_thd_id id;
-  grpc_iomgr_ref();
+  const char *label;
+  gpr_asprintf(&label, "resolve:%s", name);
+  grpc_iomgr_register_object(&r->iomgr_object, label);
+  gpr_free(label);
   r->name = gpr_strdup(name);
   r->default_port = gpr_strdup(default_port);
   r->cb = cb;
diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c
index ee5150a696..e4ba0a2b66 100644
--- a/src/core/iomgr/socket_windows.c
+++ b/src/core/iomgr/socket_windows.c
@@ -39,18 +39,17 @@
 #include <grpc/support/log.h>
 
 #include "src/core/iomgr/iocp_windows.h"
-#include "src/core/iomgr/iomgr.h"
 #include "src/core/iomgr/iomgr_internal.h"
 #include "src/core/iomgr/pollset.h"
 #include "src/core/iomgr/pollset_windows.h"
 #include "src/core/iomgr/socket_windows.h"
 
-grpc_winsocket *grpc_winsocket_create(SOCKET socket) {
+grpc_winsocket *grpc_winsocket_create(SOCKET socket, const char *name) {
   grpc_winsocket *r = gpr_malloc(sizeof(grpc_winsocket));
   memset(r, 0, sizeof(grpc_winsocket));
   r->socket = socket;
   gpr_mu_init(&r->state_mu);
-  grpc_iomgr_ref();
+  grpc_iomgr_register_object(&r->iomgr_object, name);
   grpc_iocp_add_socket(r);
   return r;
 }
@@ -64,13 +63,15 @@ int grpc_winsocket_shutdown(grpc_winsocket *socket) {
   gpr_mu_lock(&socket->state_mu);
   if (socket->read_info.cb) {
     callbacks_set++;
-    grpc_iomgr_add_delayed_callback(socket->read_info.cb,
-                                    socket->read_info.opaque, 0);
+    grpc_iomgr_closure_init(&socket->shutdown_closure, socket->read_info.cb,
+                            socket->read_info.opaque);
+    grpc_iomgr_add_delayed_callback(&socket->shutdown_closure, 0);
   }
   if (socket->write_info.cb) {
     callbacks_set++;
-    grpc_iomgr_add_delayed_callback(socket->write_info.cb,
-                                    socket->write_info.opaque, 0);
+    grpc_iomgr_closure_init(&socket->shutdown_closure, socket->write_info.cb,
+                            socket->write_info.opaque);
+    grpc_iomgr_add_delayed_callback(&socket->shutdown_closure, 0);
   }
   gpr_mu_unlock(&socket->state_mu);
   return callbacks_set;
@@ -90,7 +91,7 @@ void grpc_winsocket_orphan(grpc_winsocket *winsocket) {
     grpc_winsocket_destroy(winsocket);
   }
   closesocket(socket);
-  grpc_iomgr_unref();
+  grpc_iomgr_unregister_object(&winsocket->iomgr_object);
 }
 
 void grpc_winsocket_destroy(grpc_winsocket *winsocket) {
diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h
index b27eb14219..7080919af0 100644
--- a/src/core/iomgr/socket_windows.h
+++ b/src/core/iomgr/socket_windows.h
@@ -39,6 +39,8 @@
 #include <grpc/support/sync.h>
 #include <grpc/support/atm.h>
 
+#include "src/core/iomgr/iomgr_internal.h"
+
 /* This holds the data for an outstanding read or write on a socket.
    The mutex to protect the concurrent access to that data is the one
    inside the winsocket wrapper. */
@@ -93,11 +95,16 @@ typedef struct grpc_winsocket {
      there is a pending operation that the IO Completion Port will have to
      wait for. The socket will be collected at that time. */
   int orphan;
+
+  grpc_iomgr_closure shutdown_closure;
+
+  /* A label for iomgr to track outstanding objects */
+  grpc_iomgr_object iomgr_object;
 } grpc_winsocket;
 
 /* Create a wrapped windows handle. This takes ownership of it, meaning that
    it will be responsible for closing it. */
-grpc_winsocket *grpc_winsocket_create(SOCKET socket);
+grpc_winsocket *grpc_winsocket_create(SOCKET socket, const char *name);
 
 /* Initiate an asynchronous shutdown of the socket. Will call off any pending
    operation to cancel them. Returns the number of callbacks that got setup. */
diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c
index 2401fe00e4..aa21ba9b9e 100644
--- a/src/core/iomgr/tcp_client_posix.c
+++ b/src/core/iomgr/tcp_client_posix.c
@@ -48,6 +48,7 @@
 #include "src/core/iomgr/sockaddr_utils.h"
 #include "src/core/iomgr/socket_utils_posix.h"
 #include "src/core/iomgr/tcp_posix.h"
+#include "src/core/support/string.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/time.h>
@@ -185,6 +186,8 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep),
   async_connect *ac;
   struct sockaddr_in6 addr6_v4mapped;
   struct sockaddr_in addr4_copy;
+  char *name;
+  char *addr_str;
 
   /* Use dualstack sockets where available. */
   if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
@@ -211,24 +214,27 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep),
     err = connect(fd, addr, addr_len);
   } while (err < 0 && errno == EINTR);
 
+  grpc_sockaddr_to_string(&addr_str, addr, 1);
+  gpr_asprintf(&name, "tcp-client:%s", addr_str);
+
   if (err >= 0) {
     gpr_log(GPR_DEBUG, "instant connect");
-    cb(arg,
-       grpc_tcp_create(grpc_fd_create(fd), GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
-    return;
+    cb(arg, grpc_tcp_create(grpc_fd_create(fd, name),
+                            GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
+    goto done;
   }
 
   if (errno != EWOULDBLOCK && errno != EINPROGRESS) {
-    gpr_log(GPR_ERROR, "connect error: %s", strerror(errno));
+    gpr_log(GPR_ERROR, "connect error to '%s': %s", addr_str, strerror(errno));
     close(fd);
     cb(arg, NULL);
-    return;
+    goto done;
   }
 
   ac = gpr_malloc(sizeof(async_connect));
   ac->cb = cb;
   ac->cb_arg = arg;
-  ac->fd = grpc_fd_create(fd);
+  ac->fd = grpc_fd_create(fd, name);
   gpr_mu_init(&ac->mu);
   ac->refs = 2;
   ac->write_closure.cb = on_writable;
@@ -236,6 +242,10 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep),
 
   grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, gpr_now());
   grpc_fd_notify_on_write(ac->fd, &ac->write_closure);
+
+done:
+  gpr_free(name);
+  gpr_free(addr_str);
 }
 
 #endif
diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c
index cf5174327d..2a040ffc4a 100644
--- a/src/core/iomgr/tcp_client_windows.c
+++ b/src/core/iomgr/tcp_client_windows.c
@@ -193,7 +193,7 @@ void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
     goto failure;
   }
 
-  socket = grpc_winsocket_create(sock);
+  socket = grpc_winsocket_create(sock, "client");
   info = &socket->write_info;
   info->outstanding = 1;
   success = ConnectEx(sock, addr, addr_len, NULL, 0, NULL, &info->overlapped);
diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c
index cd6b2ecae6..2f19f9d442 100644
--- a/src/core/iomgr/tcp_posix.c
+++ b/src/core/iomgr/tcp_posix.c
@@ -280,6 +280,8 @@ typedef struct {
 
   grpc_iomgr_closure read_closure;
   grpc_iomgr_closure write_closure;
+
+  grpc_iomgr_closure handle_read_closure;
 } grpc_tcp;
 
 static void grpc_tcp_handle_read(void *arg /* grpc_tcp */, int success);
@@ -443,7 +445,8 @@ static void grpc_tcp_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
     tcp->finished_edge = 0;
     grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure);
   } else {
-    grpc_iomgr_add_callback(grpc_tcp_handle_read, tcp);
+    tcp->handle_read_closure.cb_arg = tcp;
+    grpc_iomgr_add_callback(&tcp->handle_read_closure);
   }
 }
 
@@ -592,6 +595,8 @@ grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd, size_t slice_size) {
   tcp->read_closure.cb_arg = tcp;
   tcp->write_closure.cb = grpc_tcp_handle_write;
   tcp->write_closure.cb_arg = tcp;
+
+  tcp->handle_read_closure.cb = grpc_tcp_handle_read;
   return &tcp->base;
 }
 
diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c
index d1cd8a769c..3cd40faafc 100644
--- a/src/core/iomgr/tcp_server_posix.c
+++ b/src/core/iomgr/tcp_server_posix.c
@@ -60,6 +60,7 @@
 #include "src/core/iomgr/sockaddr_utils.h"
 #include "src/core/iomgr/socket_utils_posix.h"
 #include "src/core/iomgr/tcp_posix.h"
+#include "src/core/support/string.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/sync.h>
@@ -281,6 +282,8 @@ static void on_read(void *arg, int success) {
   for (;;) {
     struct sockaddr_storage addr;
     socklen_t addrlen = sizeof(addr);
+    char *addr_str;
+    char *name;
     /* Note: If we ever decide to return this address to the user, remember to
              strip off the ::ffff:0.0.0.0/96 prefix first. */
     int fd = grpc_accept4(sp->fd, (struct sockaddr *)&addr, &addrlen, 1, 1);
@@ -299,9 +302,15 @@ static void on_read(void *arg, int success) {
 
     grpc_set_socket_no_sigpipe_if_possible(fd);
 
-    sp->server->cb(
-        sp->server->cb_arg,
-        grpc_tcp_create(grpc_fd_create(fd), GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
+    grpc_sockaddr_to_string(&addr_str, (struct sockaddr *)&addr, 1);
+    gpr_asprintf(&name, "tcp-server-connection:%s", addr_str);
+
+    sp->server->cb(sp->server->cb_arg,
+                   grpc_tcp_create(grpc_fd_create(fd, name),
+                                   GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
+
+    gpr_free(addr_str);
+    gpr_free(name);
   }
 
   abort();
@@ -318,9 +327,13 @@ static int add_socket_to_server(grpc_tcp_server *s, int fd,
                                 const struct sockaddr *addr, int addr_len) {
   server_port *sp;
   int port;
+  char *addr_str;
+  char *name;
 
   port = prepare_socket(fd, addr, addr_len);
   if (port >= 0) {
+    grpc_sockaddr_to_string(&addr_str, (struct sockaddr *)&addr, 1);
+    gpr_asprintf(&name, "tcp-server-listener:%s", addr_str);
     gpr_mu_lock(&s->mu);
     GPR_ASSERT(!s->cb && "must add ports before starting server");
     /* append it to the list under a lock */
@@ -331,11 +344,13 @@ static int add_socket_to_server(grpc_tcp_server *s, int fd,
     sp = &s->ports[s->nports++];
     sp->server = s;
     sp->fd = fd;
-    sp->emfd = grpc_fd_create(fd);
+    sp->emfd = grpc_fd_create(fd, name);
     memcpy(sp->addr.untyped, addr, addr_len);
     sp->addr_len = addr_len;
     GPR_ASSERT(sp->emfd);
     gpr_mu_unlock(&s->mu);
+    gpr_free(addr_str);
+    gpr_free(name);
   }
 
   return port;
diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c
index d22acc7453..9ef369dfd8 100644
--- a/src/core/iomgr/tcp_server_windows.c
+++ b/src/core/iomgr/tcp_server_windows.c
@@ -270,7 +270,8 @@ static void on_accept(void *arg, int from_iocp) {
       gpr_free(utf8_message);
       closesocket(sock);
     } else {
-      ep = grpc_tcp_create(grpc_winsocket_create(sock));
+	  /* TODO(ctiller): add sockaddr address to label */
+      ep = grpc_tcp_create(grpc_winsocket_create(sock, "server"));
     }
   } else {
     /* If we're not notified from the IOCP, it means we are asked to shutdown.
@@ -336,7 +337,7 @@ static int add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
     }
     sp = &s->ports[s->nports++];
     sp->server = s;
-    sp->socket = grpc_winsocket_create(sock);
+    sp->socket = grpc_winsocket_create(sock, "listener");
     sp->shutting_down = 0;
     sp->AcceptEx = AcceptEx;
     sp->new_socket = INVALID_SOCKET;