Merge github.com:grpc/grpc into grand-unified-closures

3 files changed, 28 insertions, 11 deletions

diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h
index bc7781250e..6c769accdb 100644
--- a/src/core/lib/iomgr/error.h
+++ b/src/core/lib/iomgr/error.h
@@ -47,7 +47,8 @@
 ///  if a grpc_error is passed to a grpc_closure callback function (functions
 ///    with the signature:
 ///      void (*f)(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error))
-///    then those functions do not automatically own a ref to error
+///    then those functions do not own a ref to error (but are free to manually
+///    take a reference).
 ///  if a grpc_error is passed to *ANY OTHER FUNCTION* then that function takes
 ///    ownership of the error
 /// Errors have:
diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c
index 57c26138be..740920d760 100644
--- a/src/core/lib/iomgr/ev_epoll_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_linux.c
@@ -42,6 +42,7 @@
 #include <assert.h>
 #include <errno.h>
 #include <poll.h>
+#include <pthread.h>
 #include <signal.h>
 #include <string.h>
 #include <sys/epoll.h>
@@ -927,7 +928,8 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
     fd->polling_island = NULL;
   }
 
-  grpc_exec_ctx_sched(exec_ctx, fd->on_done_closure, error, NULL);
+  grpc_exec_ctx_sched(exec_ctx, fd->on_done_closure, GRPC_ERROR_REF(error),
+                      NULL);
 
   gpr_mu_unlock(&fd->mu);
   UNREF_BY(fd, 2, reason); /* Drop the reference */
@@ -939,6 +941,7 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
     PI_UNREF(exec_ctx, unref_pi, "fd_orphan");
   }
   GRPC_LOG_IF_ERROR("fd_orphan", GRPC_ERROR_REF(error));
+  GRPC_ERROR_UNREF(error);
 }
 
 static grpc_error *fd_shutdown_error(bool shutdown) {
@@ -1352,8 +1355,10 @@ static void pollset_work_and_unlock(grpc_exec_ctx *exec_ctx,
   gpr_mu_unlock(&pollset->mu);
 
   do {
+    GRPC_SCHEDULING_START_BLOCKING_REGION;
     ep_rv = epoll_pwait(epoll_fd, ep_ev, GRPC_EPOLL_MAX_EVENTS, timeout_ms,
                         sig_mask);
+    GRPC_SCHEDULING_END_BLOCKING_REGION;
     if (ep_rv < 0) {
       if (errno != EINTR) {
         gpr_asprintf(&err_msg,
diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c
index 38ebd2dbcb..2d3f6cf9a7 100644
--- a/src/core/lib/iomgr/tcp_server_posix.c
+++ b/src/core/lib/iomgr/tcp_server_posix.c
@@ -90,10 +90,12 @@ struct grpc_tcp_listener {
   grpc_closure read_closure;
   grpc_closure destroyed_closure;
   struct grpc_tcp_listener *next;
-  /* When we add a listener, more than one can be created, mainly because of
-     IPv6. A sibling will still be in the normal list, but will be flagged
-     as such. Any action, such as ref or unref, will affect all of the
-     siblings in the list. */
+  /* sibling is a linked list of all listeners for a given port. add_port and
+     clone_port place all new listeners in the same sibling list. A member of
+     the 'sibling' list is also a member of the 'next' list. The head of each
+     sibling list has is_sibling==0, and subsequent members of sibling lists
+     have is_sibling==1. is_sibling allows separate sibling lists to be
+     identified while iterating through 'next'. */
   struct grpc_tcp_listener *sibling;
   int is_sibling;
 };
@@ -138,15 +140,17 @@ struct grpc_tcp_server {
 };
 
 static gpr_once check_init = GPR_ONCE_INIT;
-static bool has_so_reuseport;
+static bool has_so_reuseport = false;
 
 static void init(void) {
+#ifndef GPR_MANYLINUX1
   int s = socket(AF_INET, SOCK_STREAM, 0);
   if (s >= 0) {
     has_so_reuseport = GRPC_LOG_IF_ERROR("check for SO_REUSEPORT",
                                          grpc_set_socket_reuse_port(s, 1));
     close(s);
   }
+#endif
 }
 
 grpc_error *grpc_tcp_server_create(grpc_closure *shutdown_complete,
@@ -306,7 +310,7 @@ static grpc_error *prepare_socket(int fd, const struct sockaddr *addr,
 
   GPR_ASSERT(fd >= 0);
 
-  if (so_reuseport) {
+  if (so_reuseport && !grpc_is_unix_socket(addr)) {
     err = grpc_set_socket_reuse_port(fd, 1);
     if (err != GRPC_ERROR_NONE) goto error;
   }
@@ -480,6 +484,9 @@ static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd,
   return err;
 }
 
+/* Insert count new listeners after listener. Every new listener will have the
+   same listen address as listener (SO_REUSEPORT must be enabled). Every new
+   listener is a sibling of listener. */
 static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
   grpc_tcp_listener *sp = NULL;
   char *addr_str;
@@ -506,6 +513,11 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
     sp = gpr_malloc(sizeof(grpc_tcp_listener));
     sp->next = listener->next;
     listener->next = sp;
+    /* sp (the new listener) is a sibling of 'listener' (the original
+       listener). */
+    sp->is_sibling = 1;
+    sp->sibling = listener->sibling;
+    listener->sibling = sp;
     sp->server = listener->server;
     sp->fd = fd;
     sp->emfd = grpc_fd_create(fd, name);
@@ -514,8 +526,6 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
     sp->port = port;
     sp->port_index = listener->port_index;
     sp->fd_index = listener->fd_index + count - i;
-    sp->is_sibling = 1;
-    sp->sibling = listener->is_sibling ? listener->sibling : listener;
     GPR_ASSERT(sp->emfd);
     while (listener->server->tail->next != NULL) {
       listener->server->tail = listener->server->tail->next;
@@ -685,7 +695,8 @@ void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s,
   s->pollset_count = pollset_count;
   sp = s->head;
   while (sp != NULL) {
-    if (s->so_reuseport && pollset_count > 1) {
+    if (s->so_reuseport && !grpc_is_unix_socket(&sp->addr.sockaddr) &&
+        pollset_count > 1) {
       GPR_ASSERT(GRPC_LOG_IF_ERROR(
           "clone_port", clone_port(sp, (unsigned)(pollset_count - 1))));
       for (i = 0; i < pollset_count; i++) {