Split event notification out of ev_epoll_linux.c
diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c
index e5cf54f..a7105a6 100644
--- a/src/core/lib/iomgr/ev_epoll_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_linux.c
@@ -56,6 +56,7 @@
#include "src/core/lib/iomgr/ev_posix.h"
#include "src/core/lib/iomgr/iomgr_internal.h"
+#include "src/core/lib/iomgr/lockfree_event.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/iomgr/wakeup_fd_posix.h"
#include "src/core/lib/iomgr/workqueue.h"
@@ -141,52 +142,11 @@
Ref/Unref by two to avoid altering the orphaned bit */
gpr_atm refst;
- /* Internally stores data of type (grpc_error *). If the FD is shutdown, this
- contains reason for shutdown (i.e a pointer to grpc_error) ORed with
- FD_SHUTDOWN_BIT. Since address allocations are word-aligned, the lower bit
- of (grpc_error *) addresses is guaranteed to be zero. Even if the
- (grpc_error *), is of special types like GRPC_ERROR_NONE, GRPC_ERROR_OOM
- etc, the lower bit is guaranteed to be zero.
-
- Once an fd is shutdown, any pending or future read/write closures on the
- fd should fail */
- gpr_atm shutdown_error;
-
/* The fd is either closed or we relinquished control of it. In either
cases, this indicates that the 'fd' on this structure is no longer
valid */
bool orphaned;
- /* Closures to call when the fd is readable or writable respectively. These
- fields contain one of the following values:
- CLOSURE_READY : The fd has an I/O event of interest but there is no
- closure yet to execute
-
- CLOSURE_NOT_READY : The fd has no I/O event of interest
-
- closure ptr : The closure to be executed when the fd has an I/O
- event of interest
-
- shutdown_error | FD_SHUTDOWN_BIT :
- 'shutdown_error' field ORed with FD_SHUTDOWN_BIT.
- This indicates that the fd is shutdown. Since all
- memory allocations are word-aligned, the lower two
- bits of the shutdown_error pointer are always 0. So
- it is safe to OR these with FD_SHUTDOWN_BIT
-
- Valid state transitions:
-
- <closure ptr> <-----3------ CLOSURE_NOT_READY ----1----> CLOSURE_READY
- | | ^ | ^ | |
- | | | | | | |
- | +--------------4----------+ 6 +---------2---------------+ |
- | | |
- | v |
- +-----5-------> [shutdown_error | FD_SHUTDOWN_BIT] <----7---------+
-
- For 1, 4 : See set_ready() function
- For 2, 3 : See notify_on() function
- For 5,6,7: See set_shutdown() function */
gpr_atm read_closure;
gpr_atm write_closure;
@@ -218,11 +178,6 @@
static void fd_global_init(void);
static void fd_global_shutdown(void);
-#define CLOSURE_NOT_READY ((gpr_atm)0)
-#define CLOSURE_READY ((gpr_atm)2)
-
-#define FD_SHUTDOWN_BIT 1
-
/*******************************************************************************
* Polling island Declarations
*/
@@ -949,10 +904,8 @@
fd_freelist = fd;
grpc_iomgr_unregister_object(&fd->iomgr_object);
- grpc_error *err = (grpc_error *)gpr_atm_acq_load(&fd->shutdown_error);
- /* Clear the least significant bit if it set (in case fd was shutdown) */
- err = (grpc_error *)((intptr_t)err & ~FD_SHUTDOWN_BIT);
- GRPC_ERROR_UNREF(err);
+ grpc_lfev_destroy(&fd->read_closure);
+ grpc_lfev_destroy(&fd->write_closure);
gpr_mu_unlock(&fd_freelist_mu);
} else {
@@ -1018,8 +971,8 @@
new_fd->fd = fd;
gpr_atm_no_barrier_store(&new_fd->shutdown_error, (gpr_atm)GRPC_ERROR_NONE);
new_fd->orphaned = false;
- gpr_atm_no_barrier_store(&new_fd->read_closure, CLOSURE_NOT_READY);
- gpr_atm_no_barrier_store(&new_fd->write_closure, CLOSURE_NOT_READY);
+ grpc_lfev_init(&new_fd->read_closure);
+ grpc_lfev_init(&new_fd->write_closure);
gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm)NULL);
new_fd->freelist_next = NULL;
@@ -1105,153 +1058,6 @@
GRPC_ERROR_UNREF(error);
}
-static void notify_on(grpc_exec_ctx *exec_ctx, grpc_fd *fd, gpr_atm *state,
- grpc_closure *closure) {
- while (true) {
- gpr_atm curr = gpr_atm_no_barrier_load(state);
- switch (curr) {
- case CLOSURE_NOT_READY: {
- /* CLOSURE_NOT_READY -> <closure>.
-
- We're guaranteed by API that there's an acquire barrier before here,
- so there's no need to double-dip and this can be a release-only.
-
- The release itself pairs with the acquire half of a set_ready full
- barrier. */
- if (gpr_atm_rel_cas(state, CLOSURE_NOT_READY, (gpr_atm)closure)) {
- return; /* Successful. Return */
- }
-
- break; /* retry */
- }
-
- case CLOSURE_READY: {
- /* Change the state to CLOSURE_NOT_READY. Schedule the closure if
- successful. If not, the state most likely transitioned to shutdown.
- We should retry.
-
- This can be a no-barrier cas since the state is being transitioned to
- CLOSURE_NOT_READY; set_ready and set_shutdown do not schedule any
- closure when transitioning out of CLOSURE_NO_READY state (i.e there
- is no other code that needs to 'happen-after' this) */
- if (gpr_atm_no_barrier_cas(state, CLOSURE_READY, CLOSURE_NOT_READY)) {
- grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_NONE);
- return; /* Successful. Return */
- }
-
- break; /* retry */
- }
-
- default: {
- /* 'curr' is either a closure or the fd is shutdown(in which case 'curr'
- contains a pointer to the shutdown-error). If the fd is shutdown,
- schedule the closure with the shutdown error */
- if ((curr & FD_SHUTDOWN_BIT) > 0) {
- grpc_error *shutdown_err = (grpc_error *)(curr & ~FD_SHUTDOWN_BIT);
- grpc_closure_sched(exec_ctx, closure,
- GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
- "FD Shutdown", &shutdown_err, 1));
- return;
- }
-
- /* There is already a closure!. This indicates a bug in the code */
- gpr_log(GPR_ERROR,
- "notify_on called with a previous callback still pending");
- abort();
- }
- }
- }
-
- GPR_UNREACHABLE_CODE(return );
-}
-
-static void set_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, gpr_atm *state,
- grpc_error *shutdown_err) {
- gpr_atm new_state = (gpr_atm)shutdown_err | FD_SHUTDOWN_BIT;
-
- while (true) {
- gpr_atm curr = gpr_atm_no_barrier_load(state);
- switch (curr) {
- case CLOSURE_READY:
- case CLOSURE_NOT_READY:
- /* Need a full barrier here so that the initial load in notify_on
- doesn't need a barrier */
- if (gpr_atm_full_cas(state, curr, new_state)) {
- return; /* early out */
- }
- break; /* retry */
-
- default: {
- /* 'curr' is either a closure or the fd is already shutdown */
-
- /* If fd is already shutdown, we are done */
- if ((curr & FD_SHUTDOWN_BIT) > 0) {
- return;
- }
-
- /* Fd is not shutdown. Schedule the closure and move the state to
- shutdown state.
- Needs an acquire to pair with setting the closure (and get a
- happens-after on that edge), and a release to pair with anything
- loading the shutdown state. */
- if (gpr_atm_full_cas(state, curr, new_state)) {
- grpc_closure_sched(exec_ctx, (grpc_closure *)curr,
- GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
- "FD Shutdown", &shutdown_err, 1));
- return;
- }
-
- /* 'curr' was a closure but now changed to a different state. We will
- have to retry */
- break;
- }
- }
- }
-
- GPR_UNREACHABLE_CODE(return );
-}
-
-static void set_ready(grpc_exec_ctx *exec_ctx, grpc_fd *fd, gpr_atm *state) {
- while (true) {
- gpr_atm curr = gpr_atm_no_barrier_load(state);
-
- switch (curr) {
- case CLOSURE_READY: {
- /* Already ready. We are done here */
- return;
- }
-
- case CLOSURE_NOT_READY: {
- /* No barrier required as we're transitioning to a state that does not
- involve a closure */
- if (gpr_atm_no_barrier_cas(state, CLOSURE_NOT_READY, CLOSURE_READY)) {
- return; /* early out */
- }
- break; /* retry */
- }
-
- default: {
- /* 'curr' is either a closure or the fd is shutdown */
- if ((curr & FD_SHUTDOWN_BIT) > 0) {
- /* The fd is shutdown. Do nothing */
- return;
- }
- /* Full cas: acquire pairs with this cas' release in the event of a
- spurious set_ready; release pairs with this or the acquire in
- notify_on (or set_shutdown) */
- else if (gpr_atm_full_cas(state, curr, CLOSURE_NOT_READY)) {
- grpc_closure_sched(exec_ctx, (grpc_closure *)curr, GRPC_ERROR_NONE);
- return;
- }
- /* else the state changed again (only possible by either a racing
- set_ready or set_shutdown functions. In both these cases, the closure
- would have been scheduled for execution. So we are done here */
- return;
- }
- }
- }
-}
-
static grpc_pollset *fd_get_read_notifier_pollset(grpc_exec_ctx *exec_ctx,
grpc_fd *fd) {
gpr_atm notifier = gpr_atm_acq_load(&fd->read_notifier_pollset);
@@ -1259,23 +1065,16 @@
}
static bool fd_is_shutdown(grpc_fd *fd) {
- grpc_error *err = (grpc_error *)gpr_atm_acq_load(&fd->shutdown_error);
- return (((intptr_t)err & FD_SHUTDOWN_BIT) > 0);
+ return grpc_lfev_is_shutdown(&fd->read_closure);
}
/* Might be called multiple times */
static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why) {
- /* Store the shutdown error ORed with FD_SHUTDOWN_BIT in fd->shutdown_error */
- if (gpr_atm_rel_cas(&fd->shutdown_error, (gpr_atm)GRPC_ERROR_NONE,
- (gpr_atm)why | FD_SHUTDOWN_BIT)) {
+ if (grpc_lfev_set_shutdown(&fd->read_closure, GRPC_ERROR_REF(why))) {
shutdown(fd->fd, SHUT_RDWR);
-
- set_shutdown(exec_ctx, fd, &fd->read_closure, why);
- set_shutdown(exec_ctx, fd, &fd->write_closure, why);
- } else {
- /* Shutdown already called */
- GRPC_ERROR_UNREF(why);
+ grpc_lfev_set_shutdown(&fd->write_closure, GRPC_ERROR_REF(why));
}
+ GRPC_ERROR_UNREF(why);
}
static void fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd,