aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/iomgr
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/iomgr')
-rw-r--r--src/core/iomgr/fd_posix.c53
-rw-r--r--src/core/iomgr/fd_posix.h10
-rw-r--r--src/core/iomgr/iomgr.c68
-rw-r--r--src/core/iomgr/iomgr.h37
-rw-r--r--src/core/iomgr/iomgr_internal.h4
-rw-r--r--src/core/iomgr/pollset_posix.c7
-rw-r--r--src/core/iomgr/pollset_windows.c10
-rw-r--r--src/core/iomgr/socket_windows.c11
-rw-r--r--src/core/iomgr/socket_windows.h4
-rw-r--r--src/core/iomgr/tcp_posix.c7
10 files changed, 123 insertions, 88 deletions
diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c
index b697fcc64a..5bbf171c6d 100644
--- a/src/core/iomgr/fd_posix.c
+++ b/src/core/iomgr/fd_posix.c
@@ -91,6 +91,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,8 +117,7 @@ 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();
} else {
@@ -180,8 +180,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 +195,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 +233,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 +249,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 +267,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 +280,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..a7c9c34802 100644
--- a/src/core/iomgr/fd_posix.h
+++ b/src/core/iomgr/fd_posix.h
@@ -40,11 +40,6 @@
#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,9 +91,10 @@ 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];
};
/* Create a wrapped file descriptor.
diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c
index d22542fc91..b983a89daa 100644
--- a/src/core/iomgr/iomgr.c
+++ b/src/core/iomgr/iomgr.c
@@ -42,17 +42,10 @@
#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;
@@ -66,12 +59,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 {
@@ -103,7 +95,7 @@ void grpc_iomgr_init(void) {
}
void grpc_iomgr_shutdown(void) {
- delayed_callback *cb;
+ grpc_iomgr_closure *closure;
gpr_timespec shutdown_deadline =
gpr_time_add(gpr_now(), gpr_time_from_seconds(10));
@@ -114,13 +106,12 @@ void grpc_iomgr_shutdown(void) {
gpr_log(GPR_DEBUG, "Waiting for %d iomgr objects to be destroyed%s", g_refs,
g_cbs_head ? " and executing final callbacks" : "");
while (g_cbs_head) {
- cb = g_cbs_head;
- g_cbs_head = cb->next;
+ 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, 0);
- gpr_free(cb);
+ closure->cb(closure->cb_arg, 0);
gpr_mu_lock(&g_mu);
}
if (g_refs) {
@@ -167,42 +158,48 @@ void grpc_iomgr_unref(void) {
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 +208,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..25a731d58c 100644
--- a/src/core/iomgr/iomgr_internal.h
+++ b/src/core/iomgr/iomgr_internal.h
@@ -35,12 +35,10 @@
#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>
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);
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index 826c792990..a8e6069002 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -257,6 +257,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) {
@@ -279,7 +280,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;
}
@@ -363,7 +364,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/socket_windows.c b/src/core/iomgr/socket_windows.c
index ee5150a696..805fa8a4fc 100644
--- a/src/core/iomgr/socket_windows.c
+++ b/src/core/iomgr/socket_windows.c
@@ -39,7 +39,6 @@
#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"
@@ -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;
diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h
index b27eb14219..d5fee39604 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.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,6 +95,8 @@ 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;
} grpc_winsocket;
/* Create a wrapped windows handle. This takes ownership of it, meaning that
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;
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-10-24 12:16:46 +0000