diff options
Diffstat (limited to 'src/core')
44 files changed, 1126 insertions, 619 deletions
diff --git a/src/core/channel/child_channel.c b/src/core/channel/child_channel.c index a2f3c54290..bdd4147117 100644 --- a/src/core/channel/child_channel.c +++ b/src/core/channel/child_channel.c @@ -154,9 +154,18 @@ static void lb_destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_child_channel_top_filter = { - lb_start_transport_op, lb_channel_op, sizeof(lb_call_data), - lb_init_call_elem, lb_destroy_call_elem, sizeof(lb_channel_data), - lb_init_channel_elem, lb_destroy_channel_elem, "child-channel", + lb_start_transport_op, + lb_channel_op, + + sizeof(lb_call_data), + lb_init_call_elem, + lb_destroy_call_elem, + + sizeof(lb_channel_data), + lb_init_channel_elem, + lb_destroy_channel_elem, + + "child-channel", }; /* grpc_child_channel proper */ diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index 42e242ae81..ac34851504 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -39,6 +39,7 @@ #include "src/core/channel/child_channel.h" #include "src/core/channel/connected_channel.h" #include "src/core/iomgr/iomgr.h" +#include "src/core/iomgr/pollset_set.h" #include "src/core/support/string.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> @@ -131,7 +132,11 @@ static void remove_waiting_child(channel_data *chand, call_data *calld) { size_t new_count; size_t i; for (i = 0, new_count = 0; i < chand->waiting_child_count; i++) { - if (chand->waiting_children[i] == calld) continue; + if (chand->waiting_children[i] == calld) { + grpc_transport_setup_del_interested_party( + chand->transport_setup, calld->s.waiting_op.bind_pollset); + continue; + } chand->waiting_children[new_count++] = chand->waiting_children[i]; } GPR_ASSERT(new_count == chand->waiting_child_count - 1 || @@ -228,6 +233,8 @@ static void cc_start_transport_op(grpc_call_element *elem, if (initiate_transport_setup) { grpc_transport_setup_initiate(chand->transport_setup); } + grpc_transport_setup_add_interested_party(chand->transport_setup, + op->bind_pollset); } } break; @@ -416,9 +423,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_client_channel_filter = { - cc_start_transport_op, channel_op, sizeof(call_data), init_call_elem, - destroy_call_elem, sizeof(channel_data), init_channel_elem, - destroy_channel_elem, "client-channel", + cc_start_transport_op, channel_op, sizeof(call_data), + init_call_elem, destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "client-channel", }; grpc_transport_setup_result grpc_client_channel_transport_setup_complete( diff --git a/src/core/channel/client_setup.c b/src/core/channel/client_setup.c index 6d892d6c92..6782ef515a 100644 --- a/src/core/channel/client_setup.c +++ b/src/core/channel/client_setup.c @@ -61,6 +61,7 @@ struct grpc_client_setup { struct grpc_client_setup_request { /* pointer back to the setup object */ grpc_client_setup *setup; + grpc_pollset_set interested_parties; gpr_timespec deadline; }; @@ -68,6 +69,11 @@ gpr_timespec grpc_client_setup_request_deadline(grpc_client_setup_request *r) { return r->deadline; } +grpc_pollset_set *grpc_client_setup_get_interested_parties( + grpc_client_setup_request *r) { + return &r->interested_parties; +} + static void destroy_setup(grpc_client_setup *s) { gpr_mu_destroy(&s->mu); gpr_cv_destroy(&s->cv); @@ -76,6 +82,11 @@ static void destroy_setup(grpc_client_setup *s) { gpr_free(s); } +static void destroy_request(grpc_client_setup_request *r) { + grpc_pollset_set_destroy(&r->interested_parties); + gpr_free(r); +} + /* initiate handshaking */ static void setup_initiate(grpc_transport_setup *sp) { grpc_client_setup *s = (grpc_client_setup *)sp; @@ -83,6 +94,7 @@ static void setup_initiate(grpc_transport_setup *sp) { int in_alarm = 0; r->setup = s; + grpc_pollset_set_init(&r->interested_parties); /* TODO(klempner): Actually set a deadline */ r->deadline = gpr_inf_future; @@ -104,8 +116,38 @@ static void setup_initiate(grpc_transport_setup *sp) { if (!in_alarm) { s->initiate(s->user_data, r); } else { - gpr_free(r); + destroy_request(r); + } +} + +static void setup_add_interested_party(grpc_transport_setup *sp, + grpc_pollset *pollset) { + grpc_client_setup *s = (grpc_client_setup *)sp; + + gpr_mu_lock(&s->mu); + if (!s->active_request) { + gpr_mu_unlock(&s->mu); + return; + } + + grpc_pollset_set_add_pollset(&s->active_request->interested_parties, pollset); + + gpr_mu_unlock(&s->mu); +} + +static void setup_del_interested_party(grpc_transport_setup *sp, + grpc_pollset *pollset) { + grpc_client_setup *s = (grpc_client_setup *)sp; + + gpr_mu_lock(&s->mu); + if (!s->active_request) { + gpr_mu_unlock(&s->mu); + return; } + + grpc_pollset_set_del_pollset(&s->active_request->interested_parties, pollset); + + gpr_mu_unlock(&s->mu); } /* cancel handshaking: cancel all requests, and shutdown (the caller promises @@ -137,7 +179,7 @@ static void setup_cancel(grpc_transport_setup *sp) { } } -int grpc_client_setup_cb_begin(grpc_client_setup_request *r) { +int grpc_client_setup_cb_begin(grpc_client_setup_request *r, const char *reason) { gpr_mu_lock(&r->setup->mu); if (r->setup->cancelled) { gpr_mu_unlock(&r->setup->mu); @@ -148,7 +190,7 @@ int grpc_client_setup_cb_begin(grpc_client_setup_request *r) { return 1; } -void grpc_client_setup_cb_end(grpc_client_setup_request *r) { +void grpc_client_setup_cb_end(grpc_client_setup_request *r, const char *reason) { gpr_mu_lock(&r->setup->mu); r->setup->in_cb--; if (r->setup->cancelled) gpr_cv_signal(&r->setup->cv); @@ -156,8 +198,9 @@ void grpc_client_setup_cb_end(grpc_client_setup_request *r) { } /* vtable for transport setup */ -static const grpc_transport_setup_vtable setup_vtable = {setup_initiate, - setup_cancel}; +static const grpc_transport_setup_vtable setup_vtable = { + setup_initiate, setup_add_interested_party, setup_del_interested_party, + setup_cancel}; void grpc_client_setup_create_and_attach( grpc_channel_stack *newly_minted_channel, const grpc_channel_args *args, @@ -184,38 +227,37 @@ void grpc_client_setup_create_and_attach( grpc_client_channel_set_transport_setup(newly_minted_channel, &s->base); } -int grpc_client_setup_request_should_continue(grpc_client_setup_request *r) { +int grpc_client_setup_request_should_continue(grpc_client_setup_request *r, const char *reason) { int result; if (gpr_time_cmp(gpr_now(), r->deadline) > 0) { - return 0; + result = 0; + } else { + gpr_mu_lock(&r->setup->mu); + result = r->setup->active_request == r; + gpr_mu_unlock(&r->setup->mu); } - gpr_mu_lock(&r->setup->mu); - result = r->setup->active_request == r; - gpr_mu_unlock(&r->setup->mu); return result; } static void backoff_alarm_done(void *arg /* grpc_client_setup */, int success) { - grpc_client_setup *s = arg; - grpc_client_setup_request *r = gpr_malloc(sizeof(grpc_client_setup_request)); - r->setup = s; - /* TODO(klempner): Set this to something useful */ - r->deadline = gpr_inf_future; + grpc_client_setup_request *r = arg; + grpc_client_setup *s = r->setup; /* Handle status cancelled? */ gpr_mu_lock(&s->mu); - s->active_request = r; s->in_alarm = 0; - if (!success) { + if (s->active_request != NULL || !success) { if (0 == --s->refs) { gpr_mu_unlock(&s->mu); destroy_setup(s); - gpr_free(r); + destroy_request(r); return; } else { gpr_mu_unlock(&s->mu); + destroy_request(r); return; } } + s->active_request = r; gpr_mu_unlock(&s->mu); s->initiate(s->user_data, r); } @@ -231,16 +273,12 @@ void grpc_client_setup_request_finish(grpc_client_setup_request *r, } else { retry = 0; } + if (!retry && 0 == --s->refs) { gpr_mu_unlock(&s->mu); destroy_setup(s); - gpr_free(r); - return; - } - - gpr_free(r); - - if (retry) { + destroy_request(r); + } else if (retry) { /* TODO(klempner): Replace these values with further consideration. 2x is probably too aggressive of a backoff. */ gpr_timespec max_backoff = gpr_time_from_minutes(2); @@ -248,15 +286,17 @@ void grpc_client_setup_request_finish(grpc_client_setup_request *r, gpr_timespec deadline = gpr_time_add(s->current_backoff_interval, now); GPR_ASSERT(!s->in_alarm); s->in_alarm = 1; - grpc_alarm_init(&s->backoff_alarm, deadline, backoff_alarm_done, s, now); + grpc_alarm_init(&s->backoff_alarm, deadline, backoff_alarm_done, r, now); s->current_backoff_interval = gpr_time_add(s->current_backoff_interval, s->current_backoff_interval); if (gpr_time_cmp(s->current_backoff_interval, max_backoff) > 0) { s->current_backoff_interval = max_backoff; } + gpr_mu_unlock(&s->mu); + } else { + gpr_mu_unlock(&s->mu); + destroy_request(r); } - - gpr_mu_unlock(&s->mu); } const grpc_channel_args *grpc_client_setup_get_channel_args( diff --git a/src/core/channel/client_setup.h b/src/core/channel/client_setup.h index 70137e1365..cbabb510b1 100644 --- a/src/core/channel/client_setup.h +++ b/src/core/channel/client_setup.h @@ -52,7 +52,7 @@ void grpc_client_setup_create_and_attach( /* Check that r is the active request: needs to be performed at each callback. If this races, we'll have two connection attempts running at once and the old one will get cleaned up in due course, which is fine. */ -int grpc_client_setup_request_should_continue(grpc_client_setup_request *r); +int grpc_client_setup_request_should_continue(grpc_client_setup_request *r, const char *reason); void grpc_client_setup_request_finish(grpc_client_setup_request *r, int was_successful); const grpc_channel_args *grpc_client_setup_get_channel_args( @@ -61,13 +61,15 @@ const grpc_channel_args *grpc_client_setup_get_channel_args( /* Call before calling back into the setup listener, and call only if this function returns 1. If it returns 1, also promise to call grpc_client_setup_cb_end */ -int grpc_client_setup_cb_begin(grpc_client_setup_request *r); -void grpc_client_setup_cb_end(grpc_client_setup_request *r); +int grpc_client_setup_cb_begin(grpc_client_setup_request *r, const char *reason); +void grpc_client_setup_cb_end(grpc_client_setup_request *r, const char *reason); /* Get the deadline for a request passed in to initiate. Implementations should make a best effort to honor this deadline. */ gpr_timespec grpc_client_setup_request_deadline(grpc_client_setup_request *r); +grpc_pollset_set *grpc_client_setup_get_interested_parties( + grpc_client_setup_request *r); grpc_mdctx *grpc_client_setup_get_mdctx(grpc_client_setup_request *r); -#endif /* GRPC_INTERNAL_CORE_CHANNEL_CLIENT_SETUP_H */ +#endif /* GRPC_INTERNAL_CORE_CHANNEL_CLIENT_SETUP_H */ diff --git a/src/core/httpcli/httpcli.c b/src/core/httpcli/httpcli.c index 6e4156c385..53ec070900 100644 --- a/src/core/httpcli/httpcli.c +++ b/src/core/httpcli/httpcli.c @@ -59,11 +59,24 @@ typedef struct { int use_ssl; grpc_httpcli_response_cb on_response; void *user_data; + grpc_httpcli_context *context; } internal_request; static grpc_httpcli_get_override g_get_override = NULL; static grpc_httpcli_post_override g_post_override = NULL; +void grpc_httpcli_context_init(grpc_httpcli_context *context) { + grpc_pollset_set_init(&context->pollset_set); +} + +void grpc_httpcli_context_destroy(grpc_httpcli_context *context) { + grpc_pollset_set_destroy(&context->pollset_set); +} + +void grpc_httpcli_context_add_interested_party(grpc_httpcli_context *context, grpc_pollset *pollset) { + grpc_pollset_set_add_pollset(&context->pollset_set, pollset); +} + static void next_address(internal_request *req); static void finish(internal_request *req, int success) { @@ -197,8 +210,9 @@ static void next_address(internal_request *req) { return; } addr = &req->addresses->addrs[req->next_address++]; - grpc_tcp_client_connect(on_connected, req, (struct sockaddr *)&addr->addr, - addr->len, req->deadline); + grpc_tcp_client_connect(on_connected, req, &req->context->pollset_set, + (struct sockaddr *)&addr->addr, addr->len, + req->deadline); } static void on_resolved(void *arg, grpc_resolved_addresses *addresses) { @@ -212,7 +226,8 @@ static void on_resolved(void *arg, grpc_resolved_addresses *addresses) { next_address(req); } -void grpc_httpcli_get(const grpc_httpcli_request *request, +void grpc_httpcli_get(grpc_httpcli_context *context, + const grpc_httpcli_request *request, gpr_timespec deadline, grpc_httpcli_response_cb on_response, void *user_data) { internal_request *req; @@ -228,6 +243,7 @@ void grpc_httpcli_get(const grpc_httpcli_request *request, req->user_data = user_data; req->deadline = deadline; req->use_ssl = request->use_ssl; + req->context = context; if (req->use_ssl) { req->host = gpr_strdup(request->host); } @@ -236,7 +252,8 @@ void grpc_httpcli_get(const grpc_httpcli_request *request, on_resolved, req); } -void grpc_httpcli_post(const grpc_httpcli_request *request, +void grpc_httpcli_post(grpc_httpcli_context *context, + const grpc_httpcli_request *request, const char *body_bytes, size_t body_size, gpr_timespec deadline, grpc_httpcli_response_cb on_response, void *user_data) { @@ -254,6 +271,7 @@ void grpc_httpcli_post(const grpc_httpcli_request *request, req->user_data = user_data; req->deadline = deadline; req->use_ssl = request->use_ssl; + req->context = context; if (req->use_ssl) { req->host = gpr_strdup(request->host); } diff --git a/src/core/httpcli/httpcli.h b/src/core/httpcli/httpcli.h index 255c5ed90f..2ed4a9436b 100644 --- a/src/core/httpcli/httpcli.h +++ b/src/core/httpcli/httpcli.h @@ -38,6 +38,8 @@ #include <grpc/support/time.h> +#include "src/core/iomgr/pollset_set.h" + /* User agent this library reports */ #define GRPC_HTTPCLI_USER_AGENT "grpc-httpcli/0.0" /* Maximum length of a header string of the form 'Key: Value\r\n' */ @@ -49,6 +51,13 @@ typedef struct grpc_httpcli_header { char *value; } grpc_httpcli_header; +/* Tracks in-progress http requests + TODO(ctiller): allow caching and capturing multiple requests for the + same content and combining them */ +typedef struct grpc_httpcli_context { + grpc_pollset_set pollset_set; +} grpc_httpcli_context; + /* A request */ typedef struct grpc_httpcli_request { /* The host name to connect to */ @@ -80,6 +89,10 @@ typedef struct grpc_httpcli_response { typedef void (*grpc_httpcli_response_cb)(void *user_data, const grpc_httpcli_response *response); +void grpc_httpcli_context_init(grpc_httpcli_context *context); +void grpc_httpcli_context_destroy(grpc_httpcli_context *context); +void grpc_httpcli_context_add_interested_party(grpc_httpcli_context *context, grpc_pollset *pollset); + /* Asynchronously perform a HTTP GET. 'request' contains request parameters - these are caller owned and can be destroyed once the call returns @@ -88,14 +101,16 @@ typedef void (*grpc_httpcli_response_cb)(void *user_data, lifetime of the request 'on_response' is a callback to report results to (and 'user_data' is a user supplied pointer to pass to said call) */ -void grpc_httpcli_get(const grpc_httpcli_request *request, +void grpc_httpcli_get(grpc_httpcli_context *context, + const grpc_httpcli_request *request, gpr_timespec deadline, grpc_httpcli_response_cb on_response, void *user_data); /* Asynchronously perform a HTTP POST. When there is no body, pass in NULL as body_bytes. Does not support ?var1=val1&var2=val2 in the path. */ -void grpc_httpcli_post(const grpc_httpcli_request *request, +void grpc_httpcli_post(grpc_httpcli_context *context, + const grpc_httpcli_request *request, const char *body_bytes, size_t body_size, gpr_timespec deadline, grpc_httpcli_response_cb on_response, void *user_data); @@ -115,4 +130,4 @@ typedef int (*grpc_httpcli_post_override)(const grpc_httpcli_request *request, void grpc_httpcli_set_override(grpc_httpcli_get_override get, grpc_httpcli_post_override post); -#endif /* GRPC_INTERNAL_CORE_HTTPCLI_HTTPCLI_H */ +#endif /* GRPC_INTERNAL_CORE_HTTPCLI_HTTPCLI_H */ diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c index b697fcc64a..65dff84a68 100644 --- a/src/core/iomgr/fd_posix.c +++ b/src/core/iomgr/fd_posix.c @@ -109,12 +109,28 @@ static void destroy(grpc_fd *fd) { gpr_free(fd); } +#ifdef GRPC_FD_REF_COUNT_DEBUG +#define REF_BY(fd, n, reason) ref_by(fd, n, reason, __FILE__, __LINE__) +#define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__) +static void ref_by(grpc_fd *fd, int n, const char *reason, const char *file, int line) { + gpr_log(GPR_DEBUG, "FD %d ref %d %d -> %d [%s; %s:%d]", fd->fd, n, fd->refst, fd->refst + n, reason, file, line); +#else +#define REF_BY(fd, n, reason) ref_by(fd, n) +#define UNREF_BY(fd, n, reason) unref_by(fd, n) static void ref_by(grpc_fd *fd, int n) { +#endif GPR_ASSERT(gpr_atm_no_barrier_fetch_add(&fd->refst, n) > 0); } +#ifdef GRPC_FD_REF_COUNT_DEBUG +static void unref_by(grpc_fd *fd, int n, const char *reason, const char *file, int line) { + gpr_atm old; + gpr_log(GPR_DEBUG, "FD %d unref %d %d -> %d [%s; %s:%d]", fd->fd, n, fd->refst, fd->refst - n, reason, file, line); +#else static void unref_by(grpc_fd *fd, int n) { - gpr_atm old = gpr_atm_full_fetch_add(&fd->refst, -n); + gpr_atm old; +#endif + 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); @@ -141,7 +157,6 @@ static void do_nothing(void *ignored, int success) {} grpc_fd *grpc_fd_create(int fd) { grpc_fd *r = alloc_fd(fd); grpc_iomgr_ref(); - grpc_pollset_add_fd(grpc_backup_pollset(), r); return r; } @@ -183,17 +198,31 @@ 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; shutdown(fd->fd, SHUT_RDWR); - ref_by(fd, 1); /* remove active status, but keep referenced */ + REF_BY(fd, 1, "orphan"); /* remove active status, but keep referenced */ gpr_mu_lock(&fd->watcher_mu); wake_all_watchers_locked(fd); gpr_mu_unlock(&fd->watcher_mu); - unref_by(fd, 2); /* drop the reference */ + UNREF_BY(fd, 2, "orphan"); /* drop the reference */ } /* increment refcount by two to avoid changing the orphan bit */ -void grpc_fd_ref(grpc_fd *fd) { ref_by(fd, 2); } +#ifdef GRPC_FD_REF_COUNT_DEBUG +void grpc_fd_ref(grpc_fd *fd, const char *reason, const char *file, int line) { + ref_by(fd, 2, reason, file, line); +} + +void grpc_fd_unref(grpc_fd *fd, const char *reason, const char *file, int line) { + unref_by(fd, 2, reason, file, line); +} +#else +void grpc_fd_ref(grpc_fd *fd) { + ref_by(fd, 2); +} -void grpc_fd_unref(grpc_fd *fd) { unref_by(fd, 2); } +void grpc_fd_unref(grpc_fd *fd) { + unref_by(fd, 2); +} +#endif static void make_callback(grpc_iomgr_cb_func cb, void *arg, int success, int allow_synchronous_callback) { @@ -317,7 +346,7 @@ gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset, gpr_uint32 mask = 0; /* keep track of pollers that have requested our events, in case they change */ - grpc_fd_ref(fd); + GRPC_FD_REF(fd, "poll"); gpr_mu_lock(&fd->watcher_mu); /* if there is nobody polling for read, but we need to, then start doing so */ @@ -372,7 +401,7 @@ void grpc_fd_end_poll(grpc_fd_watcher *watcher, int got_read, int got_write) { } gpr_mu_unlock(&fd->watcher_mu); - grpc_fd_unref(fd); + GRPC_FD_UNREF(fd, "poll"); } void grpc_fd_become_readable(grpc_fd *fd, int allow_synchronous_callback) { diff --git a/src/core/iomgr/fd_posix.h b/src/core/iomgr/fd_posix.h index cfc533b7f5..7873ce6128 100644 --- a/src/core/iomgr/fd_posix.h +++ b/src/core/iomgr/fd_posix.h @@ -161,8 +161,17 @@ void grpc_fd_become_readable(grpc_fd *fd, int allow_synchronous_callback); void grpc_fd_become_writable(grpc_fd *fd, int allow_synchronous_callback); /* Reference counting for fds */ +#ifdef GRPC_FD_REF_COUNT_DEBUG +void grpc_fd_ref(grpc_fd *fd, const char *reason, const char *file, int line); +void grpc_fd_unref(grpc_fd *fd, const char *reason, const char *file, int line); +#define GRPC_FD_REF(fd, reason) grpc_fd_ref(fd, reason, __FILE__, __LINE__) +#define GRPC_FD_UNREF(fd, reason) grpc_fd_unref(fd, reason, __FILE__, __LINE__) +#else void grpc_fd_ref(grpc_fd *fd); void grpc_fd_unref(grpc_fd *fd); +#define GRPC_FD_REF(fd, reason) grpc_fd_ref(fd) +#define GRPC_FD_UNREF(fd, reason) grpc_fd_unref(fd) +#endif void grpc_fd_global_init(void); void grpc_fd_global_shutdown(void); diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index d22542fc91..9bbff4104c 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -107,12 +107,18 @@ void grpc_iomgr_shutdown(void) { gpr_timespec shutdown_deadline = gpr_time_add(gpr_now(), gpr_time_from_seconds(10)); + grpc_alarm_list_shutdown(); gpr_mu_lock(&g_mu); g_shutdown = 1; - while (g_cbs_head || g_refs) { - 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 != NULL || g_refs > 0) { + if (g_cbs_head != NULL && g_refs > 0) { + gpr_log(GPR_DEBUG, "Waiting for %d iomgr objects to be destroyed and executing final callbacks", g_refs); + } else if (g_cbs_head != NULL) { + gpr_log(GPR_DEBUG, "Executing final iomgr callbacks"); + } else { + gpr_log(GPR_DEBUG, "Waiting for %d iomgr objects to be destroyed", g_refs); + } while (g_cbs_head) { cb = g_cbs_head; g_cbs_head = cb->next; @@ -148,7 +154,6 @@ void grpc_iomgr_shutdown(void) { gpr_event_wait(&g_background_callback_executor_done, gpr_inf_future); grpc_iomgr_platform_shutdown(); - grpc_alarm_list_shutdown(); gpr_mu_destroy(&g_mu); gpr_cv_destroy(&g_rcv); } diff --git a/src/core/iomgr/pollset.h b/src/core/iomgr/pollset.h index 067af87c93..334e0ebde1 100644 --- a/src/core/iomgr/pollset.h +++ b/src/core/iomgr/pollset.h @@ -52,7 +52,6 @@ #include "src/core/iomgr/pollset_windows.h" #endif - void grpc_pollset_init(grpc_pollset *pollset); void grpc_pollset_shutdown(grpc_pollset *pollset, void (*shutdown_done)(void *arg), diff --git a/src/core/iomgr/pollset_kick.c b/src/core/iomgr/pollset_kick_posix.c index f0211b8274..21c9bb6542 100644 --- a/src/core/iomgr/pollset_kick.c +++ b/src/core/iomgr/pollset_kick_posix.c @@ -34,7 +34,7 @@ #include <grpc/support/port_platform.h> #ifdef GPR_POSIX_SOCKET -#include "src/core/iomgr/pollset_kick.h" +#include "src/core/iomgr/pollset_kick_posix.h" #include <errno.h> #include <string.h> @@ -96,41 +96,46 @@ static void free_wfd(grpc_kick_fd_info *fd_info) { void grpc_pollset_kick_init(grpc_pollset_kick_state *kick_state) { gpr_mu_init(&kick_state->mu); kick_state->kicked = 0; - kick_state->fd_info = NULL; + kick_state->fd_list.next = kick_state->fd_list.prev = &kick_state->fd_list; } void grpc_pollset_kick_destroy(grpc_pollset_kick_state *kick_state) { gpr_mu_destroy(&kick_state->mu); - GPR_ASSERT(kick_state->fd_info == NULL); + GPR_ASSERT(kick_state->fd_list.next == &kick_state->fd_list); } -int grpc_pollset_kick_pre_poll(grpc_pollset_kick_state *kick_state) { +grpc_kick_fd_info *grpc_pollset_kick_pre_poll(grpc_pollset_kick_state *kick_state) { + grpc_kick_fd_info *fd_info; gpr_mu_lock(&kick_state->mu); if (kick_state->kicked) { kick_state->kicked = 0; gpr_mu_unlock(&kick_state->mu); - return -1; + return NULL; } - kick_state->fd_info = allocate_wfd(); + fd_info = allocate_wfd(); + fd_info->next = &kick_state->fd_list; + fd_info->prev = fd_info->next->prev; + fd_info->next->prev = fd_info->prev->next = fd_info; gpr_mu_unlock(&kick_state->mu); - return GRPC_WAKEUP_FD_GET_READ_FD(&kick_state->fd_info->wakeup_fd); + return fd_info; } -void grpc_pollset_kick_consume(grpc_pollset_kick_state *kick_state) { - grpc_wakeup_fd_consume_wakeup(&kick_state->fd_info->wakeup_fd); +void grpc_pollset_kick_consume(grpc_pollset_kick_state *kick_state, grpc_kick_fd_info *fd_info) { + grpc_wakeup_fd_consume_wakeup(&fd_info->wakeup_fd); } -void grpc_pollset_kick_post_poll(grpc_pollset_kick_state *kick_state) { +void grpc_pollset_kick_post_poll(grpc_pollset_kick_state *kick_state, grpc_kick_fd_info *fd_info) { gpr_mu_lock(&kick_state->mu); - free_wfd(kick_state->fd_info); - kick_state->fd_info = NULL; + fd_info->next->prev = fd_info->prev; + fd_info->prev->next = fd_info->next; + free_wfd(fd_info); gpr_mu_unlock(&kick_state->mu); } void grpc_pollset_kick_kick(grpc_pollset_kick_state *kick_state) { gpr_mu_lock(&kick_state->mu); - if (kick_state->fd_info != NULL) { - grpc_wakeup_fd_wakeup(&kick_state->fd_info->wakeup_fd); + if (kick_state->fd_list.next != &kick_state->fd_list) { + grpc_wakeup_fd_wakeup(&kick_state->fd_list.next->wakeup_fd); } else { kick_state->kicked = 1; } diff --git a/src/core/iomgr/pollset_kick_posix.h b/src/core/iomgr/pollset_kick_posix.h index 427699198c..b35c2cfbe0 100644 --- a/src/core/iomgr/pollset_kick_posix.h +++ b/src/core/iomgr/pollset_kick_posix.h @@ -39,13 +39,45 @@ typedef struct grpc_kick_fd_info { grpc_wakeup_fd_info wakeup_fd; + /* used for polling list and free list */ struct grpc_kick_fd_info *next; + /* only used when polling */ + struct grpc_kick_fd_info *prev; } grpc_kick_fd_info; typedef struct grpc_pollset_kick_state { gpr_mu mu; int kicked; - struct grpc_kick_fd_info *fd_info; + struct grpc_kick_fd_info fd_list; } grpc_pollset_kick_state; +#define GRPC_POLLSET_KICK_GET_FD(kick_fd_info) GRPC_WAKEUP_FD_GET_READ_FD(&(kick_fd_info)->wakeup_fd) + +/* This is an abstraction around the typical pipe mechanism for waking up a + thread sitting in a poll() style call. */ + +void grpc_pollset_kick_global_init(void); +void grpc_pollset_kick_global_destroy(void); + +void grpc_pollset_kick_init(grpc_pollset_kick_state *kick_state); +void grpc_pollset_kick_destroy(grpc_pollset_kick_state *kick_state); + +/* Guarantees a pure posix implementation rather than a specialized one, if + * applicable. Intended for testing. */ +void grpc_pollset_kick_global_init_fallback_fd(void); + +/* Must be called before entering poll(). If return value is -1, this consumed + an existing kick. Otherwise the return value is an FD to add to the poll set. + */ +grpc_kick_fd_info *grpc_pollset_kick_pre_poll(grpc_pollset_kick_state *kick_state); + +/* Consume an existing kick. Must be called after poll returns that the fd was + readable, and before calling kick_post_poll. */ +void grpc_pollset_kick_consume(grpc_pollset_kick_state *kick_state, grpc_kick_fd_info *fd_info); + +/* Must be called after pre_poll, and after consume if applicable */ +void grpc_pollset_kick_post_poll(grpc_pollset_kick_state *kick_state, grpc_kick_fd_info *fd_info); + +void grpc_pollset_kick_kick(grpc_pollset_kick_state *kick_state); + #endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_POSIX_H */ diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c index 40b7935a57..05b522bbf2 100644 --- a/src/core/iomgr/pollset_multipoller_with_epoll.c +++ b/src/core/iomgr/pollset_multipoller_with_epoll.c @@ -140,13 +140,11 @@ static int multipoll_with_epoll_pollset_maybe_work( gpr_mu_lock(&pollset->mu); pollset->counter -= 1; - /* TODO(klempner): This should signal once per event rather than broadcast, - * although it probably doesn't matter because threads will generally be - * blocked in epoll_wait rather than being blocked on the cv. */ - gpr_cv_broadcast(&pollset->cv); return 1; } +static void multipoll_with_epoll_pollset_finish_shutdown(grpc_pollset *pollset) {} + static void multipoll_with_epoll_pollset_destroy(grpc_pollset *pollset) { pollset_hdr *h = pollset->data.ptr; grpc_wakeup_fd_destroy(&h->wakeup_fd); @@ -162,7 +160,7 @@ static void epoll_kick(grpc_pollset *pollset) { static const grpc_pollset_vtable multipoll_with_epoll_pollset = { multipoll_with_epoll_pollset_add_fd, multipoll_with_epoll_pollset_del_fd, multipoll_with_epoll_pollset_maybe_work, epoll_kick, - multipoll_with_epoll_pollset_destroy}; + multipoll_with_epoll_pollset_finish_shutdown, multipoll_with_epoll_pollset_destroy}; static void epoll_become_multipoller(grpc_pollset *pollset, grpc_fd **fds, size_t nfds) { diff --git a/src/core/iomgr/pollset_multipoller_with_poll_posix.c b/src/core/iomgr/pollset_multipoller_with_poll_posix.c index d781c9b4bb..afce5f65db 100644 --- a/src/core/iomgr/pollset_multipoller_with_poll_posix.c +++ b/src/core/iomgr/pollset_multipoller_with_poll_posix.c @@ -78,7 +78,7 @@ static void multipoll_with_poll_pollset_add_fd(grpc_pollset *pollset, h->fds = gpr_realloc(h->fds, sizeof(grpc_fd *) * h->fd_capacity); } h->fds[h->fd_count++] = fd; - grpc_fd_ref(fd); + GRPC_FD_REF(fd, "multipoller"); } static void multipoll_with_poll_pollset_del_fd(grpc_pollset *pollset, @@ -90,7 +90,7 @@ static void multipoll_with_poll_pollset_del_fd(grpc_pollset *pollset, h->dels = gpr_realloc(h->dels, sizeof(grpc_fd *) * h->del_capacity); } h->dels[h->del_count++] = fd; - grpc_fd_ref(fd); + GRPC_FD_REF(fd, "multipoller_del"); } static void end_polling(grpc_pollset *pollset) { @@ -110,10 +110,8 @@ static int multipoll_with_poll_pollset_maybe_work( int r; size_t i, np, nf, nd; pollset_hdr *h; + grpc_kick_fd_info *kfd; - if (pollset->counter) { - return 0; - } h = pollset->data.ptr; if (gpr_time_cmp(deadline, gpr_inf_future) == 0) { timeout = -1; @@ -132,11 +130,12 @@ static int multipoll_with_poll_pollset_maybe_work( } nf = 0; np = 1; - h->pfds[0].fd = grpc_pollset_kick_pre_poll(&pollset->kick_state); - if (h->pfds[0].fd < 0) { + kfd = grpc_pollset_kick_pre_poll(&pollset->kick_state); + if (kfd == NULL) { /* Already kicked */ return 1; } + h->pfds[0].fd = GRPC_POLLSET_KICK_GET_FD(kfd); h->pfds[0].events = POLLIN; h->pfds[0].revents = POLLOUT; for (i = 0; i < h->fd_count; i++) { @@ -145,7 +144,7 @@ static int multipoll_with_poll_pollset_maybe_work( if (h->fds[i] == h->dels[nd]) remove = 1; } if (remove) { - grpc_fd_unref(h->fds[i]); + GRPC_FD_UNREF(h->fds[i], "multipoller"); } else { h->fds[nf++] = h->fds[i]; h->watchers[np].fd = h->fds[i]; @@ -157,14 +156,14 @@ static int multipoll_with_poll_pollset_maybe_work( h->pfd_count = np; h->fd_count = nf; for (nd = 0; nd < h->del_count; nd++) { - grpc_fd_unref(h->dels[nd]); + GRPC_FD_UNREF(h->dels[nd], "multipoller_del"); } h->del_count = 0; if (h->pfd_count == 0) { end_polling(pollset); return 0; } - pollset->counter = 1; + pollset->counter++; gpr_mu_unlock(&pollset->mu); for (i = 1; i < np; i++) { @@ -184,7 +183,7 @@ static int multipoll_with_poll_pollset_maybe_work( /* do nothing */ } else { if (h->pfds[0].revents & POLLIN) { - grpc_pollset_kick_consume(&pollset->kick_state); + grpc_pollset_kick_consume(&pollset->kick_state, kfd); } for (i = 1; i < np; i++) { if (h->pfds[i].revents & (POLLIN | POLLHUP | POLLERR)) { @@ -195,11 +194,11 @@ static int multipoll_with_poll_pollset_maybe_work( } } } - grpc_pollset_kick_post_poll(&pollset->kick_state); + grpc_pollset_kick_post_poll(&pollset->kick_state, kfd); gpr_mu_lock(&pollset->mu); - pollset->counter = 0; - gpr_cv_broadcast(&pollset->cv); + pollset->counter--; + return 1; } @@ -207,16 +206,23 @@ static void multipoll_with_poll_pollset_kick(grpc_pollset *p) { grpc_pollset_force_kick(p); } -static void multipoll_with_poll_pollset_destroy(grpc_pollset *pollset) { +static void multipoll_with_poll_pollset_finish_shutdown(grpc_pollset *pollset) { size_t i; pollset_hdr *h = pollset->data.ptr; GPR_ASSERT(pollset->counter == 0); for (i = 0; i < h->fd_count; i++) { - grpc_fd_unref(h->fds[i]); + GRPC_FD_UNREF(h->fds[i], "multipoller"); } for (i = 0; i < h->del_count; i++) { - grpc_fd_unref(h->dels[i]); + GRPC_FD_UNREF(h->dels[i], "multipoller_del"); } + h->fd_count = 0; + h->del_count = 0; +} + +static void multipoll_with_poll_pollset_destroy(grpc_pollset *pollset) { + pollset_hdr *h = pollset->data.ptr; + multipoll_with_poll_pollset_finish_shutdown(pollset); gpr_free(h->pfds); gpr_free(h->watchers); gpr_free(h->fds); @@ -227,7 +233,7 @@ static void multipoll_with_poll_pollset_destroy(grpc_pollset *pollset) { static const grpc_pollset_vtable multipoll_with_poll_pollset = { multipoll_with_poll_pollset_add_fd, multipoll_with_poll_pollset_del_fd, multipoll_with_poll_pollset_maybe_work, multipoll_with_poll_pollset_kick, - multipoll_with_poll_pollset_destroy}; + multipoll_with_poll_pollset_finish_shutdown, multipoll_with_poll_pollset_destroy}; void grpc_poll_become_multipoller(grpc_pollset *pollset, grpc_fd **fds, size_t nfds) { @@ -247,7 +253,7 @@ void grpc_poll_become_multipoller(grpc_pollset *pollset, grpc_fd **fds, h->dels = NULL; for (i = 0; i < nfds; i++) { h->fds[i] = fds[i]; - grpc_fd_ref(fds[i]); + GRPC_FD_REF(fds[i], "multipoller"); } } diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c index 826c792990..96f8acfd7e 100644 --- a/src/core/iomgr/pollset_posix.c +++ b/src/core/iomgr/pollset_posix.c @@ -54,31 +54,8 @@ #include <grpc/support/tls.h> #include <grpc/support/useful.h> -static grpc_pollset g_backup_pollset; -static int g_shutdown_backup_poller; -static gpr_event g_backup_poller_done; -static gpr_event g_backup_pollset_shutdown_done; - GPR_TLS_DECL(g_current_thread_poller); -static void backup_poller(void *p) { - gpr_timespec delta = gpr_time_from_millis(100); - gpr_timespec last_poll = gpr_now(); - - gpr_mu_lock(&g_backup_pollset.mu); - while (g_shutdown_backup_poller == 0) { - gpr_timespec next_poll = gpr_time_add(last_poll, delta); - grpc_pollset_work(&g_backup_pollset, gpr_time_add(gpr_now(), gpr_time_from_seconds(1))); - gpr_mu_unlock(&g_backup_pollset.mu); - gpr_sleep_until(next_poll); - gpr_mu_lock(&g_backup_pollset.mu); - last_poll = next_poll; - } - gpr_mu_unlock(&g_backup_pollset.mu); - - gpr_event_set(&g_backup_poller_done, (void *)1); -} - void grpc_pollset_kick(grpc_pollset *p) { if (gpr_tls_get(&g_current_thread_poller) != (gpr_intptr)p && p->counter) { p->vtable->kick(p); @@ -99,44 +76,14 @@ static void kick_using_pollset_kick(grpc_pollset *p) { /* global state management */ -grpc_pollset *grpc_backup_pollset(void) { return &g_backup_pollset; } - void grpc_pollset_global_init(void) { - gpr_thd_id id; - gpr_tls_init(&g_current_thread_poller); /* Initialize kick fd state */ grpc_pollset_kick_global_init(); - - /* initialize the backup pollset */ - grpc_pollset_init(&g_backup_pollset); - - /* start the backup poller thread */ - g_shutdown_backup_poller = 0; - gpr_event_init(&g_backup_poller_done); - gpr_event_init(&g_backup_pollset_shutdown_done); - gpr_thd_new(&id, backup_poller, NULL, NULL); -} - -static void on_backup_pollset_shutdown_done(void *arg) { - gpr_event_set(&g_backup_pollset_shutdown_done, (void *)1); } void grpc_pollset_global_shutdown(void) { - /* terminate the backup poller thread */ - gpr_mu_lock(&g_backup_pollset.mu); - g_shutdown_backup_poller = 1; - gpr_mu_unlock(&g_backup_pollset.mu); - gpr_event_wait(&g_backup_poller_done, gpr_inf_future); - - grpc_pollset_shutdown(&g_backup_pollset, on_backup_pollset_shutdown_done, - NULL); - gpr_event_wait(&g_backup_pollset_shutdown_done, gpr_inf_future); - - /* destroy the backup pollset */ - grpc_pollset_destroy(&g_backup_pollset); - /* destroy the kick pipes */ grpc_pollset_kick_global_destroy(); @@ -145,37 +92,36 @@ void grpc_pollset_global_shutdown(void) { /* main interface */ -static void become_empty_pollset(grpc_pollset *pollset); -static void become_unary_pollset(grpc_pollset *pollset, grpc_fd *fd); +static void become_basic_pollset(grpc_pollset *pollset, grpc_fd *fd_or_null); void grpc_pollset_init(grpc_pollset *pollset) { gpr_mu_init(&pollset->mu); - gpr_cv_init(&pollset->cv); grpc_pollset_kick_init(&pollset->kick_state); pollset->in_flight_cbs = 0; pollset->shutting_down = 0; - become_empty_pollset(pollset); + become_basic_pollset(pollset, NULL); } void grpc_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { gpr_mu_lock(&pollset->mu); pollset->vtable->add_fd(pollset, fd); - gpr_cv_broadcast(&pollset->cv); gpr_mu_unlock(&pollset->mu); } void grpc_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) { gpr_mu_lock(&pollset->mu); pollset->vtable->del_fd(pollset, fd); - gpr_cv_broadcast(&pollset->cv); gpr_mu_unlock(&pollset->mu); } +static void finish_shutdown(grpc_pollset *pollset) { + pollset->vtable->finish_shutdown(pollset); + pollset->shutdown_done_cb(pollset->shutdown_done_arg); +} + int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) { /* pollset->mu already held */ gpr_timespec now = gpr_now(); - /* FIXME(ctiller): see below */ - gpr_timespec maximum_deadline = gpr_time_add(now, gpr_time_from_seconds(1)); int r; if (gpr_time_cmp(now, deadline) > 0) { return 0; @@ -186,14 +132,25 @@ int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) { if (grpc_alarm_check(&pollset->mu, now, &deadline)) { return 1; } - /* FIXME(ctiller): we should not clamp deadline, however we have some - stuck at shutdown bugs that this resolves */ - if (gpr_time_cmp(deadline, maximum_deadline) > 0) { - deadline = maximum_deadline; + if (pollset->shutting_down) { + return 1; } gpr_tls_set(&g_current_thread_poller, (gpr_intptr)pollset); r = pollset->vtable->maybe_work(pollset, deadline, now, 1); gpr_tls_set(&g_current_thread_poller, 0); + if (pollset->shutting_down) { + if (pollset->counter > 0) { + grpc_pollset_kick(pollset); + } else if (pollset->in_flight_cbs == 0) { + gpr_mu_unlock(&pollset->mu); + finish_shutdown(pollset); + /* Continuing to access pollset here is safe -- it is the caller's + * responsibility to not destroy when it has outstanding calls to + * grpc_pollset_work. + * TODO(dklempner): Can we refactor the shutdown logic to avoid this? */ + gpr_mu_lock(&pollset->mu); + } + } return r; } @@ -201,14 +158,21 @@ void grpc_pollset_shutdown(grpc_pollset *pollset, void (*shutdown_done)(void *arg), void *shutdown_done_arg) { int in_flight_cbs; + int counter; gpr_mu_lock(&pollset->mu); + GPR_ASSERT(!pollset->shutting_down); pollset->shutting_down = 1; in_flight_cbs = pollset->in_flight_cbs; + counter = pollset->counter; pollset->shutdown_done_cb = shutdown_done; pollset->shutdown_done_arg = shutdown_done_arg; + if (counter > 0) { + grpc_pollset_kick(pollset); + } gpr_mu_unlock(&pollset->mu); - if (in_flight_cbs == 0) { - shutdown_done(shutdown_done_arg); + + if (in_flight_cbs == 0 && counter == 0) { + finish_shutdown(pollset); } } @@ -218,48 +182,20 @@ void grpc_pollset_destroy(grpc_pollset *pollset) { pollset->vtable->destroy(pollset); grpc_pollset_kick_destroy(&pollset->kick_state); gpr_mu_destroy(&pollset->mu); - gpr_cv_destroy(&pollset->cv); -} - -/* - * empty_pollset - a vtable that provides polling for NO file descriptors - */ - -static void empty_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { - become_unary_pollset(pollset, fd); -} - -static void empty_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) {} - -static int empty_pollset_maybe_work(grpc_pollset *pollset, - gpr_timespec deadline, gpr_timespec now, - int allow_synchronous_callback) { - return 0; -} - -static void empty_pollset_destroy(grpc_pollset *pollset) {} - -static const grpc_pollset_vtable empty_pollset = { - empty_pollset_add_fd, empty_pollset_del_fd, empty_pollset_maybe_work, - kick_using_pollset_kick, empty_pollset_destroy}; - -static void become_empty_pollset(grpc_pollset *pollset) { - pollset->vtable = &empty_pollset; } /* - * unary_poll_pollset - a vtable that provides polling for one file descriptor - * via poll() + * basic_pollset - a vtable that provides polling for zero or one file + * descriptor via poll() */ - typedef struct grpc_unary_promote_args { const grpc_pollset_vtable *original_vtable; grpc_pollset *pollset; grpc_fd *fd; } grpc_unary_promote_args; -static void unary_poll_do_promote(void *args, int success) { +static void basic_do_promote(void *args, int success) { grpc_unary_promote_args *up_args = args; const grpc_pollset_vtable *original_vtable = up_args->original_vtable; grpc_pollset *pollset = up_args->pollset; @@ -277,9 +213,9 @@ static void unary_poll_do_promote(void *args, int success) { gpr_mu_lock(&pollset->mu); /* First we need to ensure that nobody is polling concurrently */ - while (pollset->counter != 0) { + if (pollset->counter != 0) { grpc_pollset_kick(pollset); - grpc_iomgr_add_callback(unary_poll_do_promote, up_args); + grpc_iomgr_add_callback(basic_do_promote, up_args); gpr_mu_unlock(&pollset->mu); return; } @@ -293,7 +229,7 @@ static void unary_poll_do_promote(void *args, int success) { pollset->in_flight_cbs--; if (pollset->shutting_down) { /* We don't care about this pollset anymore. */ - if (pollset->in_flight_cbs == 0) { + if (pollset->in_flight_cbs == 0 && pollset->counter == 0) { do_shutdown_cb = 1; } } else if (grpc_fd_is_orphaned(fd)) { @@ -305,33 +241,33 @@ static void unary_poll_do_promote(void *args, int success) { fds[0] = pollset->data.ptr; fds[1] = fd; - if (!grpc_fd_is_orphaned(fds[0])) { + if (fds[0] && !grpc_fd_is_orphaned(fds[0])) { grpc_platform_become_multipoller(pollset, fds, GPR_ARRAY_SIZE(fds)); - grpc_fd_unref(fds[0]); + GRPC_FD_UNREF(fds[0], "basicpoll"); } else { /* old fd is orphaned and we haven't cleaned it up until now, so remain a * unary poller */ /* Note that it is possible that fds[1] is also orphaned at this point. * That's okay, we'll correct it at the next add or poll. */ - grpc_fd_unref(fds[0]); + if (fds[0]) GRPC_FD_UNREF(fds[0], "basicpoll"); pollset->data.ptr = fd; - grpc_fd_ref(fd); + GRPC_FD_REF(fd, "basicpoll"); } } - gpr_cv_broadcast(&pollset->cv); gpr_mu_unlock(&pollset->mu); if (do_shutdown_cb) { pollset->shutdown_done_cb(pollset->shutdown_done_arg); } - /* Matching ref in unary_poll_pollset_add_fd */ - grpc_fd_unref(fd); + /* Matching ref in basic_pollset_add_fd */ + GRPC_FD_UNREF(fd, "basicpoll_add"); } -static void unary_poll_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { +static void basic_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { grpc_unary_promote_args *up_args; + GPR_ASSERT(fd); if (fd == pollset->data.ptr) return; if (!pollset->counter) { @@ -342,61 +278,62 @@ static void unary_poll_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { fds[0] = pollset->data.ptr; fds[1] = fd; - if (!grpc_fd_is_orphaned(fds[0])) { + if (fds[0] == NULL) { + pollset->data.ptr = fd; + GRPC_FD_REF(fd, "basicpoll"); + } else if (!grpc_fd_is_orphaned(fds[0])) { grpc_platform_become_multipoller(pollset, fds, GPR_ARRAY_SIZE(fds)); - grpc_fd_unref(fds[0]); + GRPC_FD_UNREF(fds[0], "basicpoll"); } else { /* old fd is orphaned and we haven't cleaned it up until now, so remain a * unary poller */ - grpc_fd_unref(fds[0]); + GRPC_FD_UNREF(fds[0], "basicpoll"); pollset->data.ptr = fd; - grpc_fd_ref(fd); + GRPC_FD_REF(fd, "basicpoll"); } return; } /* Now we need to promote. This needs to happen when we're not polling. Since * this may be called from poll, the wait needs to happen asynchronously. */ - grpc_fd_ref(fd); + GRPC_FD_REF(fd, "basicpoll_add"); pollset->in_flight_cbs++; up_args = gpr_malloc(sizeof(*up_args)); up_args->pollset = pollset; up_args->fd = fd; up_args->original_vtable = pollset->vtable; - grpc_iomgr_add_callback(unary_poll_do_promote, up_args); + grpc_iomgr_add_callback(basic_do_promote, up_args); grpc_pollset_kick(pollset); } -static void unary_poll_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) { +static void basic_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) { + GPR_ASSERT(fd); if (fd == pollset->data.ptr) { - grpc_fd_unref(pollset->data.ptr); - become_empty_pollset(pollset); + GRPC_FD_UNREF(pollset->data.ptr, "basicpoll"); + pollset->data.ptr = NULL; } } -static int unary_poll_pollset_maybe_work(grpc_pollset *pollset, - gpr_timespec deadline, - gpr_timespec now, - int allow_synchronous_callback) { +static int basic_pollset_maybe_work(grpc_pollset *pollset, + gpr_timespec deadline, gpr_timespec now, + int allow_synchronous_callback) { struct pollfd pfd[2]; grpc_fd *fd; grpc_fd_watcher fd_watcher; + grpc_kick_fd_info *kfd; int timeout; int r; + int nfds; - if (pollset->counter) { - return 0; - } if (pollset->in_flight_cbs) { /* Give do_promote priority so we don't starve it out */ - return 0; + return 1; } fd = pollset->data.ptr; - if (grpc_fd_is_orphaned(fd)) { - grpc_fd_unref(fd); - become_empty_pollset(pollset); - return 0; + if (fd && grpc_fd_is_orphaned(fd)) { + GRPC_FD_UNREF(fd, "basicpoll"); + fd = pollset->data.ptr = NULL; } if (gpr_time_cmp(deadline, gpr_inf_future) == 0) { timeout = -1; @@ -406,26 +343,38 @@ static int unary_poll_pollset_maybe_work(grpc_pollset *pollset, return 1; } } - pfd[0].fd = grpc_pollset_kick_pre_poll(&pollset->kick_state); - if (pfd[0].fd < 0) { + kfd = grpc_pollset_kick_pre_poll(&pollset->kick_state); + if (kfd == NULL) { /* Already kicked */ return 1; } + pfd[0].fd = GRPC_POLLSET_KICK_GET_FD(kfd); pfd[0].events = POLLIN; pfd[0].revents = 0; - pfd[1].fd = fd->fd; - pfd[1].revents = 0; - pollset->counter = 1; - gpr_mu_unlock(&pollset->mu); - - pfd[1].events = grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT, &fd_watcher); + nfds = 1; + pollset->counter++; + if (fd) { + pfd[1].fd = fd->fd; + pfd[1].revents = 0; + gpr_mu_unlock(&pollset->mu); + pfd[1].events = + grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT, &fd_watcher); + if (pfd[1].events != 0) { + nfds++; + } + } else { + gpr_mu_unlock(&pollset->mu); + } /* poll fd count (argument 2) is shortened by one if we have no events to poll on - such that it only includes the kicker */ - r = poll(pfd, GPR_ARRAY_SIZE(pfd) - (pfd[1].events == 0), timeout); + r = poll(pfd, nfds, timeout); GRPC_TIMER_MARK(GRPC_PTAG_POLL_FINISHED, r); - grpc_fd_end_poll(&fd_watcher, pfd[1].revents & POLLIN, pfd[1].revents & POLLOUT); + if (fd) { + grpc_fd_end_poll(&fd_watcher, pfd[1].revents & POLLIN, + pfd[1].revents & POLLOUT); + } if (r < 0) { if (errno != EINTR) { @@ -435,7 +384,7 @@ static int unary_poll_pollset_maybe_work(grpc_pollset *pollset, /* do nothing */ } else { if (pfd[0].revents & POLLIN) { - grpc_pollset_kick_consume(&pollset->kick_state); + grpc_pollset_kick_consume(&pollset->kick_state, kfd); } if (pfd[1].revents & (POLLIN | POLLHUP | POLLERR)) { grpc_fd_become_readable(fd, allow_synchronous_callback); @@ -445,29 +394,32 @@ static int unary_poll_pollset_maybe_work(grpc_pollset *pollset, } } - grpc_pollset_kick_post_poll(&pollset->kick_state); + grpc_pollset_kick_post_poll(&pollset->kick_state, kfd); gpr_mu_lock(&pollset->mu); - pollset->counter = 0; - gpr_cv_broadcast(&pollset->cv); + pollset->counter--; return 1; } -static void unary_poll_pollset_destroy(grpc_pollset *pollset) { +static void basic_pollset_destroy(grpc_pollset *pollset) { GPR_ASSERT(pollset->counter == 0); - grpc_fd_unref(pollset->data.ptr); + if (pollset->data.ptr != NULL) { + GRPC_FD_UNREF(pollset->data.ptr, "basicpoll"); + pollset->data.ptr = NULL; + } } -static const grpc_pollset_vtable unary_poll_pollset = { - unary_poll_pollset_add_fd, unary_poll_pollset_del_fd, - unary_poll_pollset_maybe_work, kick_using_pollset_kick, - unary_poll_pollset_destroy}; +static const grpc_pollset_vtable basic_pollset = { + basic_pollset_add_fd, basic_pollset_del_fd, basic_pollset_maybe_work, + kick_using_pollset_kick, basic_pollset_destroy, basic_pollset_destroy}; -static void become_unary_pollset(grpc_pollset *pollset, grpc_fd *fd) { - pollset->vtable = &unary_poll_pollset; +static void become_basic_pollset(grpc_pollset *pollset, grpc_fd *fd_or_null) { + pollset->vtable = &basic_pollset; pollset->counter = 0; - pollset->data.ptr = fd; - grpc_fd_ref(fd); + pollset->data.ptr = fd_or_null; + if (fd_or_null) { + GRPC_FD_REF(fd_or_null, "basicpoll"); + } } #endif /* GPR_POSIX_POLLSET */ diff --git a/src/core/iomgr/pollset_posix.h b/src/core/iomgr/pollset_posix.h index 088ec910c2..2b897caa4b 100644 --- a/src/core/iomgr/pollset_posix.h +++ b/src/core/iomgr/pollset_posix.h @@ -36,7 +36,7 @@ #include <grpc/support/sync.h> -#include "src/core/iomgr/pollset_kick.h" +#include "src/core/iomgr/pollset_kick_posix.h" typedef struct grpc_pollset_vtable grpc_pollset_vtable; @@ -52,7 +52,6 @@ typedef struct grpc_pollset { few fds, and an epoll() based implementation for many fds */ const grpc_pollset_vtable *vtable; gpr_mu mu; - gpr_cv cv; grpc_pollset_kick_state kick_state; int counter; int in_flight_cbs; @@ -71,11 +70,11 @@ struct grpc_pollset_vtable { int (*maybe_work)(grpc_pollset *pollset, gpr_timespec deadline, gpr_timespec now, int allow_synchronous_callback); void (*kick)(grpc_pollset *pollset); + void (*finish_shutdown)(grpc_pollset *pollset); void (*destroy)(grpc_pollset *pollset); }; #define GRPC_POLLSET_MU(pollset) (&(pollset)->mu) -#define GRPC_POLLSET_CV(pollset) (&(pollset)->cv) /* Add an fd to a pollset */ void grpc_pollset_add_fd(grpc_pollset *pollset, struct grpc_fd *fd); @@ -94,12 +93,6 @@ int grpc_kick_read_fd(grpc_pollset *p); /* Call after polling has been kicked to leave the kicked state */ void grpc_kick_drain(grpc_pollset *p); -/* All fds get added to a backup pollset to ensure that progress is made - regardless of applications listening to events. Relying on this is slow - however (the backup pollset only listens every 100ms or so) - so it's not - to be relied on. */ -grpc_pollset *grpc_backup_pollset(void); - /* turn a pollset into a multipoller: platform specific */ typedef void (*grpc_platform_become_multipoller_type)(grpc_pollset *pollset, struct grpc_fd **fds, diff --git a/src/core/iomgr/pollset_kick.h b/src/core/iomgr/pollset_set.h index cc9357de1f..335ffb21b9 100644 --- a/src/core/iomgr/pollset_kick.h +++ b/src/core/iomgr/pollset_set.h @@ -31,44 +31,29 @@ * */ -#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_H -#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_H +#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_H +#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_H -#include <grpc/support/port_platform.h> +#include "src/core/iomgr/pollset.h" + +/* A grpc_pollset_set is a set of pollsets that are interested in an + action. Adding a pollset to a pollset_set automatically adds any + fd's (etc) that have been registered with the set_set with that pollset. + Registering fd's automatically iterates all current pollsets. */ #ifdef GPR_POSIX_SOCKET -#include "src/core/iomgr/pollset_kick_posix.h" +#include "src/core/iomgr/pollset_set_posix.h" #endif #ifdef GPR_WIN32 -#include "src/core/iomgr/pollset_kick_windows.h" +#include "src/core/iomgr/pollset_set_windows.h" #endif -/* This is an abstraction around the typical pipe mechanism for waking up a - thread sitting in a poll() style call. */ - -void grpc_pollset_kick_global_init(void); -void grpc_pollset_kick_global_destroy(void); - -void grpc_pollset_kick_init(grpc_pollset_kick_state *kick_state); -void grpc_pollset_kick_destroy(grpc_pollset_kick_state *kick_state); - -/* Guarantees a pure posix implementation rather than a specialized one, if - * applicable. Intended for testing. */ -void grpc_pollset_kick_global_init_fallback_fd(void); - -/* Must be called before entering poll(). If return value is -1, this consumed - an existing kick. Otherwise the return value is an FD to add to the poll set. - */ -int grpc_pollset_kick_pre_poll(grpc_pollset_kick_state *kick_state); - -/* Consume an existing kick. Must be called after poll returns that the fd was - readable, and before calling kick_post_poll. */ -void grpc_pollset_kick_consume(grpc_pollset_kick_state *kick_state); - -/* Must be called after pre_poll, and after consume if applicable */ -void grpc_pollset_kick_post_poll(grpc_pollset_kick_state *kick_state); - -void grpc_pollset_kick_kick(grpc_pollset_kick_state *kick_state); +void grpc_pollset_set_init(grpc_pollset_set *pollset_set); +void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set); +void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, + grpc_pollset *pollset); +void grpc_pollset_set_del_pollset(grpc_pollset_set *pollset_set, + grpc_pollset *pollset); -#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_H */ diff --git a/src/core/iomgr/pollset_set_posix.c b/src/core/iomgr/pollset_set_posix.c new file mode 100644 index 0000000000..c00d6e6e57 --- /dev/null +++ b/src/core/iomgr/pollset_set_posix.c @@ -0,0 +1,119 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/support/port_platform.h> + +#ifdef GPR_POSIX_SOCKET + +#include <stdlib.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/useful.h> + +#include "src/core/iomgr/pollset_set.h" + +void grpc_pollset_set_init(grpc_pollset_set *pollset_set) { + memset(pollset_set, 0, sizeof(*pollset_set)); + gpr_mu_init(&pollset_set->mu); +} + +void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set) { + gpr_mu_destroy(&pollset_set->mu); + gpr_free(pollset_set->pollsets); + gpr_free(pollset_set->fds); +} + +void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, + grpc_pollset *pollset) { + size_t i; + gpr_mu_lock(&pollset_set->mu); + if (pollset_set->pollset_count == pollset_set->pollset_capacity) { + pollset_set->pollset_capacity = + GPR_MAX(8, 2 * pollset_set->pollset_capacity); + pollset_set->pollsets = + gpr_realloc(pollset_set->pollsets, pollset_set->pollset_capacity * + sizeof(*pollset_set->pollsets)); + } + pollset_set->pollsets[pollset_set->pollset_count++] = pollset; + for (i = 0; i < pollset_set->fd_count; i++) { + grpc_pollset_add_fd(pollset, pollset_set->fds[i]); + } + gpr_mu_unlock(&pollset_set->mu); +} + +void grpc_pollset_set_del_pollset(grpc_pollset_set *pollset_set, + grpc_pollset *pollset) { + size_t i; + gpr_mu_lock(&pollset_set->mu); + for (i = 0; i < pollset_set->pollset_count; i++) { + if (pollset_set->pollsets[i] == pollset) { + pollset_set->pollset_count--; + GPR_SWAP(grpc_pollset *, pollset_set->pollsets[i], + pollset_set->pollsets[pollset_set->pollset_count]); + break; + } + } + gpr_mu_unlock(&pollset_set->mu); +} + +void grpc_pollset_set_add_fd(grpc_pollset_set *pollset_set, grpc_fd *fd) { + size_t i; + gpr_mu_lock(&pollset_set->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( + pollset_set->fds, pollset_set->fd_capacity * sizeof(*pollset_set->fds)); + } + pollset_set->fds[pollset_set->fd_count++] = fd; + for (i = 0; i < pollset_set->pollset_count; i++) { + grpc_pollset_add_fd(pollset_set->pollsets[i], fd); + } + gpr_mu_unlock(&pollset_set->mu); +} + +void grpc_pollset_set_del_fd(grpc_pollset_set *pollset_set, grpc_fd *fd) { + size_t i; + gpr_mu_lock(&pollset_set->mu); + for (i = 0; i < pollset_set->fd_count; i++) { + if (pollset_set->fds[i] == fd) { + pollset_set->fd_count--; + GPR_SWAP(grpc_fd *, pollset_set->fds[i], + pollset_set->fds[pollset_set->pollset_count]); + break; + } + } + gpr_mu_unlock(&pollset_set->mu); +} + +#endif /* GPR_POSIX_SOCKET */ diff --git a/src/core/iomgr/pollset_set_posix.h b/src/core/iomgr/pollset_set_posix.h new file mode 100644 index 0000000000..e88740bde1 --- /dev/null +++ b/src/core/iomgr/pollset_set_posix.h @@ -0,0 +1,55 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_POSIX_H +#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_POSIX_H + +#include "src/core/iomgr/fd_posix.h" +#include "src/core/iomgr/pollset_posix.h" + +typedef struct grpc_pollset_set { + gpr_mu mu; + + size_t pollset_count; + size_t pollset_capacity; + grpc_pollset **pollsets; + + size_t fd_count; + size_t fd_capacity; + grpc_fd **fds; +} grpc_pollset_set; + +void grpc_pollset_set_add_fd(grpc_pollset_set *pollset_set, grpc_fd *fd); +void grpc_pollset_set_del_fd(grpc_pollset_set *pollset_set, grpc_fd *fd); + +#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_WINDOWS_H */ diff --git a/src/core/iomgr/pollset_kick_windows.h b/src/core/iomgr/pollset_set_windows.c index c675c119ab..d295c64b5e 100644 --- a/src/core/iomgr/pollset_kick_windows.h +++ b/src/core/iomgr/pollset_set_windows.c @@ -31,18 +31,17 @@ * */ -#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_WINDOWS_H -#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_WINDOWS_H +#include <grpc/support/port_platform.h> -#include <grpc/support/sync.h> +#ifdef GPR_WINSOCK_SOCKET -/* There isn't really any such thing as a pollset under Windows, due to the - nature of the IO completion ports. */ +#include "src/core/iomgr/pollset_set.h" -struct grpc_kick_fd_info; +void grpc_pollset_set_init(grpc_pollset_set *pollset_set) {} -typedef struct grpc_pollset_kick_state { - int unused; -} grpc_pollset_kick_state; +void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set) {} -#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_KICK_WINDOWS_H */ +void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, + grpc_pollset *pollset) {} + +#endif /* GPR_WINSOCK_SOCKET */ diff --git a/src/core/iomgr/pollset_set_windows.h b/src/core/iomgr/pollset_set_windows.h new file mode 100644 index 0000000000..0f91f1ede7 --- /dev/null +++ b/src/core/iomgr/pollset_set_windows.h @@ -0,0 +1,41 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_WINDOWS_H +#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_WINDOWS_H + +typedef struct grpc_pollset_set { + void *unused; +} grpc_pollset_set; + +#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_WINDOWS_H */ diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c index 5af0685f9d..8484bb1ff0 100644 --- a/src/core/iomgr/pollset_windows.c +++ b/src/core/iomgr/pollset_windows.c @@ -48,7 +48,6 @@ void grpc_pollset_init(grpc_pollset *pollset) { gpr_mu_init(&pollset->mu); - gpr_cv_init(&pollset->cv); } void grpc_pollset_shutdown(grpc_pollset *pollset, @@ -59,7 +58,6 @@ void grpc_pollset_shutdown(grpc_pollset *pollset, void grpc_pollset_destroy(grpc_pollset *pollset) { gpr_mu_destroy(&pollset->mu); - gpr_cv_destroy(&pollset->cv); } int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) { diff --git a/src/core/iomgr/pollset_windows.h b/src/core/iomgr/pollset_windows.h index e1115bac4f..acd82d0a0a 100644 --- a/src/core/iomgr/pollset_windows.h +++ b/src/core/iomgr/pollset_windows.h @@ -47,10 +47,8 @@ typedef struct grpc_pollset { gpr_mu mu; - gpr_cv cv; } grpc_pollset; #define GRPC_POLLSET_MU(pollset) (&(pollset)->mu) -#define GRPC_POLLSET_CV(pollset) (&(pollset)->cv) #endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_WINDOWS_H */ diff --git a/src/core/iomgr/resolve_address_posix.c b/src/core/iomgr/resolve_address_posix.c index 9a9283c93c..43fd704a6d 100644 --- a/src/core/iomgr/resolve_address_posix.c +++ b/src/core/iomgr/resolve_address_posix.c @@ -166,13 +166,14 @@ void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) { void grpc_resolve_address(const char *name, const char *default_port, grpc_resolve_cb cb, void *arg) { request *r = gpr_malloc(sizeof(request)); - gpr_thd_id id; + /*gpr_thd_id id;*/ grpc_iomgr_ref(); r->name = gpr_strdup(name); r->default_port = gpr_strdup(default_port); r->cb = cb; r->arg = arg; - gpr_thd_new(&id, do_request, r, NULL); + /*gpr_thd_new(&id, do_request, r, NULL);*/ + do_request(r); } #endif diff --git a/src/core/iomgr/tcp_client.h b/src/core/iomgr/tcp_client.h index 2e91497fb7..e1fdf235ec 100644 --- a/src/core/iomgr/tcp_client.h +++ b/src/core/iomgr/tcp_client.h @@ -35,6 +35,7 @@ #define GRPC_INTERNAL_CORE_IOMGR_TCP_CLIENT_H #include "src/core/iomgr/endpoint.h" +#include "src/core/iomgr/pollset_set.h" #include "src/core/iomgr/sockaddr.h" #include <grpc/support/time.h> @@ -42,7 +43,8 @@ cb with arg and the completed connection when done (or call cb with arg and NULL on failure) */ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), - void *arg, const struct sockaddr *addr, - int addr_len, gpr_timespec deadline); + void *arg, grpc_pollset_set *interested_parties, + const struct sockaddr *addr, int addr_len, + gpr_timespec deadline); #endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_CLIENT_H */ diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c index 2401fe00e4..f88ce19448 100644 --- a/src/core/iomgr/tcp_client_posix.c +++ b/src/core/iomgr/tcp_client_posix.c @@ -177,14 +177,16 @@ finish: } void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), - void *arg, const struct sockaddr *addr, - int addr_len, gpr_timespec deadline) { + void *arg, grpc_pollset_set *interested_parties, + const struct sockaddr *addr, int addr_len, + gpr_timespec deadline) { int fd; grpc_dualstack_mode dsmode; int err; async_connect *ac; struct sockaddr_in6 addr6_v4mapped; struct sockaddr_in addr4_copy; + grpc_fd *fdobj; /* Use dualstack sockets where available. */ if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) { @@ -211,16 +213,18 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), err = connect(fd, addr, addr_len); } while (err < 0 && errno == EINTR); + fdobj = grpc_fd_create(fd); + grpc_pollset_set_add_fd(interested_parties, fdobj); + if (err >= 0) { - gpr_log(GPR_DEBUG, "instant connect"); cb(arg, - grpc_tcp_create(grpc_fd_create(fd), GRPC_TCP_DEFAULT_READ_SLICE_SIZE)); + grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE)); return; } if (errno != EWOULDBLOCK && errno != EINPROGRESS) { gpr_log(GPR_ERROR, "connect error: %s", strerror(errno)); - close(fd); + grpc_fd_orphan(fdobj, NULL, NULL); cb(arg, NULL); return; } @@ -228,7 +232,7 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), ac = gpr_malloc(sizeof(async_connect)); ac->cb = cb; ac->cb_arg = arg; - ac->fd = grpc_fd_create(fd); + ac->fd = fdobj; gpr_mu_init(&ac->mu); ac->refs = 2; ac->write_closure.cb = on_writable; diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c index cf5174327d..f5d0ceccb9 100644 --- a/src/core/iomgr/tcp_client_windows.c +++ b/src/core/iomgr/tcp_client_windows.c @@ -138,9 +138,10 @@ static void on_connect(void *acp, int from_iocp) { /* Tries to issue one async connection, then schedules both an IOCP notification request for the connection, and one timeout alert. */ -void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp), - void *arg, const struct sockaddr *addr, - int addr_len, gpr_timespec deadline) { +void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), + void *arg, grpc_pollset_set *interested_parties, + const struct sockaddr *addr, int addr_len, + gpr_timespec deadline) { SOCKET sock = INVALID_SOCKET; BOOL success; int status; diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c index d1cd8a769c..002756f115 100644 --- a/src/core/iomgr/tcp_server_posix.c +++ b/src/core/iomgr/tcp_server_posix.c @@ -99,13 +99,14 @@ struct grpc_tcp_server { void *cb_arg; gpr_mu mu; - gpr_cv cv; /* active port count: how many ports are actually still listening */ size_t active_ports; /* destroyed port count: how many ports are completely destroyed */ size_t destroyed_ports; + int shutdown; + /* all listening ports */ server_port *ports; size_t nports; @@ -114,14 +115,17 @@ struct grpc_tcp_server { /* shutdown callback */ void (*shutdown_complete)(void *); void *shutdown_complete_arg; + + grpc_pollset **pollsets; + size_t pollset_count; }; grpc_tcp_server *grpc_tcp_server_create(void) { grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server)); gpr_mu_init(&s->mu); - gpr_cv_init(&s->cv); s->active_ports = 0; s->destroyed_ports = 0; + s->shutdown = 0; s->cb = NULL; s->cb_arg = NULL; s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP); @@ -134,7 +138,6 @@ static void finish_shutdown(grpc_tcp_server *s) { s->shutdown_complete(s->shutdown_complete_arg); gpr_mu_destroy(&s->mu); - gpr_cv_destroy(&s->cv); gpr_free(s->ports); gpr_free(s); @@ -154,28 +157,17 @@ static void destroyed_port(void *server, int success) { static void dont_care_about_shutdown_completion(void *ignored) {} -void grpc_tcp_server_destroy( - grpc_tcp_server *s, void (*shutdown_complete)(void *shutdown_complete_arg), - void *shutdown_complete_arg) { +static void deactivated_all_ports(grpc_tcp_server *s) { size_t i; - gpr_mu_lock(&s->mu); - s->shutdown_complete = shutdown_complete - ? shutdown_complete - : dont_care_about_shutdown_completion; - s->shutdown_complete_arg = shutdown_complete_arg; + /* delete ALL the things */ + gpr_mu_lock(&s->mu); - /* shutdown all fd's */ - for (i = 0; i < s->nports; i++) { - grpc_fd_shutdown(s->ports[i].emfd); - } - /* wait while that happens */ - /* TODO(ctiller): make this asynchronous also */ - while (s->active_ports) { - gpr_cv_wait(&s->cv, &s->mu, gpr_inf_future); + if (!s->shutdown) { + gpr_mu_unlock(&s->mu); + return; } - /* delete ALL the things */ if (s->nports) { for (i = 0; i < s->nports; i++) { server_port *sp = &s->ports[i]; @@ -191,6 +183,32 @@ void grpc_tcp_server_destroy( } } +void grpc_tcp_server_destroy( + grpc_tcp_server *s, void (*shutdown_complete)(void *shutdown_complete_arg), + void *shutdown_complete_arg) { + size_t i; + gpr_mu_lock(&s->mu); + + GPR_ASSERT(!s->shutdown); + s->shutdown = 1; + + s->shutdown_complete = shutdown_complete + ? shutdown_complete + : dont_care_about_shutdown_completion; + s->shutdown_complete_arg = shutdown_complete_arg; + + /* shutdown all fd's */ + if (s->active_ports) { + for (i = 0; i < s->nports; i++) { + grpc_fd_shutdown(s->ports[i].emfd); + } + gpr_mu_unlock(&s->mu); + } else { + gpr_mu_unlock(&s->mu); + deactivated_all_ports(s); + } +} + /* get max listen queue size on linux */ static void init_max_accept_queue_size(void) { int n = SOMAXCONN; @@ -272,6 +290,8 @@ error: /* event manager callback when reads are ready */ static void on_read(void *arg, int success) { server_port *sp = arg; + grpc_fd *fdobj; + size_t i; if (!success) { goto error; @@ -299,9 +319,16 @@ static void on_read(void *arg, int success) { grpc_set_socket_no_sigpipe_if_possible(fd); + fdobj = grpc_fd_create(fd); + /* TODO(ctiller): revise this when we have server-side sharding + of channels -- we certainly should not be automatically adding every + incoming channel to every pollset owned by the server */ + for (i = 0; i < sp->server->pollset_count; i++) { + grpc_pollset_add_fd(sp->server->pollsets[i], fdobj); + } sp->server->cb( sp->server->cb_arg, - grpc_tcp_create(grpc_fd_create(fd), GRPC_TCP_DEFAULT_READ_SLICE_SIZE)); + grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE)); } abort(); @@ -309,9 +336,11 @@ static void on_read(void *arg, int success) { error: gpr_mu_lock(&sp->server->mu); if (0 == --sp->server->active_ports) { - gpr_cv_broadcast(&sp->server->cv); + gpr_mu_unlock(&sp->server->mu); + deactivated_all_ports(sp->server); + } else { + gpr_mu_unlock(&sp->server->mu); } - gpr_mu_unlock(&sp->server->mu); } static int add_socket_to_server(grpc_tcp_server *s, int fd, @@ -436,6 +465,8 @@ void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollsets, GPR_ASSERT(s->active_ports == 0); s->cb = cb; s->cb_arg = cb_arg; + s->pollsets = pollsets; + s->pollset_count = pollset_count; for (i = 0; i < s->nports; i++) { for (j = 0; j < pollset_count; j++) { grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd); diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c index ae22bf47a0..79b14c0a83 100644 --- a/src/core/security/credentials.c +++ b/src/core/security/credentials.c @@ -186,9 +186,7 @@ static void ssl_server_destroy(grpc_server_credentials *creds) { gpr_free(creds); } -static int ssl_has_request_metadata(const grpc_credentials *creds) { - return 0; -} +static int ssl_has_request_metadata(const grpc_credentials *creds) { return 0; } static int ssl_has_request_metadata_only(const grpc_credentials *creds) { return 0; @@ -363,7 +361,6 @@ static int jwt_has_request_metadata_only(const grpc_credentials *creds) { return 1; } - static void jwt_get_request_metadata(grpc_credentials *creds, const char *service_url, grpc_credentials_metadata_cb cb, @@ -445,6 +442,7 @@ grpc_credentials *grpc_jwt_credentials_create(const char *json_key, from an http service. */ typedef void (*grpc_fetch_oauth2_func)(grpc_credentials_metadata_request *req, + grpc_httpcli_context *http_context, grpc_httpcli_response_cb response_cb, gpr_timespec deadline); @@ -453,6 +451,8 @@ typedef struct { gpr_mu mu; grpc_credentials_md_store *access_token_md; gpr_timespec token_expiration; + grpc_httpcli_context httpcli_context; + grpc_pollset_set pollset_set; grpc_fetch_oauth2_func fetch_func; } grpc_oauth2_token_fetcher_credentials; @@ -461,6 +461,7 @@ static void oauth2_token_fetcher_destroy(grpc_credentials *creds) { (grpc_oauth2_token_fetcher_credentials *)creds; grpc_credentials_md_store_unref(c->access_token_md); gpr_mu_destroy(&c->mu); + grpc_httpcli_context_destroy(&c->httpcli_context); gpr_free(c); } @@ -611,7 +612,7 @@ static void oauth2_token_fetcher_get_request_metadata( } else { c->fetch_func( grpc_credentials_metadata_request_create(creds, cb, user_data), - on_oauth2_token_fetcher_http_response, + &c->httpcli_context, on_oauth2_token_fetcher_http_response, gpr_time_add(gpr_now(), refresh_threshold)); } } @@ -624,18 +625,21 @@ static void init_oauth2_token_fetcher(grpc_oauth2_token_fetcher_credentials *c, gpr_mu_init(&c->mu); c->token_expiration = gpr_inf_past; c->fetch_func = fetch_func; + grpc_pollset_set_init(&c->pollset_set); } /* -- ComputeEngine credentials. -- */ static grpc_credentials_vtable compute_engine_vtable = { - oauth2_token_fetcher_destroy, oauth2_token_fetcher_has_request_metadata, + oauth2_token_fetcher_destroy, + oauth2_token_fetcher_has_request_metadata, oauth2_token_fetcher_has_request_metadata_only, oauth2_token_fetcher_get_request_metadata, NULL}; static void compute_engine_fetch_oauth2( grpc_credentials_metadata_request *metadata_req, - grpc_httpcli_response_cb response_cb, gpr_timespec deadline) { + grpc_httpcli_context *httpcli_context, grpc_httpcli_response_cb response_cb, + gpr_timespec deadline) { grpc_httpcli_header header = {"Metadata-Flavor", "Google"}; grpc_httpcli_request request; memset(&request, 0, sizeof(grpc_httpcli_request)); @@ -643,7 +647,8 @@ static void compute_engine_fetch_oauth2( request.path = GRPC_COMPUTE_ENGINE_METADATA_TOKEN_PATH; request.hdr_count = 1; request.hdrs = &header; - grpc_httpcli_get(&request, deadline, response_cb, metadata_req); + grpc_httpcli_get(httpcli_context, &request, deadline, response_cb, + metadata_req); } grpc_credentials *grpc_compute_engine_credentials_create(void) { @@ -672,13 +677,15 @@ static void service_account_destroy(grpc_credentials *creds) { } static grpc_credentials_vtable service_account_vtable = { - service_account_destroy, oauth2_token_fetcher_has_request_metadata, + service_account_destroy, + oauth2_token_fetcher_has_request_metadata, oauth2_token_fetcher_has_request_metadata_only, oauth2_token_fetcher_get_request_metadata, NULL}; static void service_account_fetch_oauth2( grpc_credentials_metadata_request *metadata_req, - grpc_httpcli_response_cb response_cb, gpr_timespec deadline) { + grpc_httpcli_context *httpcli_context, grpc_httpcli_response_cb response_cb, + gpr_timespec deadline) { grpc_service_account_credentials *c = (grpc_service_account_credentials *)metadata_req->creds; grpc_httpcli_header header = {"Content-Type", @@ -703,8 +710,8 @@ static void service_account_fetch_oauth2( request.hdr_count = 1; request.hdrs = &header; request.use_ssl = 1; - grpc_httpcli_post(&request, body, strlen(body), deadline, response_cb, - metadata_req); + grpc_httpcli_post(httpcli_context, &request, body, strlen(body), deadline, + response_cb, metadata_req); gpr_free(body); gpr_free(jwt); } @@ -738,20 +745,21 @@ typedef struct { } grpc_refresh_token_credentials; static void refresh_token_destroy(grpc_credentials *creds) { - grpc_refresh_token_credentials *c = - (grpc_refresh_token_credentials *)creds; + grpc_refresh_token_credentials *c = (grpc_refresh_token_credentials *)creds; grpc_auth_refresh_token_destruct(&c->refresh_token); oauth2_token_fetcher_destroy(&c->base.base); } static grpc_credentials_vtable refresh_token_vtable = { - refresh_token_destroy, oauth2_token_fetcher_has_request_metadata, + refresh_token_destroy, + oauth2_token_fetcher_has_request_metadata, oauth2_token_fetcher_has_request_metadata_only, oauth2_token_fetcher_get_request_metadata, NULL}; static void refresh_token_fetch_oauth2( grpc_credentials_metadata_request *metadata_req, - grpc_httpcli_response_cb response_cb, gpr_timespec deadline) { + grpc_httpcli_context *httpcli_context, grpc_httpcli_response_cb response_cb, + gpr_timespec deadline) { grpc_refresh_token_credentials *c = (grpc_refresh_token_credentials *)metadata_req->creds; grpc_httpcli_header header = {"Content-Type", @@ -767,8 +775,8 @@ static void refresh_token_fetch_oauth2( request.hdr_count = 1; request.hdrs = &header; request.use_ssl = 1; - grpc_httpcli_post(&request, body, strlen(body), deadline, response_cb, - metadata_req); + grpc_httpcli_post(httpcli_context, &request, body, strlen(body), deadline, + response_cb, metadata_req); gpr_free(body); } @@ -779,8 +787,7 @@ grpc_credentials *grpc_refresh_token_credentials_create( grpc_auth_refresh_token_create_from_string(json_refresh_token); if (!grpc_auth_refresh_token_is_valid(&refresh_token)) { - gpr_log(GPR_ERROR, - "Invalid input for refresh token credentials creation"); + gpr_log(GPR_ERROR, "Invalid input for refresh token credentials creation"); return NULL; } c = gpr_malloc(sizeof(grpc_refresh_token_credentials)); @@ -1171,9 +1178,7 @@ static void iam_destroy(grpc_credentials *creds) { gpr_free(c); } -static int iam_has_request_metadata(const grpc_credentials *creds) { - return 1; -} +static int iam_has_request_metadata(const grpc_credentials *creds) { return 1; } static int iam_has_request_metadata_only(const grpc_credentials *creds) { return 1; diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h index 4768ce6990..85931d384d 100644 --- a/src/core/security/credentials.h +++ b/src/core/security/credentials.h @@ -131,7 +131,6 @@ typedef struct { grpc_credentials *c, const char *target, const grpc_channel_args *args, grpc_credentials *request_metadata_creds, grpc_channel_security_connector **sc, grpc_channel_args **new_args); - } grpc_credentials_vtable; struct grpc_credentials { @@ -200,4 +199,4 @@ struct grpc_server_credentials { grpc_security_status grpc_server_credentials_create_security_connector( grpc_server_credentials *creds, grpc_security_connector **sc); -#endif /* GRPC_INTERNAL_CORE_SECURITY_CREDENTIALS_H */ +#endif /* GRPC_INTERNAL_CORE_SECURITY_CREDENTIALS_H */ diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c index 0e4b9fc9d3..6cc3e27f4c 100644 --- a/src/core/security/google_default_credentials.c +++ b/src/core/security/google_default_credentials.c @@ -60,8 +60,7 @@ static void init_default_credentials(void) { } typedef struct { - gpr_cv cv; - gpr_mu mu; + grpc_pollset pollset; int is_done; int success; } compute_engine_detector; @@ -82,22 +81,22 @@ static void on_compute_engine_detection_http_response( } } } - gpr_mu_lock(&detector->mu); + gpr_mu_lock(GRPC_POLLSET_MU(&detector->pollset)); detector->is_done = 1; - gpr_mu_unlock(&detector->mu); - gpr_cv_signal(&detector->cv); + grpc_pollset_kick(&detector->pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&detector->pollset)); } static int is_stack_running_on_compute_engine(void) { compute_engine_detector detector; grpc_httpcli_request request; + grpc_httpcli_context context; /* The http call is local. If it takes more than one sec, it is for sure not on compute engine. */ gpr_timespec max_detection_delay = {1, 0}; - gpr_mu_init(&detector.mu); - gpr_cv_init(&detector.cv); + grpc_pollset_init(&detector.pollset); detector.is_done = 0; detector.success = 0; @@ -105,19 +104,24 @@ static int is_stack_running_on_compute_engine(void) { request.host = GRPC_COMPUTE_ENGINE_DETECTION_HOST; request.path = "/"; - grpc_httpcli_get(&request, gpr_time_add(gpr_now(), max_detection_delay), - on_compute_engine_detection_http_response, &detector); + grpc_httpcli_context_init(&context); + grpc_httpcli_context_add_interested_party(&context, &detector.pollset); + + grpc_httpcli_get(&context, &request, gpr_time_add(gpr_now(), max_detection_delay), + on_compute_engine_detection_http_response, + &detector); /* Block until we get the response. This is not ideal but this should only be called once for the lifetime of the process by the default credentials. */ - gpr_mu_lock(&detector.mu); + gpr_mu_lock(GRPC_POLLSET_MU(&detector.pollset)); while (!detector.is_done) { - gpr_cv_wait(&detector.cv, &detector.mu, gpr_inf_future); + grpc_pollset_work(&detector.pollset, gpr_inf_future); } - gpr_mu_unlock(&detector.mu); + gpr_mu_unlock(GRPC_POLLSET_MU(&detector.pollset)); + + grpc_httpcli_context_destroy(&context); + grpc_pollset_destroy(&detector.pollset); - gpr_mu_destroy(&detector.mu); - gpr_cv_destroy(&detector.cv); return detector.success; } diff --git a/src/core/support/sync.c b/src/core/support/sync.c index ccfe1e25f4..856b5adb86 100644 --- a/src/core/support/sync.c +++ b/src/core/support/sync.c @@ -118,7 +118,9 @@ void gpr_refn(gpr_refcount *r, int n) { } int gpr_unref(gpr_refcount *r) { - return gpr_atm_full_fetch_add(&r->count, -1) == 1; + gpr_atm prior = gpr_atm_full_fetch_add(&r->count, -1); + GPR_ASSERT(prior > 0); + return prior == 1; } void gpr_stats_init(gpr_stats_counter *c, gpr_intptr n) { diff --git a/src/core/surface/call.c b/src/core/surface/call.c index e3995a407b..138cd084c5 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -150,6 +150,8 @@ struct grpc_call { gpr_uint8 num_completed_requests; /* are we currently reading a message? */ gpr_uint8 reading_message; + /* have we bound a pollset yet? */ + gpr_uint8 bound_pollset; /* flags with bits corresponding to write states allowing us to determine what was sent */ gpr_uint16 last_send_contains; @@ -242,9 +244,12 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op); static void execute_op(grpc_call *call, grpc_transport_op *op); static void recv_metadata(grpc_call *call, grpc_metadata_batch *metadata); static void finish_read_ops(grpc_call *call); -static grpc_call_error cancel_with_status( - grpc_call *c, grpc_status_code status, const char *description, - gpr_uint8 locked); +static grpc_call_error cancel_with_status(grpc_call *c, grpc_status_code status, + const char *description, + gpr_uint8 locked); + +static void lock(grpc_call *call); +static void unlock(grpc_call *call); grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, const void *server_transport_data, @@ -275,7 +280,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, } call->send_initial_metadata_count = add_initial_metadata_count; call->send_deadline = send_deadline; - grpc_channel_internal_ref(channel); + GRPC_CHANNEL_INTERNAL_REF(channel, "call"); call->metadata_context = grpc_channel_get_metadata_context(channel); grpc_sopb_init(&call->send_ops); grpc_sopb_init(&call->recv_ops); @@ -305,7 +310,9 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, void grpc_call_set_completion_queue(grpc_call *call, grpc_completion_queue *cq) { + lock(call); call->cq = cq; + unlock(call); } grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) { @@ -326,7 +333,7 @@ static void destroy_call(void *call, int ignored_success) { size_t i; grpc_call *c = call; grpc_call_stack_destroy(CALL_STACK_FROM_CALL(c)); - grpc_channel_internal_unref(c->channel); + GRPC_CHANNEL_INTERNAL_UNREF(c->channel, "call"); gpr_mu_destroy(&c->mu); for (i = 0; i < STATUS_SOURCE_COUNT; i++) { if (c->status[i].details) { @@ -403,7 +410,8 @@ static void lock(grpc_call *call) { gpr_mu_lock(&call->mu); } static int need_more_data(grpc_call *call) { if (call->read_state == READ_STATE_STREAM_CLOSED) return 0; return is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA) || - (is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) && grpc_bbq_empty(&call->incoming_queue)) || + (is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) && + grpc_bbq_empty(&call->incoming_queue)) || is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA) || is_op_live(call, GRPC_IOREQ_RECV_STATUS) || is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS) || @@ -421,6 +429,12 @@ static void unlock(grpc_call *call) { memset(&op, 0, sizeof(op)); + if (!call->bound_pollset && call->cq) { + call->bound_pollset = 1; + op.bind_pollset = grpc_cq_pollset(call->cq); + start_op = 1; + } + if (!call->receiving && need_more_data(call)) { op.recv_ops = &call->recv_ops; op.recv_state = &call->recv_state; @@ -556,13 +570,13 @@ static void finish_live_ioreq_op(grpc_call *call, grpc_ioreq_op op, break; case GRPC_IOREQ_RECV_INITIAL_METADATA: GPR_SWAP(grpc_metadata_array, call->buffered_metadata[0], - *call->request_data[GRPC_IOREQ_RECV_INITIAL_METADATA] - .recv_metadata); + *call->request_data[GRPC_IOREQ_RECV_INITIAL_METADATA] + .recv_metadata); break; case GRPC_IOREQ_RECV_TRAILING_METADATA: GPR_SWAP(grpc_metadata_array, call->buffered_metadata[1], - *call->request_data[GRPC_IOREQ_RECV_TRAILING_METADATA] - .recv_metadata); + *call->request_data[GRPC_IOREQ_RECV_TRAILING_METADATA] + .recv_metadata); break; case GRPC_IOREQ_OP_COUNT: abort(); @@ -676,9 +690,8 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) { } /* we have to be reading a message to know what to do here */ if (!call->reading_message) { - cancel_with_status( - call, GRPC_STATUS_INVALID_ARGUMENT, - "Received payload data while not reading a message", 1); + cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, + "Received payload data while not reading a message", 1); return 0; } /* append the slice to the incoming buffer */ @@ -831,7 +844,6 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op) { } grpc_sopb_add_metadata(&call->send_ops, mdb); op->send_ops = &call->send_ops; - op->bind_pollset = grpc_cq_pollset(call->cq); call->last_send_contains |= 1 << GRPC_IOREQ_SEND_INITIAL_METADATA; call->send_initial_metadata_count = 0; /* fall through intended */ @@ -1025,9 +1037,9 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c, return cancel_with_status(c, status, description, 0); } -static grpc_call_error cancel_with_status( - grpc_call *c, grpc_status_code status, const char *description, - gpr_uint8 locked) { +static grpc_call_error cancel_with_status(grpc_call *c, grpc_status_code status, + const char *description, + gpr_uint8 locked) { grpc_transport_op op; grpc_mdstr *details = description ? grpc_mdstr_from_string(c->metadata_context, description) @@ -1294,8 +1306,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, grpc_cq_begin_op(call->cq, call); - return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_func, - tag); + return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_func, tag); } void grpc_call_context_set(grpc_call *call, grpc_context_index elem, void *value, diff --git a/src/core/surface/call.h b/src/core/surface/call.h index 9116538948..17db8c2cdc 100644 --- a/src/core/surface/call.h +++ b/src/core/surface/call.h @@ -95,8 +95,10 @@ grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call); #ifdef GRPC_CALL_REF_COUNT_DEBUG void grpc_call_internal_ref(grpc_call *call, const char *reason); -void grpc_call_internal_unref(grpc_call *call, const char *reason, int allow_immediate_deletion); -#define GRPC_CALL_INTERNAL_REF(call, reason) grpc_call_internal_ref(call, reason) +void grpc_call_internal_unref(grpc_call *call, const char *reason, + int allow_immediate_deletion); +#define GRPC_CALL_INTERNAL_REF(call, reason) \ + grpc_call_internal_ref(call, reason) #define GRPC_CALL_INTERNAL_UNREF(call, reason, allow_immediate_deletion) \ grpc_call_internal_unref(call, reason, allow_immediate_deletion) #else @@ -124,8 +126,7 @@ void grpc_call_log_batch(char *file, int line, gpr_log_severity severity, void grpc_server_log_request_call(char *file, int line, gpr_log_severity severity, - grpc_server *server, - grpc_call **call, + grpc_server *server, grpc_call **call, grpc_call_details *details, grpc_metadata_array *initial_metadata, grpc_completion_queue *cq_bound_to_call, @@ -134,16 +135,20 @@ void grpc_server_log_request_call(char *file, int line, /* Set a context pointer. No thread safety guarantees are made wrt this value. */ -void grpc_call_context_set(grpc_call *call, grpc_context_index elem, void *value, - void (*destroy)(void *value)); +void grpc_call_context_set(grpc_call *call, grpc_context_index elem, + void *value, void (*destroy)(void *value)); /* Get a context pointer. */ void *grpc_call_context_get(grpc_call *call, grpc_context_index elem); #define GRPC_CALL_LOG_BATCH(sev, call, ops, nops, tag) \ if (grpc_trace_batch) grpc_call_log_batch(sev, call, ops, nops, tag) -#define GRPC_SERVER_LOG_REQUEST_CALL(sev, server, call, details, initial_metadata, cq_bound_to_call, cq_for_notifications, tag) \ - if (grpc_trace_batch) grpc_server_log_request_call(sev, server, call, details, initial_metadata, cq_bound_to_call, cq_for_notifications, tag) +#define GRPC_SERVER_LOG_REQUEST_CALL(sev, server, call, details, \ + initial_metadata, cq_bound_to_call, \ + cq_for_notifications, tag) \ + if (grpc_trace_batch) \ + grpc_server_log_request_call(sev, server, call, details, initial_metadata, \ + cq_bound_to_call, cq_for_notifications, tag) gpr_uint8 grpc_call_is_client(grpc_call *call); diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c index be9da2b7f9..d8fd442702 100644 --- a/src/core/surface/channel.c +++ b/src/core/surface/channel.c @@ -168,8 +168,14 @@ grpc_call *grpc_channel_create_registered_call( grpc_mdelem_ref(rc->authority), deadline); } -void grpc_channel_internal_ref(grpc_channel *channel) { - gpr_ref(&channel->refs); +#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG +void grpc_channel_internal_ref(grpc_channel *c, const char *reason) { + gpr_log(GPR_DEBUG, "CHANNEL: ref %p %d -> %d [%s]", c, c->refs.count, + c->refs.count + 1, reason); +#else +void grpc_channel_internal_ref(grpc_channel *c) { +#endif + gpr_ref(&c->refs); } static void destroy_channel(void *p, int ok) { @@ -191,9 +197,15 @@ static void destroy_channel(void *p, int ok) { gpr_free(channel); } -void grpc_channel_internal_unref(grpc_channel *channel) { - if (gpr_unref(&channel->refs)) { - grpc_iomgr_add_callback(destroy_channel, channel); +#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG +void grpc_channel_internal_unref(grpc_channel *c, const char *reason) { + gpr_log(GPR_DEBUG, "CHANNEL: unref %p %d -> %d [%s]", c, c->refs.count, + c->refs.count - 1, reason); +#else +void grpc_channel_internal_unref(grpc_channel *c) { +#endif + if (gpr_unref(&c->refs)) { + grpc_iomgr_add_callback(destroy_channel, c); } } @@ -213,11 +225,11 @@ void grpc_channel_destroy(grpc_channel *channel) { op.dir = GRPC_CALL_DOWN; elem->filter->channel_op(elem, NULL, &op); - grpc_channel_internal_unref(channel); + GRPC_CHANNEL_INTERNAL_UNREF(channel, "channel"); } void grpc_client_channel_closed(grpc_channel_element *elem) { - grpc_channel_internal_unref(CHANNEL_FROM_TOP_ELEM(elem)); + GRPC_CHANNEL_INTERNAL_UNREF(CHANNEL_FROM_TOP_ELEM(elem), "closed"); } grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel) { diff --git a/src/core/surface/channel.h b/src/core/surface/channel.h index 388be35711..d1f62f2598 100644 --- a/src/core/surface/channel.h +++ b/src/core/surface/channel.h @@ -48,7 +48,20 @@ gpr_uint32 grpc_channel_get_max_message_length(grpc_channel *channel); void grpc_client_channel_closed(grpc_channel_element *elem); +#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG +void grpc_channel_internal_ref(grpc_channel *channel, const char *reason); +void grpc_channel_internal_unref(grpc_channel *channel, const char *reason); +#define GRPC_CHANNEL_INTERNAL_REF(channel, reason) \ + grpc_channel_internal_ref(channel, reason) +#define GRPC_CHANNEL_INTERNAL_UNREF(channel, reason) \ + grpc_channel_internal_unref(channel, reason) +#else void grpc_channel_internal_ref(grpc_channel *channel); void grpc_channel_internal_unref(grpc_channel *channel); +#define GRPC_CHANNEL_INTERNAL_REF(channel, reason) \ + grpc_channel_internal_ref(channel) +#define GRPC_CHANNEL_INTERNAL_UNREF(channel, reason) \ + grpc_channel_internal_unref(channel) +#endif #endif /* GRPC_INTERNAL_CORE_SURFACE_CHANNEL_H */ diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c index daa8d3a7c6..061669ec6e 100644 --- a/src/core/surface/channel_create.c +++ b/src/core/surface/channel_create.c @@ -90,7 +90,7 @@ static void done(request *r, int was_successful) { static void on_connect(void *rp, grpc_endpoint *tcp) { request *r = rp; - if (!grpc_client_setup_request_should_continue(r->cs_request)) { + if (!grpc_client_setup_request_should_continue(r->cs_request, "on_connect")) { if (tcp) { grpc_endpoint_shutdown(tcp); grpc_endpoint_destroy(tcp); @@ -106,12 +106,12 @@ static void on_connect(void *rp, grpc_endpoint *tcp) { } else { return; } - } else if (grpc_client_setup_cb_begin(r->cs_request)) { + } else if (grpc_client_setup_cb_begin(r->cs_request, "on_connect")) { grpc_create_chttp2_transport( r->setup->setup_callback, r->setup->setup_user_data, grpc_client_setup_get_channel_args(r->cs_request), tcp, NULL, 0, grpc_client_setup_get_mdctx(r->cs_request), 1); - grpc_client_setup_cb_end(r->cs_request); + grpc_client_setup_cb_end(r->cs_request, "on_connect"); done(r, 1); return; } else { @@ -125,9 +125,10 @@ static int maybe_try_next_resolved(request *r) { if (!r->resolved) return 0; if (r->resolved_index == r->resolved->naddrs) return 0; addr = &r->resolved->addrs[r->resolved_index++]; - grpc_tcp_client_connect(on_connect, r, (struct sockaddr *)&addr->addr, - addr->len, - grpc_client_setup_request_deadline(r->cs_request)); + grpc_tcp_client_connect( + on_connect, r, grpc_client_setup_get_interested_parties(r->cs_request), + (struct sockaddr *)&addr->addr, addr->len, + grpc_client_setup_request_deadline(r->cs_request)); return 1; } @@ -136,7 +137,7 @@ static void on_resolved(void *rp, grpc_resolved_addresses *resolved) { request *r = rp; /* if we're not still the active request, abort */ - if (!grpc_client_setup_request_should_continue(r->cs_request)) { + if (!grpc_client_setup_request_should_continue(r->cs_request, "on_resolved")) { if (resolved) { grpc_resolved_addresses_destroy(resolved); } diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index 8c9ca48a05..6808c976e1 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -59,9 +59,6 @@ typedef struct event { /* Completion queue structure */ struct grpc_completion_queue { - /* TODO(ctiller): see if this can be removed */ - int allow_polling; - /* When refs drops to zero, we are in shutdown mode, and will be destroyable once all queued events are drained */ gpr_refcount refs; @@ -83,33 +80,51 @@ grpc_completion_queue *grpc_completion_queue_create(void) { memset(cc, 0, sizeof(*cc)); /* Initial ref is dropped by grpc_completion_queue_shutdown */ gpr_ref_init(&cc->refs, 1); - gpr_ref_init(&cc->owning_refs, 1); + /* One for destroy(), one for pollset_shutdown */ + gpr_ref_init(&cc->owning_refs, 2); grpc_pollset_init(&cc->pollset); - cc->allow_polling = 1; return cc; } + + + + + + + + + + + + +#ifdef GRPC_CQ_REF_COUNT_DEBUG +void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason) { + gpr_log(GPR_DEBUG, "CQ:%p ref %d -> %d %s", cc, (int)cc->owning_refs.count, (int)cc->owning_refs.count + 1, reason); +#else void grpc_cq_internal_ref(grpc_completion_queue *cc) { +#endif gpr_ref(&cc->owning_refs); } static void on_pollset_destroy_done(void *arg) { grpc_completion_queue *cc = arg; - grpc_pollset_destroy(&cc->pollset); - gpr_free(cc); + GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy"); } +#ifdef GRPC_CQ_REF_COUNT_DEBUG +void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason) { + gpr_log(GPR_DEBUG, "CQ:%p unref %d -> %d %s", cc, (int)cc->owning_refs.count, (int)cc->owning_refs.count - 1, reason); +#else void grpc_cq_internal_unref(grpc_completion_queue *cc) { +#endif if (gpr_unref(&cc->owning_refs)) { GPR_ASSERT(cc->queue == NULL); - grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc); + grpc_pollset_destroy(&cc->pollset); + gpr_free(cc); } } -void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc) { - cc->allow_polling = 0; -} - /* Create and append an event to the queue. Returns the event so that its data members can be filled in. Requires GRPC_POLLSET_MU(&cc->pollset) locked. */ @@ -133,7 +148,6 @@ static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type, ev->bucket_prev = cc->buckets[bucket]->bucket_prev; ev->bucket_next->bucket_prev = ev->bucket_prev->bucket_next = ev; } - gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset)); grpc_pollset_kick(&cc->pollset); return ev; } @@ -145,25 +159,24 @@ void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call) { /* Signal the end of an operation - if this is the last waiting-to-be-queued event, then enter shutdown mode */ -static void end_op_locked(grpc_completion_queue *cc, - grpc_completion_type type) { - if (gpr_unref(&cc->refs)) { - GPR_ASSERT(!cc->shutdown); - GPR_ASSERT(cc->shutdown_called); - cc->shutdown = 1; - gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset)); - } -} - void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call, int success) { event *ev; + int shutdown = 0; gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); ev = add_locked(cc, GRPC_OP_COMPLETE, tag, call); ev->base.success = success; - end_op_locked(cc, GRPC_OP_COMPLETE); + if (gpr_unref(&cc->refs)) { + GPR_ASSERT(!cc->shutdown); + GPR_ASSERT(cc->shutdown_called); + cc->shutdown = 1; + shutdown = 1; + } gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); if (call) GRPC_CALL_INTERNAL_UNREF(call, "cq", 0); + if (shutdown) { + grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc); + } } /* Create a GRPC_QUEUE_SHUTDOWN event without queuing it anywhere */ @@ -179,6 +192,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, event *ev = NULL; grpc_event ret; + GRPC_CQ_INTERNAL_REF(cc, "next"); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); for (;;) { if (cc->queue != NULL) { @@ -205,15 +219,12 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, ev = create_shutdown_event(); break; } - if (cc->allow_polling && grpc_pollset_work(&cc->pollset, deadline)) { - continue; - } - if (gpr_cv_wait(GRPC_POLLSET_CV(&cc->pollset), - GRPC_POLLSET_MU(&cc->pollset), deadline)) { + if (!grpc_pollset_work(&cc->pollset, deadline)) { gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); memset(&ret, 0, sizeof(ret)); ret.type = GRPC_QUEUE_TIMEOUT; GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); + GRPC_CQ_INTERNAL_UNREF(cc, "next"); return ret; } } @@ -221,6 +232,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, ret = ev->base; gpr_free(ev); GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); + GRPC_CQ_INTERNAL_UNREF(cc, "next"); return ret; } @@ -258,6 +270,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, event *ev = NULL; grpc_event ret; + GRPC_CQ_INTERNAL_REF(cc, "pluck"); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); for (;;) { if ((ev = pluck_event(cc, tag))) { @@ -267,15 +280,12 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, ev = create_shutdown_event(); break; } - if (cc->allow_polling && grpc_pollset_work(&cc->pollset, deadline)) { - continue; - } - if (gpr_cv_wait(GRPC_POLLSET_CV(&cc->pollset), - GRPC_POLLSET_MU(&cc->pollset), deadline)) { + if (!grpc_pollset_work(&cc->pollset, deadline)) { gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); memset(&ret, 0, sizeof(ret)); ret.type = GRPC_QUEUE_TIMEOUT; GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); + GRPC_CQ_INTERNAL_UNREF(cc, "pluck"); return ret; } } @@ -283,6 +293,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, ret = ev->base; gpr_free(ev); GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); + GRPC_CQ_INTERNAL_UNREF(cc, "pluck"); return ret; } @@ -290,6 +301,10 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, to zero here, then enter shutdown mode and wake up any waiters */ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) { gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); + if (cc->shutdown_called) { + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); + return; + } cc->shutdown_called = 1; gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); @@ -297,13 +312,14 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) { gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); GPR_ASSERT(!cc->shutdown); cc->shutdown = 1; - gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset)); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); + grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc); } } void grpc_completion_queue_destroy(grpc_completion_queue *cc) { - grpc_cq_internal_unref(cc); + grpc_completion_queue_shutdown(cc); + GRPC_CQ_INTERNAL_UNREF(cc, "destroy"); } grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) { diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h index 7b6fad98fd..2249d0e789 100644 --- a/src/core/surface/completion_queue.h +++ b/src/core/surface/completion_queue.h @@ -39,8 +39,17 @@ #include "src/core/iomgr/pollset.h" #include <grpc/grpc.h> +#ifdef GRPC_CQ_REF_COUNT_DEBUG +void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason); +void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason); +#define GRPC_CQ_INTERNAL_REF(cc, reason) grpc_cq_internal_ref(cc, reason) +#define GRPC_CQ_INTERNAL_UNREF(cc, reason) grpc_cq_internal_unref(cc, reason) +#else void grpc_cq_internal_ref(grpc_completion_queue *cc); void grpc_cq_internal_unref(grpc_completion_queue *cc); +#define GRPC_CQ_INTERNAL_REF(cc, reason) grpc_cq_internal_ref(cc) +#define GRPC_CQ_INTERNAL_UNREF(cc, reason) grpc_cq_internal_unref(cc) +#endif /* Flag that an operation is beginning: the completion channel will not finish shutdown until a corrensponding grpc_cq_end_* call is made */ @@ -50,9 +59,6 @@ void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call); void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call, int success); -/* disable polling for some tests */ -void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc); - grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc); void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc); diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index a71d12291e..b9c6d5f760 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -96,12 +96,12 @@ static void on_secure_transport_setup_done(void *rp, if (status != GRPC_SECURITY_OK) { gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status); done(r, 0); - } else if (grpc_client_setup_cb_begin(r->cs_request)) { + } else if (grpc_client_setup_cb_begin(r->cs_request, "on_secure_transport_setup_done")) { grpc_create_chttp2_transport( r->setup->setup_callback, r->setup->setup_user_data, grpc_client_setup_get_channel_args(r->cs_request), secure_endpoint, NULL, 0, grpc_client_setup_get_mdctx(r->cs_request), 1); - grpc_client_setup_cb_end(r->cs_request); + grpc_client_setup_cb_end(r->cs_request, "on_secure_transport_setup_done"); done(r, 1); } else { done(r, 0); @@ -112,7 +112,7 @@ static void on_secure_transport_setup_done(void *rp, static void on_connect(void *rp, grpc_endpoint *tcp) { request *r = rp; - if (!grpc_client_setup_request_should_continue(r->cs_request)) { + if (!grpc_client_setup_request_should_continue(r->cs_request, "on_connect.secure")) { if (tcp) { grpc_endpoint_shutdown(tcp); grpc_endpoint_destroy(tcp); @@ -140,9 +140,10 @@ static int maybe_try_next_resolved(request *r) { if (!r->resolved) return 0; if (r->resolved_index == r->resolved->naddrs) return 0; addr = &r->resolved->addrs[r->resolved_index++]; - grpc_tcp_client_connect(on_connect, r, (struct sockaddr *)&addr->addr, - addr->len, - grpc_client_setup_request_deadline(r->cs_request)); + grpc_tcp_client_connect( + on_connect, r, grpc_client_setup_get_interested_parties(r->cs_request), + (struct sockaddr *)&addr->addr, addr->len, + grpc_client_setup_request_deadline(r->cs_request)); return 1; } @@ -151,7 +152,7 @@ static void on_resolved(void *rp, grpc_resolved_addresses *resolved) { request *r = rp; /* if we're not still the active request, abort */ - if (!grpc_client_setup_request_should_continue(r->cs_request)) { + if (!grpc_client_setup_request_should_continue(r->cs_request, "on_resolved.secure")) { if (resolved) { grpc_resolved_addresses_destroy(resolved); } diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 60606c75e4..99e3ee6d2e 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -113,6 +113,7 @@ typedef struct channel_registered_method { struct channel_data { grpc_server *server; + size_t num_calls; grpc_channel *channel; grpc_mdstr *path_key; grpc_mdstr *authority_key; @@ -124,6 +125,11 @@ struct channel_data { gpr_uint32 registered_method_max_probes; }; +typedef struct shutdown_tag { + void *tag; + grpc_completion_queue *cq; +} shutdown_tag; + struct grpc_server { size_t channel_filter_count; const grpc_channel_filter **channel_filters; @@ -134,14 +140,14 @@ struct grpc_server { size_t cq_count; gpr_mu mu; - gpr_cv cv; registered_method *registered_methods; requested_call_array requested_calls; gpr_uint8 shutdown; + gpr_uint8 shutdown_published; size_t num_shutdown_tags; - void **shutdown_tags; + shutdown_tag *shutdown_tags; call_data *lists[CALL_LIST_COUNT]; channel_data root_channel_data; @@ -188,6 +194,9 @@ struct call_data { static void begin_call(grpc_server *server, call_data *calld, requested_call *rc); static void fail_call(grpc_server *server, requested_call *rc); +static void shutdown_channel(channel_data *chand, int send_goaway, + int send_disconnect); +static void maybe_finish_shutdown(grpc_server *server); static int call_list_join(call_data **root, call_data *call, call_list list) { GPR_ASSERT(!call->root[list]); @@ -256,29 +265,32 @@ static void server_ref(grpc_server *server) { gpr_ref(&server->internal_refcount); } -static void server_unref(grpc_server *server) { +static void server_delete(grpc_server *server) { registered_method *rm; size_t i; + grpc_channel_args_destroy(server->channel_args); + gpr_mu_destroy(&server->mu); + gpr_free(server->channel_filters); + requested_call_array_destroy(&server->requested_calls); + while ((rm = server->registered_methods) != NULL) { + server->registered_methods = rm->next; + gpr_free(rm->method); + gpr_free(rm->host); + requested_call_array_destroy(&rm->requested); + gpr_free(rm); + } + for (i = 0; i < server->cq_count; i++) { + GRPC_CQ_INTERNAL_UNREF(server->cqs[i], "server"); + } + gpr_free(server->cqs); + gpr_free(server->pollsets); + gpr_free(server->shutdown_tags); + gpr_free(server); +} + +static void server_unref(grpc_server *server) { if (gpr_unref(&server->internal_refcount)) { - grpc_channel_args_destroy(server->channel_args); - gpr_mu_destroy(&server->mu); - gpr_cv_destroy(&server->cv); - gpr_free(server->channel_filters); - requested_call_array_destroy(&server->requested_calls); - while ((rm = server->registered_methods) != NULL) { - server->registered_methods = rm->next; - gpr_free(rm->method); - gpr_free(rm->host); - requested_call_array_destroy(&rm->requested); - gpr_free(rm); - } - for (i = 0; i < server->cq_count; i++) { - grpc_cq_internal_unref(server->cqs[i]); - } - gpr_free(server->cqs); - gpr_free(server->pollsets); - gpr_free(server->shutdown_tags); - gpr_free(server); + server_delete(server); } } @@ -295,7 +307,7 @@ static void orphan_channel(channel_data *chand) { static void finish_destroy_channel(void *cd, int success) { channel_data *chand = cd; grpc_server *server = chand->server; - grpc_channel_internal_unref(chand->channel); + GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "server"); server_unref(server); } @@ -304,6 +316,7 @@ static void destroy_channel(channel_data *chand) { GPR_ASSERT(chand->server != NULL); orphan_channel(chand); server_ref(chand->server); + maybe_finish_shutdown(chand->server); grpc_iomgr_add_callback(finish_destroy_channel, chand); } @@ -371,6 +384,42 @@ static void kill_zombie(void *elem, int success) { grpc_call_destroy(grpc_call_from_top_element(elem)); } +static int num_listeners(grpc_server *server) { + listener *l; + int n = 0; + for (l = server->listeners; l; l = l->next) { + n++; + } + return n; +} + +static void maybe_finish_shutdown(grpc_server *server) { + size_t i; + if (!server->shutdown || server->shutdown_published) { + return; + } + if (server->lists[ALL_CALLS] != NULL) { + gpr_log(GPR_DEBUG, + "Waiting for all calls to finish before destroying server"); + return; + } + if (server->root_channel_data.next != &server->root_channel_data) { + gpr_log(GPR_DEBUG, + "Waiting for all channels to close before destroying server"); + return; + } + if (server->listeners_destroyed < num_listeners(server)) { + gpr_log(GPR_DEBUG, "Waiting for all listeners to be destroyed (@ %d/%d)", + server->listeners_destroyed, num_listeners(server)); + return; + } + server->shutdown_published = 1; + for (i = 0; i < server->num_shutdown_tags; i++) { + grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, + NULL, 1); + } +} + static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { grpc_call_element *elem = user_data; channel_data *chand = elem->channel_data; @@ -385,6 +434,14 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { return md; } +static void decrement_call_count(channel_data *chand) { + chand->num_calls--; + if (0 == chand->num_calls && chand->server->shutdown) { + shutdown_channel(chand, 0, 1); + } + maybe_finish_shutdown(chand->server); +} + static void server_on_recv(void *ptr, int success) { grpc_call_element *elem = ptr; call_data *calld = elem->call_data; @@ -430,6 +487,9 @@ static void server_on_recv(void *ptr, int success) { calld->state = ZOMBIED; grpc_iomgr_add_callback(kill_zombie, elem); } + if (call_list_remove(calld, ALL_CALLS)) { + decrement_call_count(chand); + } gpr_mu_unlock(&chand->server->mu); break; } @@ -489,20 +549,47 @@ static void channel_op(grpc_channel_element *elem, } } -static void finish_shutdown_channel(void *cd, int success) { - channel_data *chand = cd; +typedef struct { + channel_data *chand; + int send_goaway; + int send_disconnect; +} shutdown_channel_args; + +static void finish_shutdown_channel(void *p, int success) { + shutdown_channel_args *sca = p; grpc_channel_op op; - op.type = GRPC_CHANNEL_DISCONNECT; - op.dir = GRPC_CALL_DOWN; - channel_op(grpc_channel_stack_element( - grpc_channel_get_channel_stack(chand->channel), 0), - NULL, &op); - grpc_channel_internal_unref(chand->channel); + + if (sca->send_goaway) { + op.type = GRPC_CHANNEL_GOAWAY; + op.dir = GRPC_CALL_DOWN; + op.data.goaway.status = GRPC_STATUS_OK; + op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown"); + channel_op(grpc_channel_stack_element( + grpc_channel_get_channel_stack(sca->chand->channel), 0), + NULL, &op); + } + if (sca->send_disconnect) { + op.type = GRPC_CHANNEL_DISCONNECT; + op.dir = GRPC_CALL_DOWN; + channel_op(grpc_channel_stack_element( + grpc_channel_get_channel_stack(sca->chand->channel), 0), + NULL, &op); + } + GRPC_CHANNEL_INTERNAL_UNREF(sca->chand->channel, "shutdown"); + + gpr_free(sca); } -static void shutdown_channel(channel_data *chand) { - grpc_channel_internal_ref(chand->channel); - grpc_iomgr_add_callback(finish_shutdown_channel, chand); +static void shutdown_channel(channel_data *chand, int send_goaway, + int send_disconnect) { + shutdown_channel_args *sca; + gpr_log(GPR_DEBUG, "shutdown_channel: %p %d %d", chand, send_goaway, send_disconnect); + GRPC_CHANNEL_INTERNAL_REF(chand->channel, "shutdown"); + sca = gpr_malloc(sizeof(shutdown_channel_args)); + sca->chand = chand; + sca->send_goaway = send_goaway; + sca->send_disconnect = send_disconnect; + grpc_iomgr_add_callback(finish_shutdown_channel, sca); } static void init_call_elem(grpc_call_element *elem, @@ -516,6 +603,7 @@ static void init_call_elem(grpc_call_element *elem, gpr_mu_lock(&chand->server->mu); call_list_join(&chand->server->lists[ALL_CALLS], calld, ALL_CALLS); + chand->num_calls++; gpr_mu_unlock(&chand->server->mu); server_ref(chand->server); @@ -526,19 +614,15 @@ static void init_call_elem(grpc_call_element *elem, static void destroy_call_elem(grpc_call_element *elem) { channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; - size_t i, j; + int removed[CALL_LIST_COUNT]; + size_t i; gpr_mu_lock(&chand->server->mu); for (i = 0; i < CALL_LIST_COUNT; i++) { - call_list_remove(elem->call_data, i); + removed[i] = call_list_remove(elem->call_data, i); } - if (chand->server->shutdown && chand->server->lists[ALL_CALLS] == NULL) { - for (i = 0; i < chand->server->num_shutdown_tags; i++) { - for (j = 0; j < chand->server->cq_count; j++) { - grpc_cq_end_op(chand->server->cqs[j], chand->server->shutdown_tags[i], - NULL, 1); - } - } + if (removed[ALL_CALLS]) { + decrement_call_count(chand); } gpr_mu_unlock(&chand->server->mu); @@ -560,6 +644,7 @@ static void init_channel_elem(grpc_channel_element *elem, GPR_ASSERT(is_first); GPR_ASSERT(!is_last); chand->server = NULL; + chand->num_calls = 0; chand->channel = NULL; chand->path_key = grpc_mdstr_from_string(metadata_context, ":path"); chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority"); @@ -586,6 +671,7 @@ static void destroy_channel_elem(grpc_channel_element *elem) { chand->next->prev = chand->prev; chand->prev->next = chand->next; chand->next = chand->prev = chand; + maybe_finish_shutdown(chand->server); gpr_mu_unlock(&chand->server->mu); grpc_mdstr_unref(chand->path_key); grpc_mdstr_unref(chand->authority_key); @@ -594,9 +680,15 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } static const grpc_channel_filter server_surface_filter = { - server_start_transport_op, channel_op, sizeof(call_data), init_call_elem, - destroy_call_elem, sizeof(channel_data), init_channel_elem, - destroy_channel_elem, "server", + server_start_transport_op, + channel_op, + sizeof(call_data), + init_call_elem, + destroy_call_elem, + sizeof(channel_data), + init_channel_elem, + destroy_channel_elem, + "server", }; void grpc_server_register_completion_queue(grpc_server *server, @@ -605,7 +697,7 @@ void grpc_server_register_completion_queue(grpc_server *server, for (i = 0; i < server->cq_count; i++) { if (server->cqs[i] == cq) return; } - grpc_cq_internal_ref(cq); + GRPC_CQ_INTERNAL_REF(cq, "server"); n = server->cq_count++; server->cqs = gpr_realloc(server->cqs, server->cq_count * sizeof(grpc_completion_queue *)); @@ -625,7 +717,6 @@ grpc_server *grpc_server_create_from_filters(grpc_channel_filter **filters, memset(server, 0, sizeof(grpc_server)); gpr_mu_init(&server->mu); - gpr_cv_init(&server->cv); /* decremented by grpc_server_destroy */ gpr_ref_init(&server->internal_refcount, 1); @@ -665,7 +756,8 @@ void *grpc_server_register_method(grpc_server *server, const char *method, const char *host) { registered_method *m; if (!method) { - gpr_log(GPR_ERROR, "grpc_server_register_method method string cannot be NULL"); + gpr_log(GPR_ERROR, + "grpc_server_register_method method string cannot be NULL"); return NULL; } for (m = server->registered_methods; m; m = m->next) { @@ -784,55 +876,32 @@ grpc_transport_setup_result grpc_server_setup_transport( return result; } -static int num_listeners(grpc_server *server) { - listener *l; - int n = 0; - for (l = server->listeners; l; l = l->next) { - n++; - } - return n; -} - -static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag, - void *shutdown_tag) { +void grpc_server_shutdown_and_notify(grpc_server *server, + grpc_completion_queue *cq, void *tag) { listener *l; requested_call_array requested_calls; - channel_data **channels; channel_data *c; - size_t nchannels; - size_t i, j; - grpc_channel_op op; - grpc_channel_element *elem; + size_t i; registered_method *rm; + shutdown_tag *sdt; /* lock, and gather up some stuff to do */ gpr_mu_lock(&server->mu); - if (have_shutdown_tag) { - for (i = 0; i < server->cq_count; i++) { - grpc_cq_begin_op(server->cqs[i], NULL); - } - server->shutdown_tags = - gpr_realloc(server->shutdown_tags, - sizeof(void *) * (server->num_shutdown_tags + 1)); - server->shutdown_tags[server->num_shutdown_tags++] = shutdown_tag; - } + grpc_cq_begin_op(cq, NULL); + server->shutdown_tags = + gpr_realloc(server->shutdown_tags, + sizeof(shutdown_tag) * (server->num_shutdown_tags + 1)); + sdt = &server->shutdown_tags[server->num_shutdown_tags++]; + sdt->tag = tag; + sdt->cq = cq; if (server->shutdown) { gpr_mu_unlock(&server->mu); return; } - nchannels = 0; for (c = server->root_channel_data.next; c != &server->root_channel_data; c = c->next) { - nchannels++; - } - channels = gpr_malloc(sizeof(channel_data *) * nchannels); - i = 0; - for (c = server->root_channel_data.next; c != &server->root_channel_data; - c = c->next) { - grpc_channel_internal_ref(c->channel); - channels[i] = c; - i++; + shutdown_channel(c, 1, c->num_calls == 0); } /* collect all unregistered then registered calls */ @@ -856,30 +925,9 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag, } server->shutdown = 1; - if (server->lists[ALL_CALLS] == NULL) { - for (i = 0; i < server->num_shutdown_tags; i++) { - for (j = 0; j < server->cq_count; j++) { - grpc_cq_end_op(server->cqs[j], server->shutdown_tags[i], NULL, 1); - } - } - } + maybe_finish_shutdown(server); gpr_mu_unlock(&server->mu); - for (i = 0; i < nchannels; i++) { - c = channels[i]; - elem = grpc_channel_stack_element( - grpc_channel_get_channel_stack(c->channel), 0); - - op.type = GRPC_CHANNEL_GOAWAY; - op.dir = GRPC_CALL_DOWN; - op.data.goaway.status = GRPC_STATUS_OK; - op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown"); - elem->filter->channel_op(elem, NULL, &op); - - grpc_channel_internal_unref(c->channel); - } - gpr_free(channels); - /* terminate all the requested calls */ for (i = 0; i < requested_calls.count; i++) { fail_call(server, &requested_calls.calls[i]); @@ -892,65 +940,70 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag, } } -void grpc_server_shutdown(grpc_server *server) { - shutdown_internal(server, 0, NULL); -} - -void grpc_server_shutdown_and_notify(grpc_server *server, void *tag) { - shutdown_internal(server, 1, tag); -} - void grpc_server_listener_destroy_done(void *s) { grpc_server *server = s; gpr_mu_lock(&server->mu); server->listeners_destroyed++; - gpr_cv_signal(&server->cv); + maybe_finish_shutdown(server); gpr_mu_unlock(&server->mu); } -void grpc_server_destroy(grpc_server *server) { - channel_data *c; - listener *l; - size_t i; +void grpc_server_cancel_all_calls(grpc_server *server) { call_data *calld; + grpc_call **calls; + size_t call_count; + size_t call_capacity; + int is_first = 1; + size_t i; gpr_mu_lock(&server->mu); - if (!server->shutdown) { + + GPR_ASSERT(server->shutdown); + + if (!server->lists[ALL_CALLS]) { gpr_mu_unlock(&server->mu); - grpc_server_shutdown(server); - gpr_mu_lock(&server->mu); + return; } - while (server->listeners_destroyed != num_listeners(server)) { - for (i = 0; i < server->cq_count; i++) { - gpr_mu_unlock(&server->mu); - grpc_cq_hack_spin_pollset(server->cqs[i]); - gpr_mu_lock(&server->mu); + call_capacity = 8; + call_count = 0; + calls = gpr_malloc(sizeof(grpc_call *) * call_capacity); + + for (calld = server->lists[ALL_CALLS]; + calld != server->lists[ALL_CALLS] || is_first; + calld = calld->links[ALL_CALLS].next) { + if (call_count == call_capacity) { + call_capacity *= 2; + calls = gpr_realloc(calls, sizeof(grpc_call *) * call_capacity); } + calls[call_count++] = calld->call; + GRPC_CALL_INTERNAL_REF(calld->call, "cancel_all"); + is_first = 0; + } - gpr_cv_wait(&server->cv, &server->mu, - gpr_time_add(gpr_now(), gpr_time_from_millis(100))); + gpr_mu_unlock(&server->mu); + + for (i = 0; i < call_count; i++) { + grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE, + "Unavailable"); + GRPC_CALL_INTERNAL_UNREF(calls[i], "cancel_all", 1); } + gpr_free(calls); +} + +void grpc_server_destroy(grpc_server *server) { + listener *l; + + gpr_mu_lock(&server->mu); + GPR_ASSERT(server->shutdown); + GPR_ASSERT(server->listeners_destroyed == num_listeners(server)); + while (server->listeners) { l = server->listeners; server->listeners = l->next; gpr_free(l); } - - while ((calld = call_list_remove_head(&server->lists[PENDING_START], - PENDING_START)) != NULL) { - gpr_log(GPR_DEBUG, "server destroys call %p", calld->call); - calld->state = ZOMBIED; - grpc_iomgr_add_callback( - kill_zombie, - grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0)); - } - - for (c = server->root_channel_data.next; c != &server->root_channel_data; - c = c->next) { - shutdown_channel(c); - } gpr_mu_unlock(&server->mu); server_unref(server); @@ -1139,4 +1192,3 @@ int grpc_server_has_open_connections(grpc_server *server) { gpr_mu_unlock(&server->mu); return r; } - diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 9dc5f23389..07c5b0be0d 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -614,14 +614,19 @@ static void destroy_transport(grpc_transport *gt) { unref_transport(t); } +static void close_transport_locked(transport *t) { + if (!t->closed) { + t->closed = 1; + if (t->ep) { + grpc_endpoint_shutdown(t->ep); + } + } +} + static void close_transport(grpc_transport *gt) { transport *t = (transport *)gt; gpr_mu_lock(&t->mu); - GPR_ASSERT(!t->closed); - t->closed = 1; - if (t->ep) { - grpc_endpoint_shutdown(t->ep); - } + close_transport_locked(t); gpr_mu_unlock(&t->mu); } @@ -1313,6 +1318,7 @@ static void drop_connection(transport *t) { if (t->error_state == ERROR_STATE_NONE) { t->error_state = ERROR_STATE_SEEN; } + close_transport_locked(t); end_all_the_calls(t); } diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c index d9a1319c42..e0dca22005 100644 --- a/src/core/transport/transport.c +++ b/src/core/transport/transport.c @@ -86,6 +86,16 @@ void grpc_transport_setup_initiate(grpc_transport_setup *setup) { setup->vtable->initiate(setup); } +void grpc_transport_setup_add_interested_party(grpc_transport_setup *setup, + grpc_pollset *pollset) { + setup->vtable->add_interested_party(setup, pollset); +} + +void grpc_transport_setup_del_interested_party(grpc_transport_setup *setup, + grpc_pollset *pollset) { + setup->vtable->del_interested_party(setup, pollset); +} + void grpc_transport_op_finish_with_failure(grpc_transport_op *op) { if (op->send_ops) { op->on_done_send(op->send_user_data, 0); diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index 6f8d39e352..521d74c4c4 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -37,6 +37,7 @@ #include <stddef.h> #include "src/core/iomgr/pollset.h" +#include "src/core/iomgr/pollset_set.h" #include "src/core/transport/stream_op.h" #include "src/core/channel/context.h" @@ -195,6 +196,10 @@ typedef struct grpc_transport_setup_vtable grpc_transport_setup_vtable; struct grpc_transport_setup_vtable { void (*initiate)(grpc_transport_setup *setup); + void (*add_interested_party)(grpc_transport_setup *setup, + grpc_pollset *pollset); + void (*del_interested_party)(grpc_transport_setup *setup, + grpc_pollset *pollset); void (*cancel)(grpc_transport_setup *setup); }; @@ -211,6 +216,12 @@ struct grpc_transport_setup { This *may* be implemented as a no-op if the setup process monitors something continuously. */ void grpc_transport_setup_initiate(grpc_transport_setup *setup); + +void grpc_transport_setup_add_interested_party(grpc_transport_setup *setup, + grpc_pollset *pollset); +void grpc_transport_setup_del_interested_party(grpc_transport_setup *setup, + grpc_pollset *pollset); + /* Cancel transport setup. After this returns, no new transports should be created, and all pending transport setup callbacks should be completed. After this call completes, setup should be considered invalid (this can be |