Add poll_object struct (and related changes to fix compilation errors). No other functionality changes
diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c
index 91041a7..f81125c 100644
--- a/src/core/lib/iomgr/ev_epoll_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_linux.c
@@ -90,10 +90,39 @@
struct polling_island;
+typedef enum {
+ POLL_OBJ_FD,
+ POLL_OBJ_POLLSET,
+ POLL_OBJ_POLLSET_SET
+} poll_obj_type;
+
+typedef struct poll_obj {
+ gpr_mu mu;
+ struct polling_island *pi;
+} poll_obj;
+
+const char *poll_obj_string(poll_obj_type po_type) {
+ switch (po_type) {
+ case POLL_OBJ_FD:
+ return "fd";
+ case POLL_OBJ_POLLSET:
+ return "pollset";
+ case POLL_OBJ_POLLSET_SET:
+ return "pollset_set";
+ }
+
+ GPR_UNREACHABLE_CODE(return "UNKNOWN");
+}
+
/*******************************************************************************
* Fd Declarations
*/
+
+#define FD_FROM_PO(po) ((grpc_fd *)(po))
+
struct grpc_fd {
+ poll_obj po;
+
int fd;
/* refst format:
bit 0 : 1=Active / 0=Orphaned
@@ -101,8 +130,6 @@
Ref/Unref by two to avoid altering the orphaned bit */
gpr_atm refst;
- gpr_mu mu;
-
/* Indicates that the fd is shutdown and that any pending read/write closures
should fail */
bool shutdown;
@@ -115,9 +142,6 @@
grpc_closure *read_closure;
grpc_closure *write_closure;
- /* The polling island to which this fd belongs to (protected by mu) */
- struct polling_island *polling_island;
-
struct grpc_fd *freelist_next;
grpc_closure *on_done_closure;
@@ -220,16 +244,14 @@
};
struct grpc_pollset {
- gpr_mu mu;
+ poll_obj po;
+
grpc_pollset_worker root_worker;
bool kicked_without_pollers;
bool shutting_down; /* Is the pollset shutting down ? */
bool finish_shutdown_called; /* Is the 'finish_shutdown_locked()' called ? */
grpc_closure *shutdown_done; /* Called after after shutdown is complete */
-
- /* The polling island to which this pollset belongs to */
- struct polling_island *polling_island;
};
/*******************************************************************************
@@ -242,7 +264,7 @@
* simplify the grpc_fd structure since we would no longer need to explicitly
* maintain the orphaned state */
struct grpc_pollset_set {
- gpr_mu mu;
+ poll_obj po;
size_t pollset_count;
size_t pollset_capacity;
@@ -916,7 +938,7 @@
while (fd_freelist != NULL) {
grpc_fd *fd = fd_freelist;
fd_freelist = fd_freelist->freelist_next;
- gpr_mu_destroy(&fd->mu);
+ gpr_mu_destroy(&fd->po.mu);
gpr_free(fd);
}
gpr_mu_destroy(&fd_freelist_mu);
@@ -934,13 +956,14 @@
if (new_fd == NULL) {
new_fd = gpr_malloc(sizeof(grpc_fd));
- gpr_mu_init(&new_fd->mu);
+ gpr_mu_init(&new_fd->po.mu);
}
- /* Note: It is not really needed to get the new_fd->mu lock here. If this is a
- newly created fd (or an fd we got from the freelist), no one else would be
- holding a lock to it anyway. */
- gpr_mu_lock(&new_fd->mu);
+ /* Note: It is not really needed to get the new_fd->po.mu lock here. If this
+ * is a newly created fd (or an fd we got from the freelist), no one else
+ * would be holding a lock to it anyway. */
+ gpr_mu_lock(&new_fd->po.mu);
+ new_fd->po.pi = NULL;
gpr_atm_rel_store(&new_fd->refst, (gpr_atm)1);
new_fd->fd = fd;
@@ -948,12 +971,11 @@
new_fd->orphaned = false;
new_fd->read_closure = CLOSURE_NOT_READY;
new_fd->write_closure = CLOSURE_NOT_READY;
- new_fd->polling_island = NULL;
new_fd->freelist_next = NULL;
new_fd->on_done_closure = NULL;
new_fd->read_notifier_pollset = NULL;
- gpr_mu_unlock(&new_fd->mu);
+ gpr_mu_unlock(&new_fd->po.mu);
char *fd_name;
gpr_asprintf(&fd_name, "%s fd=%d", name, fd);
@@ -971,11 +993,11 @@
static int fd_wrapped_fd(grpc_fd *fd) {
int ret_fd = -1;
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
if (!fd->orphaned) {
ret_fd = fd->fd;
}
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
return ret_fd;
}
@@ -987,7 +1009,7 @@
grpc_error *error = GRPC_ERROR_NONE;
polling_island *unref_pi = NULL;
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
fd->on_done_closure = on_done;
/* If release_fd is not NULL, we should be relinquishing control of the file
@@ -1007,25 +1029,25 @@
/* Remove the fd from the polling island:
- 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
+ linked list pointed by fd->po.pi). 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
+ - Set fd->po.pi to NULL (but remove the ref on the polling island
before doing this.) */
- if (fd->polling_island != NULL) {
- polling_island *pi_latest = polling_island_lock(fd->polling_island);
+ if (fd->po.pi != NULL) {
+ polling_island *pi_latest = polling_island_lock(fd->po.pi);
polling_island_remove_fd_locked(pi_latest, fd, is_fd_closed, &error);
gpr_mu_unlock(&pi_latest->mu);
- unref_pi = fd->polling_island;
- fd->polling_island = NULL;
+ unref_pi = fd->po.pi;
+ fd->po.pi = NULL;
}
grpc_exec_ctx_sched(exec_ctx, fd->on_done_closure, GRPC_ERROR_REF(error),
NULL);
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
UNREF_BY(fd, 2, reason); /* Drop the reference */
if (unref_pi != NULL) {
/* Unref stale polling island here, outside the fd lock above.
@@ -1090,23 +1112,23 @@
grpc_fd *fd) {
grpc_pollset *notifier = NULL;
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
notifier = fd->read_notifier_pollset;
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
return notifier;
}
static bool fd_is_shutdown(grpc_fd *fd) {
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
const bool r = fd->shutdown;
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
return r;
}
/* Might be called multiple times */
static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
/* Do the actual shutdown only once */
if (!fd->shutdown) {
fd->shutdown = true;
@@ -1117,28 +1139,28 @@
set_ready_locked(exec_ctx, fd, &fd->read_closure);
set_ready_locked(exec_ctx, fd, &fd->write_closure);
}
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
}
static void fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
grpc_closure *closure) {
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
notify_on_locked(exec_ctx, fd, &fd->read_closure, closure);
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
}
static void fd_notify_on_write(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
grpc_closure *closure) {
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
notify_on_locked(exec_ctx, fd, &fd->write_closure, closure);
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
}
static grpc_workqueue *fd_get_workqueue(grpc_fd *fd) {
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
grpc_workqueue *workqueue = GRPC_WORKQUEUE_REF(
- (grpc_workqueue *)fd->polling_island, "fd_get_workqueue");
- gpr_mu_unlock(&fd->mu);
+ (grpc_workqueue *)fd->po.pi, "fd_get_workqueue");
+ gpr_mu_unlock(&fd->po.mu);
return workqueue;
}
@@ -1278,8 +1300,9 @@
}
static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) {
- gpr_mu_init(&pollset->mu);
- *mu = &pollset->mu;
+ gpr_mu_init(&pollset->po.mu);
+ *mu = &pollset->po.mu;
+ pollset->po.pi = NULL;
pollset->root_worker.next = pollset->root_worker.prev = &pollset->root_worker;
pollset->kicked_without_pollers = false;
@@ -1287,8 +1310,6 @@
pollset->shutting_down = false;
pollset->finish_shutdown_called = false;
pollset->shutdown_done = NULL;
-
- pollset->polling_island = NULL;
}
/* Convert a timespec to milliseconds:
@@ -1318,26 +1339,26 @@
static void fd_become_readable(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
grpc_pollset *notifier) {
- /* Need the fd->mu since we might be racing with fd_notify_on_read */
- gpr_mu_lock(&fd->mu);
+ /* Need the fd->po.mu since we might be racing with fd_notify_on_read */
+ gpr_mu_lock(&fd->po.mu);
set_ready_locked(exec_ctx, fd, &fd->read_closure);
fd->read_notifier_pollset = notifier;
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
}
static void fd_become_writable(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
- /* Need the fd->mu since we might be racing with fd_notify_on_write */
- gpr_mu_lock(&fd->mu);
+ /* Need the fd->po.mu since we might be racing with fd_notify_on_write */
+ gpr_mu_lock(&fd->po.mu);
set_ready_locked(exec_ctx, fd, &fd->write_closure);
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
}
static void pollset_release_polling_island(grpc_exec_ctx *exec_ctx,
grpc_pollset *ps, char *reason) {
- if (ps->polling_island != NULL) {
- PI_UNREF(exec_ctx, ps->polling_island, reason);
+ if (ps->po.pi != NULL) {
+ PI_UNREF(exec_ctx, ps->po.pi, reason);
}
- ps->polling_island = NULL;
+ ps->po.pi = NULL;
}
static void finish_shutdown_locked(grpc_exec_ctx *exec_ctx,
@@ -1347,12 +1368,12 @@
pollset->finish_shutdown_called = true;
- /* Release the ref and set pollset->polling_island to NULL */
+ /* Release the ref and set pollset->po.pi to NULL */
pollset_release_polling_island(exec_ctx, pollset, "ps_shutdown");
grpc_exec_ctx_sched(exec_ctx, pollset->shutdown_done, GRPC_ERROR_NONE, NULL);
}
-/* pollset->mu lock must be held by the caller before calling this */
+/* pollset->po.mu lock must be held by the caller before calling this */
static void pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_closure *closure) {
GPR_TIMER_BEGIN("pollset_shutdown", 0);
@@ -1377,7 +1398,7 @@
* here */
static void pollset_destroy(grpc_pollset *pollset) {
GPR_ASSERT(!pollset_has_workers(pollset));
- gpr_mu_destroy(&pollset->mu);
+ gpr_mu_destroy(&pollset->po.mu);
}
static void pollset_reset(grpc_pollset *pollset) {
@@ -1387,7 +1408,7 @@
pollset->finish_shutdown_called = false;
pollset->kicked_without_pollers = false;
pollset->shutdown_done = NULL;
- GPR_ASSERT(pollset->polling_island == NULL);
+ GPR_ASSERT(pollset->po.pi == NULL);
}
static bool maybe_do_workqueue_work(grpc_exec_ctx *exec_ctx,
@@ -1427,7 +1448,7 @@
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
- latest polling island pointed by pollset->polling_island.
+ latest polling island pointed by pollset->po.pi
Since epoll_fd is immutable, we can read it without obtaining the polling
island lock. There is however a possibility that the polling island (from
@@ -1436,36 +1457,36 @@
right-away from epoll_wait() and pick up the latest polling_island the next
this function (i.e pollset_work_and_unlock()) is called */
- if (pollset->polling_island == NULL) {
- pollset->polling_island = polling_island_create(exec_ctx, NULL, error);
- if (pollset->polling_island == NULL) {
+ if (pollset->po.pi == NULL) {
+ pollset->po.pi = polling_island_create(exec_ctx, NULL, error);
+ if (pollset->po.pi == NULL) {
GPR_TIMER_END("pollset_work_and_unlock", 0);
return; /* Fatal error. We cannot continue */
}
- PI_ADD_REF(pollset->polling_island, "ps");
+ PI_ADD_REF(pollset->po.pi, "ps");
GRPC_POLLING_TRACE("pollset_work: pollset: %p created new pi: %p",
- (void *)pollset, (void *)pollset->polling_island);
+ (void *)pollset, (void *)pollset->po.pi);
}
- pi = polling_island_maybe_get_latest(pollset->polling_island);
+ pi = polling_island_maybe_get_latest(pollset->po.pi);
epoll_fd = pi->epoll_fd;
- /* Update the pollset->polling_island since the island being pointed by
- pollset->polling_island maybe older than the one pointed by pi) */
- if (pollset->polling_island != pi) {
+ /* Update the pollset->po.pi since the island being pointed by
+ pollset->po.pi maybe older than the one pointed by pi) */
+ if (pollset->po.pi != 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(exec_ctx, pollset->polling_island, "ps");
- pollset->polling_island = pi;
+ PI_UNREF(exec_ctx, pollset->po.pi, "ps");
+ pollset->po.pi = 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");
- gpr_mu_unlock(&pollset->mu);
+ gpr_mu_unlock(&pollset->po.mu);
/* If we get some workqueue work to do, it might end up completing an item on
the completion queue, so there's no need to poll... so we skip that and
@@ -1540,17 +1561,17 @@
GPR_ASSERT(pi != NULL);
/* 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
+ is important to use "pi" here (i.e our old copy of pollset->po.pi
that we got before releasing the polling island lock). This is because
- pollset->polling_island pointer might get udpated in other parts of the
+ pollset->po.pi 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(exec_ctx, pi, "ps_work");
GPR_TIMER_END("pollset_work_and_unlock", 0);
}
-/* pollset->mu lock must be held by the caller before calling this.
- The function pollset_work() may temporarily release the lock (pollset->mu)
+/* pollset->po.mu lock must be held by the caller before calling this.
+ The function pollset_work() may temporarily release the lock (pollset->po.mu)
during the course of its execution but it will always re-acquire the lock and
ensure that it is held by the time the function returns */
static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
@@ -1620,7 +1641,7 @@
&g_orig_sigmask, &error);
grpc_exec_ctx_flush(exec_ctx);
- gpr_mu_lock(&pollset->mu);
+ gpr_mu_lock(&pollset->po.mu);
/* Note: There is no need to reset worker.is_kicked to 0 since we are no
longer going to use this worker */
@@ -1640,9 +1661,9 @@
GPR_TIMER_MARK("pollset_work.finish_shutdown_locked", 0);
finish_shutdown_locked(exec_ctx, pollset);
- gpr_mu_unlock(&pollset->mu);
+ gpr_mu_unlock(&pollset->po.mu);
grpc_exec_ctx_flush(exec_ctx);
- gpr_mu_lock(&pollset->mu);
+ gpr_mu_lock(&pollset->po.mu);
}
*worker_hdl = NULL;
@@ -1656,42 +1677,159 @@
return error;
}
+#if 0
+static void add_poll_object(grpc_exec_ctx *exec_ctx, poll_obj *bag,
+ poll_obj *item, poll_obj_type bag_type,
+ poll_obj_type item_type) {
+ GPR_TIMER_BEGIN("add_poll_object", 0);
+
+ grpc_error *error = GRPC_ERROR_NONE;
+ polling_island *pi_new = NULL;
+
+ gpr_mu_lock(&bag->mu);
+ gpr_mu_lock(&item->mu);
+
+retry:
+ /*
+ * 1) If item->pi and bag->pi are both non-NULL and equal, do nothing
+ * 2) If item->pi and bag->pi are both NULL, create a new polling island (with
+ * a refcount of 2) and point item->pi and bag->pi to the new island
+ * 3) If exactly one of item->pi or bag->pi is NULL, update it to point to
+ * the other's non-NULL pi
+ * 4) Finally if item->pi and bag-pi are non-NULL and not-equal, merge the
+ * polling islands and update item->pi and bag->pi to point to the new
+ * island
+ */
+
+ /* Early out if we are trying to add an 'fd' to a 'bag' but the fd is already
+ * orphaned */
+ if (item_type == POLL_OBJ_FD && (FD_FROM_PO(item))->orphaned) {
+ gpr_mu_unlock(&item->mu);
+ gpr_mu_unlock(&bag->mu);
+ return;
+ }
+
+ if (item->pi == bag->pi) {
+ pi_new = item->pi;
+ if (pi_new == NULL) {
+ /* GPR_ASSERT(item->pi == bag->pi == NULL) */
+
+ /* If we are adding an fd to a bag (i.e pollset or pollset_set), then
+ * we need to do some extra work to make TSAN happy */
+ if (item_type == POLL_OBJ_FD) {
+ /* Unlock before creating a new polling island: the polling island will
+ create a workqueue which creates a file descriptor, and holding an fd
+ lock here can eventually cause a loop to appear to TSAN (making it
+ unhappy). We don't think it's a real loop (there's an epoch point
+ where that loop possibility disappears), but the advantages of
+ keeping TSAN happy outweigh any performance advantage we might have
+ by keeping the lock held. */
+ gpr_mu_unlock(&item->mu);
+ pi_new = polling_island_create(exec_ctx, FD_FROM_PO(item), &error);
+ gpr_mu_lock(&item->mu);
+
+ /* Need to reverify any assumptions made between the initial lock and
+ getting to this branch: if they've changed, we need to throw away our
+ work and figure things out again. */
+ if (item->pi != NULL) {
+ /* No need to lock 'pi_new' here since this is a new polling island
+ * and no one has a reference to it yet */
+ polling_island_remove_all_fds_locked(pi_new, true, &error);
+
+ /* Ref and unref so that the polling island gets deleted during unref
+ */
+ PI_ADD_REF(pi_new, "dance_of_destruction");
+ PI_UNREF(exec_ctx, pi_new, "dance_of_destruction");
+ goto retry;
+ }
+ } else {
+ pi_new = polling_island_create(exec_ctx, NULL, &error);
+ }
+ }
+ } else if (item->pi == NULL) {
+ /* GPR_ASSERT(bag->pi != NULL) */
+ /* Make pi_new point to latest pi*/
+ pi_new = polling_island_lock(bag->pi);
+
+ if (item_type == POLL_OBJ_FD) {
+ grpc_fd *fd = FD_FROM_PO(item);
+ polling_island_add_fds_locked(pi_new, &fd, 1, true, &error);
+ }
+
+ gpr_mu_unlock(&pi_new->mu);
+
+ } else if (bag->pi == NULL) {
+ /* GPR_ASSERT(item->pi != NULL) */
+ /* Make pi_new to point to latest pi */
+ pi_new = polling_island_lock(item->pi);
+ gpr_mu_unlock(&pi_new->mu);
+ } else {
+ pi_new = polling_island_merge(item->pi, bag->pi, &error);
+ }
+
+ /* At this point, pi_new is the polling island that both item->pi and bag->pi
+ MUST be pointing to */
+
+ if (item->pi != pi_new) {
+ PI_ADD_REF(pi_new, poll_obj_string(item_type));
+ if (item->pi != NULL) {
+ PI_UNREF(exec_ctx, item->pi, poll_obj_string(item_type));
+ }
+ item->pi = pi_new;
+ }
+
+ if (bag->pi != pi_new) {
+ PI_ADD_REF(pi_new, poll_obj_string(bag_type));
+ if (bag->pi != NULL) {
+ PI_UNREF(exec_ctx, bag->pi, poll_obj_string(bag_type));
+ }
+ bag->pi = pi_new;
+ }
+
+ gpr_mu_unlock(&item->mu);
+ gpr_mu_unlock(&bag->mu);
+
+ GRPC_LOG_IF_ERROR("add_poll_object", error);
+ GPR_TIMER_END("add_poll_object", 0);
+}
+#endif
+
static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_fd *fd) {
GPR_TIMER_BEGIN("pollset_add_fd", 0);
grpc_error *error = GRPC_ERROR_NONE;
- gpr_mu_lock(&pollset->mu);
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&pollset->po.mu);
+ gpr_mu_lock(&fd->po.mu);
polling_island *pi_new = NULL;
retry:
- /* 1) If fd->polling_island and pollset->polling_island are both non-NULL and
+ /* 1) If fd->po.pi and pollset->po.pi are both non-NULL and
* equal, do nothing.
- * 2) If fd->polling_island and pollset->polling_island are both NULL, create
+ * 2) If fd->po.pi and pollset->po.pi are both NULL, create
* a new polling island (with a refcount of 2) and make the polling_island
* fields in both fd and pollset to point to the new island
- * 3) If one of fd->polling_island or pollset->polling_island is NULL, update
+ * 3) If one of fd->po.pi or pollset->po.pi is NULL, update
* the NULL polling_island field to point to the non-NULL polling_island
* field (ensure that the refcount on the polling island is incremented by
* 1 to account for the newly added reference)
- * 4) Finally, if fd->polling_island and pollset->polling_island are non-NULL
+ * 4) Finally, if fd->po.pi and pollset->po.pi are non-NULL
* and different, merge both the polling islands and update the
* polling_island fields in both fd and pollset to point to the merged
* polling island.
*/
if (fd->orphaned) {
- gpr_mu_unlock(&fd->mu);
- gpr_mu_unlock(&pollset->mu);
+ gpr_mu_unlock(&fd->po.mu);
+ gpr_mu_unlock(&pollset->po.mu);
/* early out */
return;
}
- if (fd->polling_island == pollset->polling_island) {
- pi_new = fd->polling_island;
+ if (fd->po.pi == pollset->po.pi) {
+ pi_new = fd->po.pi;
if (pi_new == NULL) {
/* Unlock before creating a new polling island: the polling island will
create a workqueue which creates a file descriptor, and holding an fd
@@ -1700,13 +1838,13 @@
that loop possibility disappears), but the advantages of keeping TSAN
happy outweigh any performance advantage we might have by keeping the
lock held. */
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_unlock(&fd->po.mu);
pi_new = polling_island_create(exec_ctx, fd, &error);
- gpr_mu_lock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
/* Need to reverify any assumptions made between the initial lock and
getting to this branch: if they've changed, we need to throw away our
work and figure things out again. */
- if (fd->polling_island != NULL) {
+ if (fd->po.pi != NULL) {
GRPC_POLLING_TRACE(
"pollset_add_fd: Raced creating new polling island. pi_new: %p "
"(fd: %d, pollset: %p)",
@@ -1727,55 +1865,53 @@
(void *)pi_new, fd->fd, (void *)pollset);
}
}
- } else if (fd->polling_island == NULL) {
- pi_new = polling_island_lock(pollset->polling_island);
+ } else if (fd->po.pi == NULL) {
+ pi_new = polling_island_lock(pollset->po.pi);
polling_island_add_fds_locked(pi_new, &fd, 1, true, &error);
gpr_mu_unlock(&pi_new->mu);
GRPC_POLLING_TRACE(
"pollset_add_fd: fd->pi was NULL. pi_new: %p (fd: %d, pollset: %p, "
"pollset->pi: %p)",
- (void *)pi_new, fd->fd, (void *)pollset,
- (void *)pollset->polling_island);
- } else if (pollset->polling_island == NULL) {
- pi_new = polling_island_lock(fd->polling_island);
+ (void *)pi_new, fd->fd, (void *)pollset, (void *)pollset->po.pi);
+ } else if (pollset->po.pi == NULL) {
+ pi_new = polling_island_lock(fd->po.pi);
gpr_mu_unlock(&pi_new->mu);
GRPC_POLLING_TRACE(
"pollset_add_fd: pollset->pi was NULL. pi_new: %p (fd: %d, pollset: "
"%p, fd->pi: %p",
- (void *)pi_new, fd->fd, (void *)pollset, (void *)fd->polling_island);
+ (void *)pi_new, fd->fd, (void *)pollset, (void *)fd->po.pi);
} else {
- pi_new = polling_island_merge(fd->polling_island, pollset->polling_island,
- &error);
+ pi_new = polling_island_merge(fd->po.pi, pollset->po.pi, &error);
GRPC_POLLING_TRACE(
"pollset_add_fd: polling islands merged. pi_new: %p (fd: %d, pollset: "
"%p, fd->pi: %p, pollset->pi: %p)",
- (void *)pi_new, fd->fd, (void *)pollset, (void *)fd->polling_island,
- (void *)pollset->polling_island);
+ (void *)pi_new, fd->fd, (void *)pollset, (void *)fd->po.pi,
+ (void *)pollset->po.pi);
}
- /* At this point, pi_new is the polling island that both fd->polling_island
- and pollset->polling_island must be pointing to */
+ /* At this point, pi_new is the polling island that both fd->po.pi
+ and pollset->po.pi must be pointing to */
- if (fd->polling_island != pi_new) {
+ if (fd->po.pi != pi_new) {
PI_ADD_REF(pi_new, "fd");
- if (fd->polling_island != NULL) {
- PI_UNREF(exec_ctx, fd->polling_island, "fd");
+ if (fd->po.pi != NULL) {
+ PI_UNREF(exec_ctx, fd->po.pi, "fd");
}
- fd->polling_island = pi_new;
+ fd->po.pi = pi_new;
}
- if (pollset->polling_island != pi_new) {
+ if (pollset->po.pi != pi_new) {
PI_ADD_REF(pi_new, "ps");
- if (pollset->polling_island != NULL) {
- PI_UNREF(exec_ctx, pollset->polling_island, "ps");
+ if (pollset->po.pi != NULL) {
+ PI_UNREF(exec_ctx, pollset->po.pi, "ps");
}
- pollset->polling_island = pi_new;
+ pollset->po.pi = pi_new;
}
- gpr_mu_unlock(&fd->mu);
- gpr_mu_unlock(&pollset->mu);
+ gpr_mu_unlock(&fd->po.mu);
+ gpr_mu_unlock(&pollset->po.mu);
GRPC_LOG_IF_ERROR("pollset_add_fd", error);
@@ -1789,13 +1925,13 @@
static grpc_pollset_set *pollset_set_create(void) {
grpc_pollset_set *pollset_set = gpr_malloc(sizeof(*pollset_set));
memset(pollset_set, 0, sizeof(*pollset_set));
- gpr_mu_init(&pollset_set->mu);
+ gpr_mu_init(&pollset_set->po.mu);
return pollset_set;
}
static void pollset_set_destroy(grpc_pollset_set *pollset_set) {
size_t i;
- gpr_mu_destroy(&pollset_set->mu);
+ gpr_mu_destroy(&pollset_set->po.mu);
for (i = 0; i < pollset_set->fd_count; i++) {
GRPC_FD_UNREF(pollset_set->fds[i], "pollset_set");
}
@@ -1808,7 +1944,7 @@
static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx,
grpc_pollset_set *pollset_set, grpc_fd *fd) {
size_t i;
- gpr_mu_lock(&pollset_set->mu);
+ gpr_mu_lock(&pollset_set->po.mu);
if (pollset_set->fd_count == pollset_set->fd_capacity) {
pollset_set->fd_capacity = GPR_MAX(8, 2 * pollset_set->fd_capacity);
pollset_set->fds = gpr_realloc(
@@ -1822,13 +1958,13 @@
for (i = 0; i < pollset_set->pollset_set_count; i++) {
pollset_set_add_fd(exec_ctx, pollset_set->pollset_sets[i], fd);
}
- gpr_mu_unlock(&pollset_set->mu);
+ gpr_mu_unlock(&pollset_set->po.mu);
}
static void pollset_set_del_fd(grpc_exec_ctx *exec_ctx,
grpc_pollset_set *pollset_set, grpc_fd *fd) {
size_t i;
- gpr_mu_lock(&pollset_set->mu);
+ gpr_mu_lock(&pollset_set->po.mu);
for (i = 0; i < pollset_set->fd_count; i++) {
if (pollset_set->fds[i] == fd) {
pollset_set->fd_count--;
@@ -1841,14 +1977,14 @@
for (i = 0; i < pollset_set->pollset_set_count; i++) {
pollset_set_del_fd(exec_ctx, pollset_set->pollset_sets[i], fd);
}
- gpr_mu_unlock(&pollset_set->mu);
+ gpr_mu_unlock(&pollset_set->po.mu);
}
static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx,
grpc_pollset_set *pollset_set,
grpc_pollset *pollset) {
size_t i, j;
- gpr_mu_lock(&pollset_set->mu);
+ gpr_mu_lock(&pollset_set->po.mu);
if (pollset_set->pollset_count == pollset_set->pollset_capacity) {
pollset_set->pollset_capacity =
GPR_MAX(8, 2 * pollset_set->pollset_capacity);
@@ -1866,14 +2002,14 @@
}
}
pollset_set->fd_count = j;
- gpr_mu_unlock(&pollset_set->mu);
+ gpr_mu_unlock(&pollset_set->po.mu);
}
static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx,
grpc_pollset_set *pollset_set,
grpc_pollset *pollset) {
size_t i;
- gpr_mu_lock(&pollset_set->mu);
+ gpr_mu_lock(&pollset_set->po.mu);
for (i = 0; i < pollset_set->pollset_count; i++) {
if (pollset_set->pollsets[i] == pollset) {
pollset_set->pollset_count--;
@@ -1882,14 +2018,14 @@
break;
}
}
- gpr_mu_unlock(&pollset_set->mu);
+ gpr_mu_unlock(&pollset_set->po.mu);
}
static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx,
grpc_pollset_set *bag,
grpc_pollset_set *item) {
size_t i, j;
- gpr_mu_lock(&bag->mu);
+ gpr_mu_lock(&bag->po.mu);
if (bag->pollset_set_count == bag->pollset_set_capacity) {
bag->pollset_set_capacity = GPR_MAX(8, 2 * bag->pollset_set_capacity);
bag->pollset_sets =
@@ -1906,14 +2042,14 @@
}
}
bag->fd_count = j;
- gpr_mu_unlock(&bag->mu);
+ gpr_mu_unlock(&bag->po.mu);
}
static void pollset_set_del_pollset_set(grpc_exec_ctx *exec_ctx,
grpc_pollset_set *bag,
grpc_pollset_set *item) {
size_t i;
- gpr_mu_lock(&bag->mu);
+ gpr_mu_lock(&bag->po.mu);
for (i = 0; i < bag->pollset_set_count; i++) {
if (bag->pollset_sets[i] == item) {
bag->pollset_set_count--;
@@ -1922,7 +2058,7 @@
break;
}
}
- gpr_mu_unlock(&bag->mu);
+ gpr_mu_unlock(&bag->po.mu);
}
/* Test helper functions
@@ -1930,9 +2066,9 @@
void *grpc_fd_get_polling_island(grpc_fd *fd) {
polling_island *pi;
- gpr_mu_lock(&fd->mu);
- pi = fd->polling_island;
- gpr_mu_unlock(&fd->mu);
+ gpr_mu_lock(&fd->po.mu);
+ pi = fd->po.pi;
+ gpr_mu_unlock(&fd->po.mu);
return pi;
}
@@ -1940,9 +2076,9 @@
void *grpc_pollset_get_polling_island(grpc_pollset *ps) {
polling_island *pi;
- gpr_mu_lock(&ps->mu);
- pi = ps->polling_island;
- gpr_mu_unlock(&ps->mu);
+ gpr_mu_lock(&ps->po.mu);
+ pi = ps->po.pi;
+ gpr_mu_unlock(&ps->po.mu);
return pi;
}