aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/iomgr
diff options
context:
space:
mode:
authorGravatar Sree Kuchibhotla <sreek@google.com>2016-02-01 14:04:19 -0800
committerGravatar Sree Kuchibhotla <sreek@google.com>2016-02-01 14:04:19 -0800
commita0a8eaab0e86148f40fb629a36c0529eea9e1b35 (patch)
treea7c28c2b60061afbdc52137421007a63e3ea1eaf /src/core/iomgr
parent369a04ace686d2db7fff8a39473680566da6700a (diff)
parentc1fdfec641b2b27c553a8b0bb00b47e56e23bfa1 (diff)
Merge branch 'master' into server_try_cancel_api
Diffstat (limited to 'src/core/iomgr')
-rw-r--r--src/core/iomgr/closure.c8
-rw-r--r--src/core/iomgr/closure.h9
-rw-r--r--src/core/iomgr/exec_ctx.c18
-rw-r--r--src/core/iomgr/exec_ctx.h17
-rw-r--r--src/core/iomgr/executor.c8
-rw-r--r--src/core/iomgr/executor.h4
-rw-r--r--src/core/iomgr/fd_posix.c6
-rw-r--r--src/core/iomgr/iocp_windows.c4
-rw-r--r--src/core/iomgr/pollset_multipoller_with_epoll.c6
-rw-r--r--src/core/iomgr/pollset_posix.c13
-rw-r--r--src/core/iomgr/pollset_windows.c4
-rw-r--r--src/core/iomgr/resolve_address_posix.c4
-rw-r--r--src/core/iomgr/resolve_address_windows.c4
-rw-r--r--src/core/iomgr/tcp_client_posix.c14
-rw-r--r--src/core/iomgr/tcp_client_windows.c8
-rw-r--r--src/core/iomgr/tcp_posix.c14
-rw-r--r--src/core/iomgr/tcp_server_posix.c9
-rw-r--r--src/core/iomgr/tcp_server_windows.c6
-rw-r--r--src/core/iomgr/tcp_windows.c20
-rw-r--r--src/core/iomgr/timer.c6
-rw-r--r--src/core/iomgr/workqueue.h6
-rw-r--r--src/core/iomgr/workqueue_posix.c6
22 files changed, 103 insertions, 91 deletions
diff --git a/src/core/iomgr/closure.c b/src/core/iomgr/closure.c
index 7646a88ac5..3a96f7385f 100644
--- a/src/core/iomgr/closure.c
+++ b/src/core/iomgr/closure.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -43,7 +43,7 @@ void grpc_closure_init(grpc_closure *closure, grpc_iomgr_cb_func cb,
}
void grpc_closure_list_add(grpc_closure_list *closure_list,
- grpc_closure *closure, int success) {
+ grpc_closure *closure, bool success) {
if (closure == NULL) return;
closure->final_data = (success != 0);
if (closure_list->head == NULL) {
@@ -54,7 +54,7 @@ void grpc_closure_list_add(grpc_closure_list *closure_list,
closure_list->tail = closure;
}
-int grpc_closure_list_empty(grpc_closure_list closure_list) {
+bool grpc_closure_list_empty(grpc_closure_list closure_list) {
return closure_list.head == NULL;
}
@@ -77,7 +77,7 @@ typedef struct {
grpc_closure wrapper;
} wrapped_closure;
-static void closure_wrapper(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void closure_wrapper(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
wrapped_closure *wc = arg;
grpc_iomgr_cb_func cb = wc->cb;
void *cb_arg = wc->cb_arg;
diff --git a/src/core/iomgr/closure.h b/src/core/iomgr/closure.h
index 98ef91e1db..ea96c19c71 100644
--- a/src/core/iomgr/closure.h
+++ b/src/core/iomgr/closure.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -35,6 +35,7 @@
#define GRPC_INTERNAL_CORE_IOMGR_CLOSURE_H
#include <grpc/support/port_platform.h>
+#include <stdbool.h>
struct grpc_closure;
typedef struct grpc_closure grpc_closure;
@@ -54,7 +55,7 @@ typedef struct grpc_closure_list {
* \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)(grpc_exec_ctx *exec_ctx, void *arg,
- int success);
+ bool success);
/** A closure over a grpc_iomgr_cb_func. */
struct grpc_closure {
@@ -83,13 +84,13 @@ grpc_closure *grpc_closure_create(grpc_iomgr_cb_func cb, void *cb_arg);
/** add \a closure to the end of \a list and set \a closure's success to \a
* success */
void grpc_closure_list_add(grpc_closure_list *list, grpc_closure *closure,
- int success);
+ bool success);
/** append all closures from \a src to \a dst and empty \a src. */
void grpc_closure_list_move(grpc_closure_list *src, grpc_closure_list *dst);
/** return whether \a list is empty. */
-int grpc_closure_list_empty(grpc_closure_list list);
+bool grpc_closure_list_empty(grpc_closure_list list);
/** return the next pointer for a queued closure list */
grpc_closure *grpc_closure_next(grpc_closure *closure);
diff --git a/src/core/iomgr/exec_ctx.c b/src/core/iomgr/exec_ctx.c
index 6059a6031c..1fd79f6eba 100644
--- a/src/core/iomgr/exec_ctx.c
+++ b/src/core/iomgr/exec_ctx.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -37,16 +37,16 @@
#include "src/core/profiling/timers.h"
-int grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx) {
- int did_something = 0;
+bool grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx) {
+ bool did_something = 0;
GPR_TIMER_BEGIN("grpc_exec_ctx_flush", 0);
while (!grpc_closure_list_empty(exec_ctx->closure_list)) {
grpc_closure *c = exec_ctx->closure_list.head;
exec_ctx->closure_list.head = exec_ctx->closure_list.tail = NULL;
while (c != NULL) {
- int success = (int)(c->final_data & 1);
+ bool success = (bool)(c->final_data & 1);
grpc_closure *next = (grpc_closure *)(c->final_data & ~(uintptr_t)1);
- did_something++;
+ did_something = true;
GPR_TIMER_BEGIN("grpc_exec_ctx_flush.cb", 0);
c->cb(exec_ctx, c->cb_arg, success);
GPR_TIMER_END("grpc_exec_ctx_flush.cb", 0);
@@ -62,11 +62,15 @@ void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx) {
}
void grpc_exec_ctx_enqueue(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
- int success) {
+ bool success,
+ grpc_workqueue *offload_target_or_null) {
+ GPR_ASSERT(offload_target_or_null == NULL);
grpc_closure_list_add(&exec_ctx->closure_list, closure, success);
}
void grpc_exec_ctx_enqueue_list(grpc_exec_ctx *exec_ctx,
- grpc_closure_list *list) {
+ grpc_closure_list *list,
+ grpc_workqueue *offload_target_or_null) {
+ GPR_ASSERT(offload_target_or_null == NULL);
grpc_closure_list_move(list, &exec_ctx->closure_list);
}
diff --git a/src/core/iomgr/exec_ctx.h b/src/core/iomgr/exec_ctx.h
index 43df488094..9a9b2e55fa 100644
--- a/src/core/iomgr/exec_ctx.h
+++ b/src/core/iomgr/exec_ctx.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -57,22 +57,29 @@ struct grpc_exec_ctx {
grpc_closure_list closure_list;
};
+/** A workqueue represents a list of work to be executed asynchronously.
+ Forward declared here to avoid a circular dependency with workqueue.h. */
+struct grpc_workqueue;
+typedef struct grpc_workqueue grpc_workqueue;
+
#define GRPC_EXEC_CTX_INIT \
{ GRPC_CLOSURE_LIST_INIT }
/** Flush any work that has been enqueued onto this grpc_exec_ctx.
* Caller must guarantee that no interfering locks are held.
- * Returns 1 if work was performed, 0 otherwise. */
-int grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx);
+ * Returns true if work was performed, false otherwise. */
+bool grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx);
/** Finish any pending work for a grpc_exec_ctx. Must be called before
* the instance is destroyed, or work may be lost. */
void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx);
/** Add a closure to be executed at the next flush/finish point */
void grpc_exec_ctx_enqueue(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
- int success);
+ bool success,
+ grpc_workqueue *offload_target_or_null);
/** Add a list of closures to be executed at the next flush/finish point.
* Leaves \a list empty. */
void grpc_exec_ctx_enqueue_list(grpc_exec_ctx *exec_ctx,
- grpc_closure_list *list);
+ grpc_closure_list *list,
+ grpc_workqueue *offload_target_or_null);
#endif
diff --git a/src/core/iomgr/executor.c b/src/core/iomgr/executor.c
index 00c68f7828..f22d8f30ac 100644
--- a/src/core/iomgr/executor.c
+++ b/src/core/iomgr/executor.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -77,7 +77,7 @@ static void closure_exec_thread_func(void *ignored) {
gpr_mu_unlock(&g_executor.mu);
break;
} else {
- grpc_exec_ctx_enqueue_list(&exec_ctx, &g_executor.closures);
+ grpc_exec_ctx_enqueue_list(&exec_ctx, &g_executor.closures, NULL);
}
gpr_mu_unlock(&g_executor.mu);
grpc_exec_ctx_flush(&exec_ctx);
@@ -112,7 +112,7 @@ static void maybe_spawn_locked() {
g_executor.pending_join = 1;
}
-void grpc_executor_enqueue(grpc_closure *closure, int success) {
+void grpc_executor_enqueue(grpc_closure *closure, bool success) {
gpr_mu_lock(&g_executor.mu);
if (g_executor.shutting_down == 0) {
grpc_closure_list_add(&g_executor.closures, closure, success);
@@ -133,7 +133,7 @@ void grpc_executor_shutdown() {
* list below because we aren't accepting new work */
/* Execute pending callbacks, some may be performing cleanups */
- grpc_exec_ctx_enqueue_list(&exec_ctx, &g_executor.closures);
+ grpc_exec_ctx_enqueue_list(&exec_ctx, &g_executor.closures, NULL);
grpc_exec_ctx_finish(&exec_ctx);
GPR_ASSERT(grpc_closure_list_empty(g_executor.closures));
if (pending_join) {
diff --git a/src/core/iomgr/executor.h b/src/core/iomgr/executor.h
index 6da446ae9c..aac057ddf5 100644
--- a/src/core/iomgr/executor.h
+++ b/src/core/iomgr/executor.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -45,7 +45,7 @@ void grpc_executor_init();
/** Enqueue \a closure for its eventual execution of \a f(arg) on a separate
* thread */
-void grpc_executor_enqueue(grpc_closure *closure, int success);
+void grpc_executor_enqueue(grpc_closure *closure, bool success);
/** Shutdown the executor, running all pending work as part of the call */
void grpc_executor_shutdown();
diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c
index 89c938bc04..85eadd754b 100644
--- a/src/core/iomgr/fd_posix.c
+++ b/src/core/iomgr/fd_posix.c
@@ -218,7 +218,7 @@ static void close_fd_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
} else {
grpc_remove_fd_from_all_epoll_sets(fd->fd);
}
- grpc_exec_ctx_enqueue(exec_ctx, fd->on_done_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, fd->on_done_closure, true, NULL);
}
int grpc_fd_wrapped_fd(grpc_fd *fd) {
@@ -273,7 +273,7 @@ static void notify_on_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
} else if (*st == CLOSURE_READY) {
/* already ready ==> queue the closure to run immediately */
*st = CLOSURE_NOT_READY;
- grpc_exec_ctx_enqueue(exec_ctx, closure, !fd->shutdown);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, !fd->shutdown, NULL);
maybe_wake_one_watcher_locked(fd);
} else {
/* upcallptr was set to a different closure. This is an error! */
@@ -296,7 +296,7 @@ static int set_ready_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
return 0;
} else {
/* waiting ==> queue closure */
- grpc_exec_ctx_enqueue(exec_ctx, *st, !fd->shutdown);
+ grpc_exec_ctx_enqueue(exec_ctx, *st, !fd->shutdown, NULL);
*st = CLOSURE_NOT_READY;
return 1;
}
diff --git a/src/core/iomgr/iocp_windows.c b/src/core/iomgr/iocp_windows.c
index 6cbe7d2fd4..96b6f81024 100644
--- a/src/core/iomgr/iocp_windows.c
+++ b/src/core/iomgr/iocp_windows.c
@@ -120,7 +120,7 @@ void grpc_iocp_work(grpc_exec_ctx *exec_ctx, gpr_timespec deadline) {
info->has_pending_iocp = 1;
}
gpr_mu_unlock(&socket->state_mu);
- grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL);
}
void grpc_iocp_init(void) {
@@ -183,7 +183,7 @@ static void socket_notify_on_iocp(grpc_exec_ctx *exec_ctx,
gpr_mu_lock(&socket->state_mu);
if (info->has_pending_iocp) {
info->has_pending_iocp = 0;
- grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL);
} else {
info->closure = closure;
}
diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c
index d117485327..4acae2bb71 100644
--- a/src/core/iomgr/pollset_multipoller_with_epoll.c
+++ b/src/core/iomgr/pollset_multipoller_with_epoll.c
@@ -141,7 +141,7 @@ static void finally_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
}
static void perform_delayed_add(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_status) {
+ bool iomgr_status) {
delayed_add *da = arg;
if (!grpc_fd_is_orphaned(da->fd)) {
@@ -154,7 +154,7 @@ static void perform_delayed_add(grpc_exec_ctx *exec_ctx, void *arg,
/* We don't care about this pollset anymore. */
if (da->pollset->in_flight_cbs == 0 && !da->pollset->called_shutdown) {
da->pollset->called_shutdown = 1;
- grpc_exec_ctx_enqueue(exec_ctx, da->pollset->shutdown_done, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, da->pollset->shutdown_done, true, NULL);
}
}
gpr_mu_unlock(&da->pollset->mu);
@@ -178,7 +178,7 @@ static void multipoll_with_epoll_pollset_add_fd(grpc_exec_ctx *exec_ctx,
GRPC_FD_REF(fd, "delayed_add");
grpc_closure_init(&da->closure, perform_delayed_add, da);
pollset->in_flight_cbs++;
- grpc_exec_ctx_enqueue(exec_ctx, &da->closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &da->closure, true, NULL);
}
}
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index c325b634ae..a8e2e22977 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -243,7 +243,7 @@ void grpc_pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
GPR_ASSERT(grpc_closure_list_empty(pollset->idle_jobs));
pollset->vtable->finish_shutdown(pollset);
- grpc_exec_ctx_enqueue(exec_ctx, pollset->shutdown_done, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pollset->shutdown_done, true, NULL);
}
void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
@@ -271,7 +271,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
if (!grpc_pollset_has_workers(pollset) &&
!grpc_closure_list_empty(pollset->idle_jobs)) {
GPR_TIMER_MARK("grpc_pollset_work.idle_jobs", 0);
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs);
+ grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
goto done;
}
/* Check alarms - these are a global resource so we just ping
@@ -365,7 +365,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
* TODO(dklempner): Can we refactor the shutdown logic to avoid this? */
gpr_mu_lock(&pollset->mu);
} else if (!grpc_closure_list_empty(pollset->idle_jobs)) {
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs);
+ grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
gpr_mu_unlock(&pollset->mu);
grpc_exec_ctx_flush(exec_ctx);
gpr_mu_lock(&pollset->mu);
@@ -381,7 +381,7 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
pollset->shutdown_done = closure;
grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST);
if (!grpc_pollset_has_workers(pollset)) {
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs);
+ grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
}
if (!pollset->called_shutdown && pollset->in_flight_cbs == 0 &&
!grpc_pollset_has_workers(pollset)) {
@@ -419,7 +419,8 @@ typedef struct grpc_unary_promote_args {
grpc_closure promotion_closure;
} grpc_unary_promote_args;
-static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args, int success) {
+static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args,
+ bool success) {
grpc_unary_promote_args *up_args = args;
const grpc_pollset_vtable *original_vtable = up_args->original_vtable;
grpc_pollset *pollset = up_args->pollset;
diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c
index bfd9e69a16..2e650ca939 100644
--- a/src/core/iomgr/pollset_windows.c
+++ b/src/core/iomgr/pollset_windows.c
@@ -107,7 +107,7 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
pollset->shutting_down = 1;
grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST);
if (!pollset->is_iocp_worker) {
- grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL);
} else {
pollset->on_shutdown = closure;
}
@@ -165,7 +165,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
}
if (pollset->shutting_down && pollset->on_shutdown != NULL) {
- grpc_exec_ctx_enqueue(exec_ctx, pollset->on_shutdown, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pollset->on_shutdown, true, NULL);
pollset->on_shutdown = NULL;
}
goto done;
diff --git a/src/core/iomgr/resolve_address_posix.c b/src/core/iomgr/resolve_address_posix.c
index 555c74ce7e..c51745b918 100644
--- a/src/core/iomgr/resolve_address_posix.c
+++ b/src/core/iomgr/resolve_address_posix.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -152,7 +152,7 @@ done:
/* Callback to be passed to grpc_executor to asynch-ify
* grpc_blocking_resolve_address */
-static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, int success) {
+static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, bool success) {
request *r = rp;
grpc_resolved_addresses *resolved =
grpc_blocking_resolve_address(r->name, r->default_port);
diff --git a/src/core/iomgr/resolve_address_windows.c b/src/core/iomgr/resolve_address_windows.c
index 007c855d10..28c8661e73 100644
--- a/src/core/iomgr/resolve_address_windows.c
+++ b/src/core/iomgr/resolve_address_windows.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -135,7 +135,7 @@ done:
/* Callback to be passed to grpc_executor to asynch-ify
* grpc_blocking_resolve_address */
-static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, int success) {
+static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, bool success) {
request *r = rp;
grpc_resolved_addresses *resolved =
grpc_blocking_resolve_address(r->name, r->default_port);
diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c
index d9d24ee9a3..c76c2e3b0f 100644
--- a/src/core/iomgr/tcp_client_posix.c
+++ b/src/core/iomgr/tcp_client_posix.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -91,7 +91,7 @@ error:
return 0;
}
-static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int success) {
+static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, bool success) {
int done;
async_connect *ac = acp;
if (grpc_tcp_trace) {
@@ -111,7 +111,7 @@ static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int success) {
}
}
-static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, int success) {
+static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, bool success) {
async_connect *ac = acp;
int so_error = 0;
socklen_t so_error_size;
@@ -206,7 +206,7 @@ finish:
gpr_free(ac->addr_str);
gpr_free(ac);
}
- grpc_exec_ctx_enqueue(exec_ctx, closure, *ep != NULL);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, *ep != NULL, NULL);
}
void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
@@ -243,7 +243,7 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
addr_len = sizeof(addr4_copy);
}
if (!prepare_socket(addr, fd)) {
- grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
return;
}
@@ -259,14 +259,14 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
if (err >= 0) {
*ep = grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str);
- grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL);
goto done;
}
if (errno != EWOULDBLOCK && errno != EINPROGRESS) {
gpr_log(GPR_ERROR, "connect error to '%s': %s", addr_str, strerror(errno));
grpc_fd_orphan(exec_ctx, fdobj, NULL, NULL, "tcp_client_connect_error");
- grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
goto done;
}
diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c
index e5691b7e12..689c6f7b10 100644
--- a/src/core/iomgr/tcp_client_windows.c
+++ b/src/core/iomgr/tcp_client_windows.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -74,7 +74,7 @@ static void async_connect_unlock_and_cleanup(async_connect *ac) {
}
}
-static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int occured) {
+static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, bool occured) {
async_connect *ac = acp;
gpr_mu_lock(&ac->mu);
/* If the alarm didn't occur, it got cancelled. */
@@ -84,7 +84,7 @@ static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int occured) {
async_connect_unlock_and_cleanup(ac);
}
-static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, int from_iocp) {
+static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, bool from_iocp) {
async_connect *ac = acp;
SOCKET sock = ac->socket->socket;
grpc_endpoint **ep = ac->endpoint;
@@ -215,7 +215,7 @@ failure:
} else if (sock != INVALID_SOCKET) {
closesocket(sock);
}
- grpc_exec_ctx_enqueue(exec_ctx, on_done, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, on_done, false, NULL);
}
#endif /* GPR_WINSOCK_SOCKET */
diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c
index 4fa8ca8c71..048e907441 100644
--- a/src/core/iomgr/tcp_posix.c
+++ b/src/core/iomgr/tcp_posix.c
@@ -100,9 +100,9 @@ typedef struct {
} grpc_tcp;
static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- int success);
+ bool success);
static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- int success);
+ bool success);
static void tcp_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *)ep;
@@ -247,7 +247,7 @@ static void tcp_continue_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
}
static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- int success) {
+ bool success) {
grpc_tcp *tcp = (grpc_tcp *)arg;
GPR_ASSERT(!tcp->finished_edge);
@@ -273,7 +273,7 @@ static void tcp_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
tcp->finished_edge = 0;
grpc_fd_notify_on_read(exec_ctx, tcp->em_fd, &tcp->read_closure);
} else {
- grpc_exec_ctx_enqueue(exec_ctx, &tcp->read_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &tcp->read_closure, true, NULL);
}
}
@@ -360,7 +360,7 @@ static flush_result tcp_flush(grpc_tcp *tcp) {
}
static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- int success) {
+ bool success) {
grpc_tcp *tcp = (grpc_tcp *)arg;
flush_result status;
grpc_closure *cb;
@@ -407,7 +407,7 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
if (buf->length == 0) {
GPR_TIMER_END("tcp_write", 0);
- grpc_exec_ctx_enqueue(exec_ctx, cb, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, true, NULL);
return;
}
tcp->outgoing_buffer = buf;
@@ -420,7 +420,7 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
tcp->write_cb = cb;
grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_closure);
} else {
- grpc_exec_ctx_enqueue(exec_ctx, cb, status == FLUSH_DONE);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, status == FLUSH_DONE, NULL);
}
GPR_TIMER_END("tcp_write", 0);
diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c
index adf14aeb59..5e07f8261c 100644
--- a/src/core/iomgr/tcp_server_posix.c
+++ b/src/core/iomgr/tcp_server_posix.c
@@ -160,7 +160,7 @@ grpc_tcp_server *grpc_tcp_server_create(grpc_closure *shutdown_complete) {
static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
if (s->shutdown_complete != NULL) {
- grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, true, NULL);
}
gpr_mu_destroy(&s->mu);
@@ -174,7 +174,8 @@ static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
gpr_free(s);
}
-static void destroyed_port(grpc_exec_ctx *exec_ctx, void *server, int success) {
+static void destroyed_port(grpc_exec_ctx *exec_ctx, void *server,
+ bool success) {
grpc_tcp_server *s = server;
gpr_mu_lock(&s->mu);
s->destroyed_ports++;
@@ -317,7 +318,7 @@ error:
}
/* event manager callback when reads are ready */
-static void on_read(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void on_read(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
grpc_tcp_listener *sp = arg;
grpc_tcp_server_acceptor acceptor = {sp->server, sp->port_index,
sp->fd_index};
@@ -602,7 +603,7 @@ void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
/* Complete shutdown_starting work before destroying. */
grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT;
gpr_mu_lock(&s->mu);
- grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting);
+ grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting, NULL);
gpr_mu_unlock(&s->mu);
if (exec_ctx == NULL) {
grpc_exec_ctx_flush(&local_exec_ctx);
diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c
index 00d381f264..ce930b8f41 100644
--- a/src/core/iomgr/tcp_server_windows.c
+++ b/src/core/iomgr/tcp_server_windows.c
@@ -119,7 +119,7 @@ grpc_tcp_server *grpc_tcp_server_create(grpc_closure *shutdown_complete) {
static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
if (s->shutdown_complete != NULL) {
- grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, true, NULL);
}
/* Now that the accepts have been aborted, we can destroy the sockets.
@@ -173,7 +173,7 @@ void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
/* Complete shutdown_starting work before destroying. */
grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT;
gpr_mu_lock(&s->mu);
- grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting);
+ grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting, NULL);
gpr_mu_unlock(&s->mu);
if (exec_ctx == NULL) {
grpc_exec_ctx_flush(&local_exec_ctx);
@@ -311,7 +311,7 @@ failure:
}
/* Event manager callback when reads are ready. */
-static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, int from_iocp) {
+static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, bool from_iocp) {
grpc_tcp_listener *sp = arg;
grpc_tcp_server_acceptor acceptor = {sp->server, sp->port_index, 0};
SOCKET sock = sp->new_socket;
diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c
index d3f080cbf9..038e4158c8 100644
--- a/src/core/iomgr/tcp_windows.c
+++ b/src/core/iomgr/tcp_windows.c
@@ -138,7 +138,7 @@ static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); }
#endif
/* Asynchronous callback from the IOCP, or the background thread. */
-static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, int success) {
+static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, bool success) {
grpc_tcp *tcp = tcpp;
grpc_closure *cb = tcp->read_cb;
grpc_winsocket *socket = tcp->socket;
@@ -184,7 +184,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
WSABUF buffer;
if (tcp->shutting_down) {
- grpc_exec_ctx_enqueue(exec_ctx, cb, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL);
return;
}
@@ -208,7 +208,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
/* Did we get data immediately ? Yay. */
if (info->wsa_error != WSAEWOULDBLOCK) {
info->bytes_transfered = bytes_read;
- grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, true, NULL);
return;
}
@@ -221,7 +221,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
int wsa_error = WSAGetLastError();
if (wsa_error != WSA_IO_PENDING) {
info->wsa_error = wsa_error;
- grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, false, NULL);
return;
}
}
@@ -230,7 +230,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
}
/* Asynchronous callback from the IOCP, or the background thread. */
-static void on_write(grpc_exec_ctx *exec_ctx, void *tcpp, int success) {
+static void on_write(grpc_exec_ctx *exec_ctx, void *tcpp, bool success) {
grpc_tcp *tcp = (grpc_tcp *)tcpp;
grpc_winsocket *handle = tcp->socket;
grpc_winsocket_callback_info *info = &handle->write_info;
@@ -273,7 +273,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
size_t len;
if (tcp->shutting_down) {
- grpc_exec_ctx_enqueue(exec_ctx, cb, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL);
return;
}
@@ -301,9 +301,9 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
connection that has its send queue filled up. But if we don't, then we can
avoid doing an async write operation at all. */
if (info->wsa_error != WSAEWOULDBLOCK) {
- int ok = 0;
+ bool ok = false;
if (status == 0) {
- ok = 1;
+ ok = true;
GPR_ASSERT(bytes_sent == tcp->write_slices->length);
} else {
if (socket->read_info.wsa_error != WSAECONNRESET) {
@@ -313,7 +313,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
}
}
if (allocated) gpr_free(allocated);
- grpc_exec_ctx_enqueue(exec_ctx, cb, ok);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, ok, NULL);
return;
}
@@ -330,7 +330,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
int wsa_error = WSAGetLastError();
if (wsa_error != WSA_IO_PENDING) {
TCP_UNREF(tcp, "write");
- grpc_exec_ctx_enqueue(exec_ctx, cb, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL);
return;
}
}
diff --git a/src/core/iomgr/timer.c b/src/core/iomgr/timer.c
index 24d6204e07..a33d8f63a0 100644
--- a/src/core/iomgr/timer.c
+++ b/src/core/iomgr/timer.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -224,7 +224,7 @@ void grpc_timer_cancel(grpc_exec_ctx *exec_ctx, grpc_timer *timer) {
shard_type *shard = &g_shards[shard_idx(timer)];
gpr_mu_lock(&shard->mu);
if (!timer->triggered) {
- grpc_exec_ctx_enqueue(exec_ctx, &timer->closure, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, &timer->closure, false, NULL);
timer->triggered = 1;
if (timer->heap_index == INVALID_HEAP_INDEX) {
list_remove(timer);
@@ -290,7 +290,7 @@ static size_t pop_timers(grpc_exec_ctx *exec_ctx, shard_type *shard,
grpc_timer *timer;
gpr_mu_lock(&shard->mu);
while ((timer = pop_one(shard, now))) {
- grpc_exec_ctx_enqueue(exec_ctx, &timer->closure, success);
+ grpc_exec_ctx_enqueue(exec_ctx, &timer->closure, success, NULL);
n++;
}
*new_min_deadline = compute_min_deadline(shard);
diff --git a/src/core/iomgr/workqueue.h b/src/core/iomgr/workqueue.h
index 714536233c..36dd133468 100644
--- a/src/core/iomgr/workqueue.h
+++ b/src/core/iomgr/workqueue.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -47,9 +47,7 @@
#include "src/core/iomgr/workqueue_windows.h"
#endif
-/** A workqueue represents a list of work to be executed asynchronously. */
-struct grpc_workqueue;
-typedef struct grpc_workqueue grpc_workqueue;
+/* grpc_workqueue is forward declared in exec_ctx.h */
/** Create a work queue */
grpc_workqueue *grpc_workqueue_create(grpc_exec_ctx *exec_ctx);
diff --git a/src/core/iomgr/workqueue_posix.c b/src/core/iomgr/workqueue_posix.c
index d2a1c34612..da11df67ef 100644
--- a/src/core/iomgr/workqueue_posix.c
+++ b/src/core/iomgr/workqueue_posix.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -45,7 +45,7 @@
#include "src/core/iomgr/fd_posix.h"
-static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, int success);
+static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, bool success);
grpc_workqueue *grpc_workqueue_create(grpc_exec_ctx *exec_ctx) {
char name[32];
@@ -110,7 +110,7 @@ void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) {
gpr_mu_unlock(&workqueue->mu);
}
-static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
grpc_workqueue *workqueue = arg;
if (!success) {