Significantly refactor the polling island locking and refcounting code

src/core/lib/iomgr/ev_epoll_linux.c

@@ -140,18 +140,40 @@ static void fd_global_shutdown(void);
 #define CLOSURE_READY ((grpc_closure *)1)
 
 /*******************************************************************************
- * Polling-island Declarations
+ * Polling island Declarations
  */
-/* TODO: sree: Consider making ref_cnt and merged_to to gpr_atm - This would
- * significantly reduce the number of mutex acquisition calls. */
+
+// #define GRPC_PI_REF_COUNT_DEBUG
+#ifdef GRPC_PI_REF_COUNT_DEBUG
+
+#define PI_ADD_REF(p, r) pi_add_ref_dbg((p), 1, (r), __FILE__, __LINE__)
+#define PI_UNREF(p, r) pi_unref_dbg((p), 1, (r), __FILE__, __LINE__)
+
+#else /* defined(GRPC_PI_REF_COUNT_DEBUG) */
+
+#define PI_ADD_REF(p, r) pi_add_ref((p), 1)
+#define PI_UNREF(p, r) pi_unref((p), 1)
+
+#endif /* !defined(GPRC_PI_REF_COUNT_DEBUG) */
+
 typedef struct polling_island {
   gpr_mu mu;
-  int ref_cnt;
-
-  /* Points to the polling_island this merged into.
-   * If merged_to is not NULL, all the remaining fields (except mu and ref_cnt)
-   * are invalid and must be ignored */
-  struct polling_island *merged_to;
+  /* Ref count. Use PI_ADD_REF() and PI_UNREF() macros to increment/decrement
+     the refcount.
+     Once the ref count becomes zero, this structure is destroyed which means
+     we should ensure that there is never a scenario where a PI_ADD_REF() is
+     racing with a PI_UNREF() that just made the ref_count zero. */
+  gpr_atm ref_count;
+
+  /* Pointer to the polling_island this merged into.
+   * merged_to value is only set once in polling_island's lifetime (and that too
+   * only if the island is merged with another island). Because of this, we can
+   * use gpr_atm type here so that we can do atomic access on this and reduce
+   * lock contention on 'mu' mutex.
+   *
+   * Note that if this field is not NULL (i.e not 0), all the remaining fields
+   * (except mu and ref_count) are invalid and must be ignored. */
+  gpr_atm merged_to;
 
   /* The fd of the underlying epoll set */
   int epoll_fd;
@@ -236,6 +258,8 @@ static grpc_wakeup_fd polling_island_wakeup_fd;
 static gpr_mu g_pi_freelist_mu;
 static polling_island *g_pi_freelist = NULL;
 
+static void polling_island_delete(); /* Forward declaration */
+
 #ifdef GRPC_TSAN
 /* Currently TSAN may incorrectly flag data races between epoll_ctl and
    epoll_wait for any grpc_fd structs that are added to the epoll set via
@@ -247,6 +271,51 @@ static polling_island *g_pi_freelist = NULL;
 gpr_atm g_epoll_sync;
 #endif /* defined(GRPC_TSAN) */
 
+#ifdef GRPC_PI_REF_COUNT_DEBUG
+long pi_add_ref(polling_island *pi, int ref_cnt);
+long pi_unref(polling_island *pi, int ref_cnt);
+
+void pi_add_ref_dbg(polling_island *pi, int ref_cnt, char *reason, char *file,
+                    int line) {
+  long old_cnt = pi_add_ref(pi, ref_cnt);
+  gpr_log(GPR_DEBUG, "Add ref pi: %p, old:%ld -> new:%ld (%s) - (%s, %d)",
+          (void *)pi, old_cnt, (old_cnt + ref_cnt), reason, file, line);
+}
+
+void pi_unref_dbg(polling_island *pi, int ref_cnt, char *reason, char *file,
+                  int line) {
+  long old_cnt = pi_unref(pi, ref_cnt);
+  gpr_log(GPR_DEBUG, "Unref pi: %p, old:%ld -> new:%ld (%s) - (%s, %d)",
+          (void *)pi, old_cnt, (old_cnt - ref_cnt), reason, file, line);
+}
+#endif
+
+long pi_add_ref(polling_island *pi, int ref_cnt) {
+  return gpr_atm_no_barrier_fetch_add(&pi->ref_count, ref_cnt);
+}
+
+long pi_unref(polling_island *pi, int ref_cnt) {
+  long old_cnt = gpr_atm_no_barrier_fetch_add(&pi->ref_count, -ref_cnt);
+
+  /* If ref count went to zero, delete the polling island. Note that this need
+     not be done under a lock. Once the ref count goes to zero, we are
+     guaranteed that no one else holds a reference to the polling island (and
+     that there is no racing pi_add_ref() call either.
+
+     Also, if we are deleting the polling island and the merged_to field is
+     non-empty, we should remove a ref to the merged_to polling island
+   */
+  if (old_cnt == ref_cnt) {
+    polling_island *next = (polling_island *)gpr_atm_acq_load(&pi->merged_to);
+    polling_island_delete(pi);
+    if (next != NULL) {
+      PI_UNREF(next, "pi_delete"); /* Recursive call */
+    }
+  }
+
+  return old_cnt;
+}
+
 /* The caller is expected to hold pi->mu lock before calling this function */
 static void polling_island_add_fds_locked(polling_island *pi, grpc_fd **fds,
                                           size_t fd_count, bool add_fd_refs) {
@@ -355,8 +424,7 @@ static void polling_island_remove_fd_locked(polling_island *pi, grpc_fd *fd,
   }
 }
 
-static polling_island *polling_island_create(grpc_fd *initial_fd,
-                                             int initial_ref_cnt) {
+static polling_island *polling_island_create(grpc_fd *initial_fd) {
   polling_island *pi = NULL;
 
   /* Try to get one from the polling island freelist */
@@ -377,6 +445,9 @@ static polling_island *polling_island_create(grpc_fd *initial_fd,
     pi->fds = NULL;
   }
 
+  gpr_atm_no_barrier_store(&pi->ref_count, 0);
+  gpr_atm_no_barrier_store(&pi->merged_to, NULL);
+
   pi->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
 
   if (pi->epoll_fd < 0) {
@@ -387,14 +458,12 @@ static polling_island *polling_island_create(grpc_fd *initial_fd,
 
   polling_island_add_wakeup_fd_locked(pi, &grpc_global_wakeup_fd);
 
-  pi->ref_cnt = initial_ref_cnt;
-  pi->merged_to = NULL;
   pi->next_free = NULL;
 
   if (initial_fd != NULL) {
-    /* It is not really needed to get the pi->mu lock here. If this is a newly
-       created polling island (or one that we got from the freelist), no one
-       else would be holding a lock to it anyway */
+    /* Lock the polling island here just in case we got this structure from the
+       freelist and the polling island lock was not released yet (by the code
+       that adds the polling island to the freelist) */
     gpr_mu_lock(&pi->mu);
     polling_island_add_fds_locked(pi, &initial_fd, 1, true);
     gpr_mu_unlock(&pi->mu);
@@ -404,140 +473,136 @@ static polling_island *polling_island_create(grpc_fd *initial_fd,
 }
 
 static void polling_island_delete(polling_island *pi) {
-  GPR_ASSERT(pi->ref_cnt == 0);
   GPR_ASSERT(pi->fd_cnt == 0);
 
+  gpr_atm_rel_store(&pi->merged_to, NULL);
+
   close(pi->epoll_fd);
   pi->epoll_fd = -1;
 
-  pi->merged_to = NULL;
-
   gpr_mu_lock(&g_pi_freelist_mu);
   pi->next_free = g_pi_freelist;
   g_pi_freelist = pi;
   gpr_mu_unlock(&g_pi_freelist_mu);
 }
 
-void polling_island_unref_and_unlock(polling_island *pi, int unref_by) {
-  pi->ref_cnt -= unref_by;
-  int ref_cnt = pi->ref_cnt;
-  GPR_ASSERT(ref_cnt >= 0);
-
-  gpr_mu_unlock(&pi->mu);
-
-  if (ref_cnt == 0) {
-    polling_island_delete(pi);
-  }
-}
-
-polling_island *polling_island_update_and_lock(polling_island *pi, int unref_by,
-                                               int add_ref_by) {
+/* Gets the lock on the *latest* polling island i.e the last polling island in
+   the linked list (linked by 'merged_to' link). Call gpr_mu_unlock on the
+   returned polling island's mu.
+   Usage: To lock/unlock polling island "pi", do the following:
+      polling_island *pi_latest = polling_island_lock(pi);
+      ...
+      ... critical section ..
+      ...
+      gpr_mu_unlock(&pi_latest->mu); //NOTE: use pi_latest->mu. NOT pi->mu */
+polling_island *polling_island_lock(polling_island *pi) {
   polling_island *next = NULL;
-  gpr_mu_lock(&pi->mu);
-  while (pi->merged_to != NULL) {
-    next = pi->merged_to;
-    polling_island_unref_and_unlock(pi, unref_by);
+  while (true) {
+    next = (polling_island *)gpr_atm_acq_load(&pi->merged_to);
+    if (next == NULL) {
+      /* pi is the last node in the linked list. Get the lock and check again
+         (under the pi->mu lock) that pi is still the last node (because a merge
+         may have happend after the (next == NULL) check above and before
+         getting the pi->mu lock.
+         If pi is the last node, we are done. If not, unlock and continue
+         traversing the list */
+      gpr_mu_lock(&pi->mu);
+      next = (polling_island *)gpr_atm_acq_load(&pi->merged_to);
+      if (next == NULL) {
+        break;
+      }
+      gpr_mu_unlock(&pi->mu);
+    }
+
     pi = next;
-    gpr_mu_lock(&pi->mu);
   }
 
-  pi->ref_cnt += add_ref_by;
   return pi;
 }
 
-void polling_island_pair_update_and_lock(polling_island **p,
-                                         polling_island **q) {
+/* Gets the lock on the *latest* polling islands pointed by *p and *q.
+   This function is needed because calling the following block of code to obtain
+   locks on polling islands (*p and *q) is prone to deadlocks.
+     {
+       polling_island_lock(*p);
+       polling_island_lock(*q);
+     }
+
+   Usage/exmaple:
+     polling_island *p1;
+     polling_island *p2;
+     ..
+     polling_island_lock_pair(&p1, &p2);
+     ..
+     .. Critical section with both p1 and p2 locked
+     ..
+     // Release locks
+     // **IMPORTANT**: Make sure you check p1 == p2 AFTER the function
+     // polling_island_lock_pair() was called and if so, release the lock only
+     // once. Note: Even if p1 != p2 beforec calling polling_island_lock_pair(),
+     // they might be after the function returns:
+     if (p1 == p2) {
+       gpr_mu_unlock(&p1->mu)
+     } else {
+       gpr_mu_unlock(&p1->mu);
+       gpr_mu_unlock(&p2->mu);
+     }
+
+*/
+void polling_island_lock_pair(polling_island **p, polling_island **q) {
   polling_island *pi_1 = *p;
   polling_island *pi_2 = *q;
-  polling_island *temp = NULL;
-  bool pi_1_locked = false;
-  bool pi_2_locked = false;
-  int num_swaps = 0;
-
-  /* Loop until either pi_1 == pi_2 or until we acquired locks on both pi_1
-     and pi_2 */
-  while (pi_1 != pi_2 && !(pi_1_locked && pi_2_locked)) {
-    /* The following assertions are true at this point:
-       - pi_1 != pi_2  (else, the while loop would have exited)
-       - pi_1 MAY be locked
-       - pi_2 is NOT locked */
-
-    /* To maintain lock order consistency, always lock polling_island node with
-       lower address first.
-       First, make sure pi_1 < pi_2 before proceeding any further. If it turns
-       out that pi_1 > pi_2, unlock pi_1 if locked (because pi_2 is not locked
-       at this point and having pi_1 locked would violate the lock order) and
-       swap pi_1 and pi_2 so that pi_1 becomes less than pi_2 */
-    if (pi_1 > pi_2) {
-      if (pi_1_locked) {
-        gpr_mu_unlock(&pi_1->mu);
-        pi_1_locked = false;
-      }
+  polling_island *next_1 = NULL;
+  polling_island *next_2 = NULL;
+
+  /* The algorithm is simple:
+      - Go to the last polling islands in the linked lists *pi_1 and *pi_2 (and
+        keep updating pi_1 and pi_2)
+      - Then obtain locks on the islands by following a lock order rule of
+        locking polling_island with lower address first
+           Special case: Before obtaining the locks, check if pi_1 and pi_2 are
+           pointing to the same island. If that is the case, we can just call
+           polling_island_lock()
+      - After obtaining both the locks, double check that the polling islands
+        are still the last polling islands in their respective linked lists
+        (this is because there might have been polling island merges before
+        we got the lock)
+      - If the polling islands are the last islands, we are done. If not,
+        release the locks and continue the process from the first step */
+  while (true) {
+    next_1 = (polling_island *)gpr_atm_acq_load(&pi_1->merged_to);
+    while (next_1 != NULL) {
+      pi_1 = next_1;
+      next_1 = (polling_island *)gpr_atm_acq_load(&pi_1->merged_to);
+    }
 
-      GPR_SWAP(polling_island *, pi_1, pi_2);
-      num_swaps++;
+    next_2 = (polling_island *)gpr_atm_acq_load(&pi_2->merged_to);
+    while (next_2 != NULL) {
+      pi_2 = next_2;
+      next_2 = (polling_island *)gpr_atm_acq_load(&pi_2->merged_to);
     }
 
-    /* The following assertions are true at this point:
-       - pi_1 != pi_2
-       - pi_1 < pi_2  (address of pi_1 is less than that of pi_2)
-       - pi_1 MAYBE locked
-       - pi_2 is NOT locked */
+    if (pi_1 == pi_2) {
+      pi_1 = pi_2 = polling_island_lock(pi_1);
+      break;
+    }
 
-    /* Lock pi_1 (if pi_1 is pointing to the terminal node in the list) */
-    if (!pi_1_locked) {
+    if (pi_1 < pi_2) {
+      gpr_mu_lock(&pi_1->mu);
+      gpr_mu_lock(&pi_2->mu);
+    } else {
+      gpr_mu_lock(&pi_2->mu);
       gpr_mu_lock(&pi_1->mu);
-      pi_1_locked = true;
-
-      /* If pi_1 is not terminal node (i.e pi_1->merged_to != NULL), we are not
-         done locking this polling_island yet. Release the lock on this node and
-         advance pi_1 to the next node in the list; and go to the beginning of
-         the loop (we can't proceed to locking pi_2 unless we locked pi_1 first)
-         */
-      if (pi_1->merged_to != NULL) {
-        temp = pi_1->merged_to;
-        polling_island_unref_and_unlock(pi_1, 1);
-        pi_1 = temp;
-        pi_1_locked = false;
-
-        continue;
-      }
     }
 
-    /* The following assertions are true at this point:
-       - pi_1 is locked
-       - pi_2 is unlocked
-       - pi_1 != pi_2 */
-
-    gpr_mu_lock(&pi_2->mu);
-    pi_2_locked = true;
-
-    /* If pi_2 is not terminal node, we are not done locking this polling_island
-       yet. Release the lock and update pi_2 to the next node in the list */
-    if (pi_2->merged_to != NULL) {
-      temp = pi_2->merged_to;
-      polling_island_unref_and_unlock(pi_2, 1);
-      pi_2 = temp;
-      pi_2_locked = false;
+    next_1 = (polling_island *)gpr_atm_acq_load(&pi_1->merged_to);
+    next_2 = (polling_island *)gpr_atm_acq_load(&pi_2->merged_to);
+    if (next_1 == NULL && next_2 == NULL) {
+      break;
     }
-  }
 
-  /* At this point, either pi_1 == pi_2 AND/OR we got both locks */
-  if (pi_1 == pi_2) {
-    /* We may or may not have gotten the lock. If we didn't, walk the rest of
-      the polling_island list and get the lock */
-    GPR_ASSERT(pi_1_locked || (!pi_1_locked && !pi_2_locked));
-    if (!pi_1_locked) {
-      pi_1 = pi_2 = polling_island_update_and_lock(pi_1, 2, 0);
-    }
-  } else {
-    GPR_ASSERT(pi_1_locked && pi_2_locked);
-    /* If we swapped pi_1 and pi_2 odd number of times, do one more swap so that
-       pi_1 and pi_2 point to the same polling_island lists they started off
-       with at the beginning of this function (i.e *p and *q respectively) */
-    if (num_swaps % 2 > 0) {
-      GPR_SWAP(polling_island *, pi_1, pi_2);
-    }
+    gpr_mu_unlock(&pi_1->mu);
+    gpr_mu_unlock(&pi_2->mu);
   }
 
   *p = pi_1;
@@ -546,7 +611,7 @@ void polling_island_pair_update_and_lock(polling_island **p,
 
 polling_island *polling_island_merge(polling_island *p, polling_island *q) {
   /* Get locks on both the polling islands */
-  polling_island_pair_update_and_lock(&p, &q);
+  polling_island_lock_pair(&p, &q);
 
   if (p == q) {
     /* Nothing needs to be done here */
@@ -568,15 +633,14 @@ polling_island *polling_island_merge(polling_island *p, polling_island *q) {
   /* Wakeup all the pollers (if any) on p so that they can pickup this change */
   polling_island_add_wakeup_fd_locked(p, &polling_island_wakeup_fd);
 
-  p->merged_to = q;
+  /* Add the 'merged_to' link from p --> q */
+  gpr_atm_rel_store(&p->merged_to, q);
+  PI_ADD_REF(q, "pi_merge"); /* To account for the new incoming ref from p */
 
-  /* - The merged polling island (i.e q) inherits all the ref counts of the
-       island merging with it (i.e p)
-     - The island p will lose a ref count */
-  q->ref_cnt += p->ref_cnt;
-  polling_island_unref_and_unlock(p, 1); /* Decrement refcount */
-  polling_island_unref_and_unlock(q, 0); /* Just Unlock. Don't decrement ref */
+  gpr_mu_unlock(&p->mu);
+  gpr_mu_unlock(&q->mu);
 
+  /* Return the merged polling island */
   return q;
 }
 
@@ -667,6 +731,7 @@ static void unref_by(grpc_fd *fd, int n) {
     fd->freelist_next = fd_freelist;
     fd_freelist = fd;
     grpc_iomgr_unregister_object(&fd->iomgr_object);
+
     gpr_mu_unlock(&fd_freelist_mu);
   } else {
     GPR_ASSERT(old > n);
@@ -785,16 +850,20 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
   REF_BY(fd, 1, reason);
 
   /* Remove the fd from the polling island:
-     - Update the fd->polling_island to point to the latest polling island
-     - Remove the fd from the polling island.
-     - Remove a ref to the polling island and set fd->polling_island to NULL */
+     - Get a lock on the latest polling island (i.e the last island in the
+       linked list pointed by fd->polling_island). This is the island that
+       would actually contain the fd
+     - Remove the fd from the latest polling island
+     - Unlock the latest polling island
+     - Set fd->polling_island to NULL (but remove the ref on the polling island
+       before doing this.) */
   gpr_mu_lock(&fd->pi_mu);
   if (fd->polling_island != NULL) {
-    fd->polling_island =
-        polling_island_update_and_lock(fd->polling_island, 1, 0);
-    polling_island_remove_fd_locked(fd->polling_island, fd, is_fd_closed);
+    polling_island *pi_latest = polling_island_lock(fd->polling_island);
+    polling_island_remove_fd_locked(pi_latest, fd, is_fd_closed);
+    gpr_mu_unlock(&pi_latest->mu);
 
-    polling_island_unref_and_unlock(fd->polling_island, 1);
+    PI_UNREF(fd->polling_island, "fd_orphan");
     fd->polling_island = NULL;
   }
   gpr_mu_unlock(&fd->pi_mu);
@@ -1050,17 +1119,13 @@ static void fd_become_writable(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
   gpr_mu_unlock(&fd->mu);
 }
 
-/* Release the reference to pollset->polling_island and set it to NULL.
-   pollset->mu must be held */
-static void pollset_release_polling_island_locked(grpc_pollset *pollset) {
-  gpr_mu_lock(&pollset->pi_mu);
-  if (pollset->polling_island) {
-    pollset->polling_island =
-        polling_island_update_and_lock(pollset->polling_island, 1, 0);
-    polling_island_unref_and_unlock(pollset->polling_island, 1);
-    pollset->polling_island = NULL;
+static void pollset_release_polling_island(grpc_pollset *ps, char *reason) {
+  gpr_mu_lock(&ps->pi_mu);
+  if (ps->polling_island != NULL) {
+    PI_UNREF(ps->polling_island, reason);
   }
-  gpr_mu_unlock(&pollset->pi_mu);
+  ps->polling_island = NULL;
+  gpr_mu_unlock(&ps->pi_mu);
 }
 
 static void finish_shutdown_locked(grpc_exec_ctx *exec_ctx,
@@ -1069,8 +1134,9 @@ static void finish_shutdown_locked(grpc_exec_ctx *exec_ctx,
   GPR_ASSERT(!pollset_has_workers(pollset));
 
   pollset->finish_shutdown_called = true;
-  pollset_release_polling_island_locked(pollset);
 
+  /* Release the ref and set pollset->polling_island to NULL */
+  pollset_release_polling_island(pollset, "ps_shutdown");
   grpc_exec_ctx_enqueue(exec_ctx, pollset->shutdown_done, true, NULL);
 }
 
@@ -1110,7 +1176,7 @@ static void pollset_reset(grpc_pollset *pollset) {
   pollset->finish_shutdown_called = false;
   pollset->kicked_without_pollers = false;
   pollset->shutdown_done = NULL;
-  pollset_release_polling_island_locked(pollset);
+  pollset_release_polling_island(pollset, "ps_reset");
 }
 
 #define GRPC_EPOLL_MAX_EVENTS 1000
@@ -1124,28 +1190,37 @@ static void pollset_work_and_unlock(grpc_exec_ctx *exec_ctx,
   GPR_TIMER_BEGIN("pollset_work_and_unlock", 0);
 
   /* We need to get the epoll_fd to wait on. The epoll_fd is in inside the
-     polling island pointed by pollset->polling_island.
+     latest polling island pointed by pollset->polling_island.
      Acquire the following locks:
      - pollset->mu (which we already have)
      - pollset->pi_mu
-     - pollset->polling_island->mu (call polling_island_update_and_lock())*/
+     - pollset->polling_island lock  */
   gpr_mu_lock(&pollset->pi_mu);
 
-  pi = pollset->polling_island;
-  if (pi == NULL) {
-    pi = polling_island_create(NULL, 1);
+  if (pollset->polling_island == NULL) {
+    pollset->polling_island = polling_island_create(NULL);
+    PI_ADD_REF(pollset->polling_island, "ps");
   }
 
-  /* In addition to locking the polling island, add a ref so that the island
-     does not get destroyed (which means the epoll_fd won't be closed) while
-     we are are doing an epoll_wait() on the epoll_fd */
-  pi = polling_island_update_and_lock(pi, 1, 1);
+  pi = polling_island_lock(pollset->polling_island);
   epoll_fd = pi->epoll_fd;
 
-  /* Update the pollset->polling_island */
-  pollset->polling_island = pi;
+  /* Update the pollset->polling_island since the island being pointed by
+     pollset->polling_island may not be the latest (i.e pi) */
+  if (pollset->polling_island != pi) {
+    /* Always do PI_ADD_REF before PI_UNREF because PI_UNREF may cause the
+       polling island to be deleted */
+    PI_ADD_REF(pi, "ps");
+    PI_UNREF(pollset->polling_island, "ps");
+    pollset->polling_island = pi;
+  }
+
+  /* Add an extra ref so that the island does not get destroyed (which means
+     the epoll_fd won't be closed) while we are are doing an epoll_wait() on the
+     epoll_fd */
+  PI_ADD_REF(pi, "ps_work");
 
-  polling_island_unref_and_unlock(pollset->polling_island, 0); /* Keep the ref*/
+  gpr_mu_unlock(&pi->mu);
   gpr_mu_unlock(&pollset->pi_mu);
   gpr_mu_unlock(&pollset->mu);
 
@@ -1193,14 +1268,12 @@ static void pollset_work_and_unlock(grpc_exec_ctx *exec_ctx,
 
   GPR_ASSERT(pi != NULL);
 
-  /* Before leaving, release the extra ref we added to the polling island */
-  /* It is important to note that at this point 'pi' may not be the same as
-   * pollset->polling_island. This is because pollset->polling_island pointer
-   * gets updated whenever the underlying polling island is merged with another
-   * island and while we are doing epoll_wait() above, the polling island may
-   * have been merged */
-  pi = polling_island_update_and_lock(pi, 1, 0); /* No new ref added */
-  polling_island_unref_and_unlock(pi, 1);
+  /* Before leaving, release the extra ref we added to the polling island. It
+     is important to use "pi" here (i.e our old copy of pollset->polling_island
+     that we got before releasing the polling island lock). This is because
+     pollset->polling_island pointer might get udpated in other parts of the
+     code when there is an island merge while we are doing epoll_wait() above */
+  PI_UNREF(pi, "ps_work");
 
   GPR_TIMER_END("pollset_work_and_unlock", 0);
 }
@@ -1297,20 +1370,34 @@ static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
   if (fd->polling_island == pollset->polling_island) {
     pi_new = fd->polling_island;
     if (pi_new == NULL) {
-      pi_new = polling_island_create(fd, 2);
+      pi_new = polling_island_create(fd);
     }
   } else if (fd->polling_island == NULL) {
-    pi_new = polling_island_update_and_lock(pollset->polling_island, 1, 1);
-    polling_island_add_fds_locked(pollset->polling_island, &fd, 1, true);
+    pi_new = polling_island_lock(pollset->polling_island);
+    polling_island_add_fds_locked(pi_new, &fd, 1, true);
     gpr_mu_unlock(&pi_new->mu);
   } else if (pollset->polling_island == NULL) {
-    pi_new = polling_island_update_and_lock(fd->polling_island, 1, 1);
+    pi_new = polling_island_lock(fd->polling_island);
     gpr_mu_unlock(&pi_new->mu);
   } else {
     pi_new = polling_island_merge(fd->polling_island, pollset->polling_island);
   }
 
-  fd->polling_island = pollset->polling_island = pi_new;
+  if (fd->polling_island != pi_new) {
+    PI_ADD_REF(pi_new, "fd");
+    if (fd->polling_island != NULL) {
+      PI_UNREF(fd->polling_island, "fd");
+    }
+    fd->polling_island = pi_new;
+  }
+
+  if (pollset->polling_island != pi_new) {
+    PI_ADD_REF(pi_new, "ps");
+    if (pollset->polling_island != NULL) {
+      PI_UNREF(pollset->polling_island, "ps");
+    }
+    pollset->polling_island = pi_new;
+  }
 
   gpr_mu_unlock(&fd->pi_mu);
   gpr_mu_unlock(&pollset->pi_mu);
@@ -1481,28 +1568,19 @@ void *grpc_pollset_get_polling_island(grpc_pollset *ps) {
   return pi;
 }
 
-static polling_island *get_polling_island(polling_island *p) {
-  if (p == NULL) {
-    return NULL;
-  }
+bool grpc_are_polling_islands_equal(void *p, void *q) {
+  polling_island *p1 = p;
+  polling_island *p2 = q;
 
-  polling_island *next;
-  gpr_mu_lock(&p->mu);
-  while (p->merged_to != NULL) {
-    next = p->merged_to;
-    gpr_mu_unlock(&p->mu);
-    p = next;
-    gpr_mu_lock(&p->mu);
+  polling_island_lock_pair(&p1, &p2);
+  if (p1 == p2) {
+    gpr_mu_unlock(&p1->mu);
+  } else {
+    gpr_mu_unlock(&p1->mu);
+    gpr_mu_unlock(&p2->mu);
   }
-  gpr_mu_unlock(&p->mu);
-
-  return p;
-}
 
-bool grpc_are_polling_islands_equal(void *p, void *q) {
-  p = get_polling_island(p);
-  q = get_polling_island(q);
-  return p == q;
+  return p1 == p2;
 }
 
 /*******************************************************************************