diff options
author | Craig Tiller <ctiller@google.com> | 2015-09-18 17:29:00 -0700 |
---|---|---|
committer | Craig Tiller <ctiller@google.com> | 2015-09-18 17:29:00 -0700 |
commit | d1bec03fa148344b8eac2b59517252d86e4ca858 (patch) | |
tree | f359e48f9151ab7ceff72cd624ad6c7a59e4d304 /src/core | |
parent | 33825118df7157219cec15382beb006d3462ad96 (diff) |
Call list progress
Diffstat (limited to 'src/core')
78 files changed, 1340 insertions, 1340 deletions
diff --git a/src/core/census/grpc_filter.c b/src/core/census/grpc_filter.c index d98162ec9a..d66f29228b 100644 --- a/src/core/census/grpc_filter.c +++ b/src/core/census/grpc_filter.c @@ -89,19 +89,21 @@ static void client_mutate_op(grpc_call_element* elem, } static void client_start_transport_op(grpc_call_element* elem, - grpc_transport_stream_op* op) { + grpc_transport_stream_op* op, + grpc_call_list* call_list) { client_mutate_op(elem, op); - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); } -static void server_on_done_recv(void* ptr, int success) { +static void server_on_done_recv(void* ptr, int success, + grpc_call_list* call_list) { grpc_call_element* elem = ptr; call_data* calld = elem->call_data; channel_data* chand = elem->channel_data; if (success) { extract_and_annotate_method_tag(calld->recv_ops, calld, chand); } - calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); + calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success, call_list); } static void server_mutate_op(grpc_call_element* elem, @@ -116,11 +118,12 @@ static void server_mutate_op(grpc_call_element* elem, } static void server_start_transport_op(grpc_call_element* elem, - grpc_transport_stream_op* op) { + grpc_transport_stream_op* op, + grpc_call_list* call_list) { call_data* calld = elem->call_data; GPR_ASSERT((calld->op_id.upper != 0) || (calld->op_id.lower != 0)); server_mutate_op(elem, op); - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); } static void client_init_call_elem(grpc_call_element* elem, @@ -132,7 +135,8 @@ static void client_init_call_elem(grpc_call_element* elem, if (initial_op) client_mutate_op(elem, initial_op); } -static void client_destroy_call_elem(grpc_call_element* elem) { +static void client_destroy_call_elem(grpc_call_element* elem, + grpc_call_list* call_list) { call_data* d = elem->call_data; GPR_ASSERT(d != NULL); /* TODO(hongyu): record rpc client stats and census_rpc_end_op here */ @@ -149,7 +153,8 @@ static void server_init_call_elem(grpc_call_element* elem, if (initial_op) server_mutate_op(elem, initial_op); } -static void server_destroy_call_elem(grpc_call_element* elem) { +static void server_destroy_call_elem(grpc_call_element* elem, + grpc_call_list* call_list) { call_data* d = elem->call_data; GPR_ASSERT(d != NULL); /* TODO(hongyu): record rpc server stats and census_tracing_end_op here */ @@ -157,13 +162,15 @@ static void server_destroy_call_elem(grpc_call_element* elem) { static void init_channel_elem(grpc_channel_element* elem, grpc_channel* master, const grpc_channel_args* args, grpc_mdctx* mdctx, - int is_first, int is_last) { + int is_first, int is_last, + grpc_call_list* call_list) { channel_data* chand = elem->channel_data; GPR_ASSERT(chand != NULL); chand->path_str = grpc_mdstr_from_string(mdctx, ":path", 0); } -static void destroy_channel_elem(grpc_channel_element* elem) { +static void destroy_channel_elem(grpc_channel_element* elem, + grpc_call_list* call_list) { channel_data* chand = elem->channel_data; GPR_ASSERT(chand != NULL); if (chand->path_str != NULL) { diff --git a/src/core/channel/channel_stack.c b/src/core/channel/channel_stack.c index 4eb5df5de3..c767a87b20 100644 --- a/src/core/channel/channel_stack.c +++ b/src/core/channel/channel_stack.c @@ -105,7 +105,8 @@ void grpc_channel_stack_init(const grpc_channel_filter **filters, size_t filter_count, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *metadata_context, - grpc_channel_stack *stack) { + grpc_channel_stack *stack, + grpc_call_list *call_list) { size_t call_size = ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call_stack)) + ROUND_UP_TO_ALIGNMENT_SIZE(filter_count * sizeof(grpc_call_element)); @@ -125,7 +126,7 @@ void grpc_channel_stack_init(const grpc_channel_filter **filters, elems[i].channel_data = user_data; elems[i].filter->init_channel_elem(&elems[i], master, args, metadata_context, i == 0, - i == (filter_count - 1)); + i == (filter_count - 1), call_list); user_data += ROUND_UP_TO_ALIGNMENT_SIZE(filters[i]->sizeof_channel_data); call_size += ROUND_UP_TO_ALIGNMENT_SIZE(filters[i]->sizeof_call_data); } @@ -137,14 +138,15 @@ void grpc_channel_stack_init(const grpc_channel_filter **filters, stack->call_stack_size = call_size; } -void grpc_channel_stack_destroy(grpc_channel_stack *stack) { +void grpc_channel_stack_destroy(grpc_channel_stack *stack, + grpc_call_list *call_list) { grpc_channel_element *channel_elems = CHANNEL_ELEMS_FROM_STACK(stack); size_t count = stack->count; size_t i; /* destroy per-filter data */ for (i = 0; i < count; i++) { - channel_elems[i].filter->destroy_channel_elem(&channel_elems[i]); + channel_elems[i].filter->destroy_channel_elem(&channel_elems[i], call_list); } } @@ -175,30 +177,34 @@ void grpc_call_stack_init(grpc_channel_stack *channel_stack, } } -void grpc_call_stack_destroy(grpc_call_stack *stack) { +void grpc_call_stack_destroy(grpc_call_stack *stack, + grpc_call_list *call_list) { grpc_call_element *elems = CALL_ELEMS_FROM_STACK(stack); size_t count = stack->count; size_t i; /* destroy per-filter data */ for (i = 0; i < count; i++) { - elems[i].filter->destroy_call_elem(&elems[i]); + elems[i].filter->destroy_call_elem(&elems[i], call_list); } } -void grpc_call_next_op(grpc_call_element *elem, grpc_transport_stream_op *op) { +void grpc_call_next_op(grpc_call_element *elem, grpc_transport_stream_op *op, + grpc_call_list *call_list) { grpc_call_element *next_elem = elem + 1; - next_elem->filter->start_transport_stream_op(next_elem, op); + next_elem->filter->start_transport_stream_op(next_elem, op, call_list); } -char *grpc_call_next_get_peer(grpc_call_element *elem) { +char *grpc_call_next_get_peer(grpc_call_element *elem, + grpc_call_list *call_list) { grpc_call_element *next_elem = elem + 1; - return next_elem->filter->get_peer(next_elem); + return next_elem->filter->get_peer(next_elem, call_list); } -void grpc_channel_next_op(grpc_channel_element *elem, grpc_transport_op *op) { +void grpc_channel_next_op(grpc_channel_element *elem, grpc_transport_op *op, + grpc_call_list *call_list) { grpc_channel_element *next_elem = elem + 1; - next_elem->filter->start_transport_op(next_elem, op); + next_elem->filter->start_transport_op(next_elem, op, call_list); } grpc_channel_stack *grpc_channel_stack_from_top_element( @@ -212,9 +218,10 @@ grpc_call_stack *grpc_call_stack_from_top_element(grpc_call_element *elem) { sizeof(grpc_call_stack))); } -void grpc_call_element_send_cancel(grpc_call_element *cur_elem) { +void grpc_call_element_send_cancel(grpc_call_element *cur_elem, + grpc_call_list *call_list) { grpc_transport_stream_op op; memset(&op, 0, sizeof(op)); op.cancel_with_status = GRPC_STATUS_CANCELLED; - grpc_call_next_op(cur_elem, &op); + grpc_call_next_op(cur_elem, &op, call_list); } diff --git a/src/core/channel/channel_stack.h b/src/core/channel/channel_stack.h index 4a608b956e..190247b3f1 100644 --- a/src/core/channel/channel_stack.h +++ b/src/core/channel/channel_stack.h @@ -65,11 +65,13 @@ typedef struct { /* Called to eg. send/receive data on a call. See grpc_call_next_op on how to call the next element in the stack */ void (*start_transport_stream_op)(grpc_call_element *elem, - grpc_transport_stream_op *op); + grpc_transport_stream_op *op, + grpc_call_list *call_list); /* Called to handle channel level operations - e.g. new calls, or transport closure. See grpc_channel_next_op on how to call the next element in the stack */ - void (*start_transport_op)(grpc_channel_element *elem, grpc_transport_op *op); + void (*start_transport_op)(grpc_channel_element *elem, grpc_transport_op *op, + grpc_call_list *call_list); /* sizeof(per call data) */ size_t sizeof_call_data; @@ -86,7 +88,7 @@ typedef struct { grpc_transport_stream_op *initial_op); /* Destroy per call data. The filter does not need to do any chaining */ - void (*destroy_call_elem)(grpc_call_element *elem); + void (*destroy_call_elem)(grpc_call_element *elem, grpc_call_list *call_list); /* sizeof(per channel data) */ size_t sizeof_channel_data; @@ -99,13 +101,14 @@ typedef struct { void (*init_channel_elem)(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *metadata_context, int is_first, - int is_last); + int is_last, grpc_call_list *call_list); /* Destroy per channel data. The filter does not need to do any chaining */ - void (*destroy_channel_elem)(grpc_channel_element *elem); + void (*destroy_channel_elem)(grpc_channel_element *elem, + grpc_call_list *call_list); /* Implement grpc_call_get_peer() */ - char *(*get_peer)(grpc_call_element *elem); + char *(*get_peer)(grpc_call_element *elem, grpc_call_list *call_list); /* The name of this filter */ const char *name; @@ -157,9 +160,11 @@ void grpc_channel_stack_init(const grpc_channel_filter **filters, size_t filter_count, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *metadata_context, - grpc_channel_stack *stack); + grpc_channel_stack *stack, + grpc_call_list *call_list); /* Destroy a channel stack */ -void grpc_channel_stack_destroy(grpc_channel_stack *stack); +void grpc_channel_stack_destroy(grpc_channel_stack *stack, + grpc_call_list *call_list); /* Initialize a call stack given a channel stack. transport_server_data is expected to be NULL on a client, or an opaque transport owned pointer on the @@ -169,15 +174,18 @@ void grpc_call_stack_init(grpc_channel_stack *channel_stack, grpc_transport_stream_op *initial_op, grpc_call_stack *call_stack); /* Destroy a call stack */ -void grpc_call_stack_destroy(grpc_call_stack *stack); +void grpc_call_stack_destroy(grpc_call_stack *stack, grpc_call_list *call_list); /* Call the next operation in a call stack */ -void grpc_call_next_op(grpc_call_element *elem, grpc_transport_stream_op *op); +void grpc_call_next_op(grpc_call_element *elem, grpc_transport_stream_op *op, + grpc_call_list *call_list); /* Call the next operation (depending on call directionality) in a channel stack */ -void grpc_channel_next_op(grpc_channel_element *elem, grpc_transport_op *op); +void grpc_channel_next_op(grpc_channel_element *elem, grpc_transport_op *op, + grpc_call_list *call_list); /* Pass through a request to get_peer to the next child element */ -char *grpc_call_next_get_peer(grpc_call_element *elem); +char *grpc_call_next_get_peer(grpc_call_element *elem, + grpc_call_list *call_list); /* Given the top element of a channel stack, get the channel stack itself */ grpc_channel_stack *grpc_channel_stack_from_top_element( @@ -188,7 +196,8 @@ grpc_call_stack *grpc_call_stack_from_top_element(grpc_call_element *elem); void grpc_call_log_op(char *file, int line, gpr_log_severity severity, grpc_call_element *elem, grpc_transport_stream_op *op); -void grpc_call_element_send_cancel(grpc_call_element *cur_elem); +void grpc_call_element_send_cancel(grpc_call_element *cur_elem, + grpc_call_list *call_list); extern int grpc_trace_channel; diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index 588b9c36ee..6618336e93 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -128,12 +128,13 @@ static grpc_closure *merge_into_waiting_op(grpc_call_element *elem, GRPC_MUST_USE_RESULT; static void handle_op_after_cancellation(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; if (op->send_ops) { grpc_stream_ops_unref_owned_objects(op->send_ops->ops, op->send_ops->nops); - op->on_done_send->cb(op->on_done_send->cb_arg, 0); + op->on_done_send->cb(op->on_done_send->cb_arg, 0, call_list); } if (op->recv_ops) { char status[GPR_LTOA_MIN_BUFSIZE]; @@ -152,10 +153,10 @@ static void handle_op_after_cancellation(grpc_call_element *elem, mdb.deadline = gpr_inf_future(GPR_CLOCK_REALTIME); grpc_sopb_add_metadata(op->recv_ops, mdb); *op->recv_state = GRPC_STREAM_CLOSED; - op->on_done_recv->cb(op->on_done_recv->cb_arg, 1); + op->on_done_recv->cb(op->on_done_recv->cb_arg, 1, call_list); } if (op->on_consumed) { - op->on_consumed->cb(op->on_consumed->cb_arg, 0); + op->on_consumed->cb(op->on_consumed->cb_arg, 0, call_list); } } @@ -166,12 +167,14 @@ typedef struct { static void perform_transport_stream_op(grpc_call_element *elem, grpc_transport_stream_op *op, - int continuation); + int continuation, + grpc_call_list *call_list); -static void continue_with_pick(void *arg, int iomgr_success) { +static void continue_with_pick(void *arg, int iomgr_success, + grpc_call_list *call_list) { waiting_call *wc = arg; call_data *calld = wc->elem->call_data; - perform_transport_stream_op(wc->elem, &calld->waiting_op, 1); + perform_transport_stream_op(wc->elem, &calld->waiting_op, 1, call_list); gpr_free(wc); } @@ -193,7 +196,8 @@ static int is_empty(void *p, int len) { return 1; } -static void started_call(void *arg, int iomgr_success) { +static void started_call(void *arg, int iomgr_success, + grpc_call_list *call_list) { call_data *calld = arg; grpc_transport_stream_op op; int have_waiting; @@ -203,7 +207,7 @@ static void started_call(void *arg, int iomgr_success) { memset(&op, 0, sizeof(op)); op.cancel_with_status = GRPC_STATUS_CANCELLED; gpr_mu_unlock(&calld->mu_state); - grpc_subchannel_call_process_op(calld->subchannel_call, &op); + grpc_subchannel_call_process_op(calld->subchannel_call, &op, call_list); } else if (calld->state == CALL_WAITING_FOR_CALL) { have_waiting = !is_empty(&calld->waiting_op, sizeof(calld->waiting_op)); if (calld->subchannel_call != NULL) { @@ -211,13 +215,14 @@ static void started_call(void *arg, int iomgr_success) { gpr_mu_unlock(&calld->mu_state); if (have_waiting) { grpc_subchannel_call_process_op(calld->subchannel_call, - &calld->waiting_op); + &calld->waiting_op, call_list); } } else { calld->state = CALL_CANCELLED; gpr_mu_unlock(&calld->mu_state); if (have_waiting) { - handle_op_after_cancellation(calld->elem, &calld->waiting_op); + handle_op_after_cancellation(calld->elem, &calld->waiting_op, + call_list); } } } else { @@ -226,20 +231,20 @@ static void started_call(void *arg, int iomgr_success) { } } -static void picked_target(void *arg, int iomgr_success) { +static void picked_target(void *arg, int iomgr_success, + grpc_call_list *call_list) { call_data *calld = arg; grpc_pollset *pollset; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; if (calld->picked_channel == NULL) { /* treat this like a cancellation */ calld->waiting_op.cancel_with_status = GRPC_STATUS_UNAVAILABLE; - perform_transport_stream_op(calld->elem, &calld->waiting_op, 1); + perform_transport_stream_op(calld->elem, &calld->waiting_op, 1, call_list); } else { gpr_mu_lock(&calld->mu_state); if (calld->state == CALL_CANCELLED) { gpr_mu_unlock(&calld->mu_state); - handle_op_after_cancellation(calld->elem, &calld->waiting_op); + handle_op_after_cancellation(calld->elem, &calld->waiting_op, call_list); } else { GPR_ASSERT(calld->state == CALL_WAITING_FOR_PICK); calld->state = CALL_WAITING_FOR_CALL; @@ -248,10 +253,9 @@ static void picked_target(void *arg, int iomgr_success) { grpc_closure_init(&calld->async_setup_task, started_call, calld); grpc_subchannel_create_call(calld->picked_channel, pollset, &calld->subchannel_call, - &calld->async_setup_task, &call_list); + &calld->async_setup_task, call_list); } } - grpc_call_list_run(call_list); } static grpc_closure *merge_into_waiting_op(grpc_call_element *elem, @@ -283,7 +287,7 @@ static grpc_closure *merge_into_waiting_op(grpc_call_element *elem, return consumed_op; } -static char *cc_get_peer(grpc_call_element *elem) { +static char *cc_get_peer(grpc_call_element *elem, grpc_call_list *call_list) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; grpc_subchannel_call *subchannel_call; @@ -294,8 +298,8 @@ static char *cc_get_peer(grpc_call_element *elem) { subchannel_call = calld->subchannel_call; GRPC_SUBCHANNEL_CALL_REF(subchannel_call, "get_peer"); gpr_mu_unlock(&calld->mu_state); - result = grpc_subchannel_call_get_peer(subchannel_call); - GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call, "get_peer"); + result = grpc_subchannel_call_get_peer(subchannel_call, call_list); + GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call, "get_peer", call_list); return result; } else { gpr_mu_unlock(&calld->mu_state); @@ -305,13 +309,13 @@ static char *cc_get_peer(grpc_call_element *elem) { static void perform_transport_stream_op(grpc_call_element *elem, grpc_transport_stream_op *op, - int continuation) { + int continuation, + grpc_call_list *call_list) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; grpc_subchannel_call *subchannel_call; grpc_lb_policy *lb_policy; grpc_transport_stream_op op2; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; GPR_ASSERT(elem->filter == &grpc_client_channel_filter); GRPC_CALL_LOG_OP(GPR_INFO, elem, op); @@ -321,15 +325,15 @@ static void perform_transport_stream_op(grpc_call_element *elem, GPR_ASSERT(!continuation); subchannel_call = calld->subchannel_call; gpr_mu_unlock(&calld->mu_state); - grpc_subchannel_call_process_op(subchannel_call, op); + grpc_subchannel_call_process_op(subchannel_call, op, call_list); break; case CALL_CANCELLED: gpr_mu_unlock(&calld->mu_state); - handle_op_after_cancellation(elem, op); + handle_op_after_cancellation(elem, op, call_list); break; case CALL_WAITING_FOR_SEND: GPR_ASSERT(!continuation); - grpc_call_list_add(&call_list, merge_into_waiting_op(elem, op), 1); + grpc_call_list_add(call_list, merge_into_waiting_op(elem, op), 1); if (!calld->waiting_op.send_ops && calld->waiting_op.cancel_with_status == GRPC_STATUS_OK) { gpr_mu_unlock(&calld->mu_state); @@ -355,10 +359,10 @@ static void perform_transport_stream_op(grpc_call_element *elem, op2.on_consumed = NULL; } gpr_mu_unlock(&calld->mu_state); - handle_op_after_cancellation(elem, op); - handle_op_after_cancellation(elem, &op2); + handle_op_after_cancellation(elem, op, call_list); + handle_op_after_cancellation(elem, &op2, call_list); } else { - grpc_call_list_add(&call_list, merge_into_waiting_op(elem, op), 1); + grpc_call_list_add(call_list, merge_into_waiting_op(elem, op), 1); gpr_mu_unlock(&calld->mu_state); } break; @@ -368,7 +372,7 @@ static void perform_transport_stream_op(grpc_call_element *elem, if (op->cancel_with_status != GRPC_STATUS_OK) { calld->state = CALL_CANCELLED; gpr_mu_unlock(&calld->mu_state); - handle_op_after_cancellation(elem, op); + handle_op_after_cancellation(elem, op, call_list); } else { calld->waiting_op = *op; @@ -398,9 +402,9 @@ static void perform_transport_stream_op(grpc_call_element *elem, grpc_closure_init(&calld->async_setup_task, picked_target, calld); grpc_lb_policy_pick(lb_policy, bind_pollset, initial_metadata, &calld->picked_channel, - &calld->async_setup_task, &call_list); + &calld->async_setup_task, call_list); - GRPC_LB_POLICY_UNREF(lb_policy, "pick"); + GRPC_LB_POLICY_UNREF(lb_policy, "pick", call_list); } else if (chand->resolver != NULL) { calld->state = CALL_WAITING_FOR_CONFIG; add_to_lb_policy_wait_queue_locked_state_config(elem); @@ -409,7 +413,7 @@ static void perform_transport_stream_op(grpc_call_element *elem, chand->started_resolving = 1; grpc_resolver_next(chand->resolver, &chand->incoming_configuration, - &chand->on_config_changed); + &chand->on_config_changed, call_list); } gpr_mu_unlock(&chand->mu_config); gpr_mu_unlock(&calld->mu_state); @@ -417,19 +421,18 @@ static void perform_transport_stream_op(grpc_call_element *elem, calld->state = CALL_CANCELLED; gpr_mu_unlock(&chand->mu_config); gpr_mu_unlock(&calld->mu_state); - handle_op_after_cancellation(elem, op); + handle_op_after_cancellation(elem, op, call_list); } } } break; } - - grpc_call_list_run(call_list); } static void cc_start_transport_stream_op(grpc_call_element *elem, - grpc_transport_stream_op *op) { - perform_transport_stream_op(elem, op, 0); + grpc_transport_stream_op *op, + grpc_call_list *call_list) { + perform_transport_stream_op(elem, op, 0, call_list); } static void watch_lb_policy(channel_data *chand, grpc_lb_policy *lb_policy, @@ -448,16 +451,14 @@ static void on_lb_policy_state_changed_locked(lb_policy_connectivity_watcher *w, } } -static void on_lb_policy_state_changed(void *arg, int iomgr_success) { +static void on_lb_policy_state_changed(void *arg, int iomgr_success, + grpc_call_list *call_list) { lb_policy_connectivity_watcher *w = arg; - grpc_call_list cl = GRPC_CALL_LIST_INIT; gpr_mu_lock(&w->chand->mu_config); - on_lb_policy_state_changed_locked(w, &cl); + on_lb_policy_state_changed_locked(w, call_list); gpr_mu_unlock(&w->chand->mu_config); - grpc_call_list_run(cl); - GRPC_CHANNEL_INTERNAL_UNREF(w->chand->master, "watch_lb_policy"); gpr_free(w); } @@ -476,13 +477,13 @@ static void watch_lb_policy(channel_data *chand, grpc_lb_policy *lb_policy, call_list); } -static void cc_on_config_changed(void *arg, int iomgr_success) { +static void cc_on_config_changed(void *arg, int iomgr_success, + grpc_call_list *call_list) { channel_data *chand = arg; grpc_lb_policy *lb_policy = NULL; grpc_lb_policy *old_lb_policy; grpc_resolver *old_resolver; grpc_connectivity_state state = GRPC_CHANNEL_TRANSIENT_FAILURE; - grpc_call_list cl = GRPC_CALL_LIST_INIT; int exit_idle = 0; if (chand->incoming_configuration != NULL) { @@ -490,10 +491,10 @@ static void cc_on_config_changed(void *arg, int iomgr_success) { if (lb_policy != NULL) { GRPC_LB_POLICY_REF(lb_policy, "channel"); GRPC_LB_POLICY_REF(lb_policy, "config_change"); - state = grpc_lb_policy_check_connectivity(lb_policy, &cl); + state = grpc_lb_policy_check_connectivity(lb_policy, call_list); } - grpc_client_config_unref(chand->incoming_configuration); + grpc_client_config_unref(chand->incoming_configuration, call_list); } chand->incoming_configuration = NULL; @@ -502,7 +503,7 @@ static void cc_on_config_changed(void *arg, int iomgr_success) { old_lb_policy = chand->lb_policy; chand->lb_policy = lb_policy; if (lb_policy != NULL || chand->resolver == NULL /* disconnected */) { - grpc_call_list_move(&chand->waiting_for_config_closures, &cl); + grpc_call_list_move(&chand->waiting_for_config_closures, call_list); } if (lb_policy != NULL && chand->exit_idle_when_lb_policy_arrives) { GRPC_LB_POLICY_REF(lb_policy, "exit_idle"); @@ -514,57 +515,53 @@ static void cc_on_config_changed(void *arg, int iomgr_success) { grpc_resolver *resolver = chand->resolver; GRPC_RESOLVER_REF(resolver, "channel-next"); grpc_connectivity_state_set(&chand->state_tracker, state, "new_lb+resolver", - &cl); + call_list); if (lb_policy != NULL) { - watch_lb_policy(chand, lb_policy, state, &cl); + watch_lb_policy(chand, lb_policy, state, call_list); } gpr_mu_unlock(&chand->mu_config); GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver"); grpc_resolver_next(resolver, &chand->incoming_configuration, - &chand->on_config_changed); - GRPC_RESOLVER_UNREF(resolver, "channel-next"); + &chand->on_config_changed, call_list); + GRPC_RESOLVER_UNREF(resolver, "channel-next", call_list); } else { old_resolver = chand->resolver; chand->resolver = NULL; grpc_connectivity_state_set(&chand->state_tracker, GRPC_CHANNEL_FATAL_FAILURE, "resolver_gone", - &cl); + call_list); gpr_mu_unlock(&chand->mu_config); if (old_resolver != NULL) { - grpc_resolver_shutdown(old_resolver); - GRPC_RESOLVER_UNREF(old_resolver, "channel"); + grpc_resolver_shutdown(old_resolver, call_list); + GRPC_RESOLVER_UNREF(old_resolver, "channel", call_list); } } if (exit_idle) { - grpc_lb_policy_exit_idle(lb_policy, &cl); - GRPC_LB_POLICY_UNREF(lb_policy, "exit_idle"); + grpc_lb_policy_exit_idle(lb_policy, call_list); + GRPC_LB_POLICY_UNREF(lb_policy, "exit_idle", call_list); } if (old_lb_policy != NULL) { - grpc_lb_policy_shutdown(old_lb_policy, &cl); - GRPC_LB_POLICY_UNREF(old_lb_policy, "channel"); + grpc_lb_policy_shutdown(old_lb_policy, call_list); + GRPC_LB_POLICY_UNREF(old_lb_policy, "channel", call_list); } if (lb_policy != NULL) { - GRPC_LB_POLICY_UNREF(lb_policy, "config_change"); + GRPC_LB_POLICY_UNREF(lb_policy, "config_change", call_list); } - grpc_call_list_run(cl); GRPC_CHANNEL_INTERNAL_UNREF(chand->master, "resolver"); } static void cc_start_transport_op(grpc_channel_element *elem, - grpc_transport_op *op) { + grpc_transport_op *op, + grpc_call_list *call_list) { grpc_lb_policy *lb_policy = NULL; channel_data *chand = elem->channel_data; grpc_resolver *destroy_resolver = NULL; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; - if (op->on_consumed) { - grpc_call_list_add(&call_list, op->on_consumed, 1); - op->on_consumed = NULL; - } + grpc_call_list_add(call_list, op->on_consumed, 1); GPR_ASSERT(op->set_accept_stream == NULL); GPR_ASSERT(op->bind_pollset == NULL); @@ -573,7 +570,7 @@ static void cc_start_transport_op(grpc_channel_element *elem, if (op->on_connectivity_state_change != NULL) { grpc_connectivity_state_notify_on_state_change( &chand->state_tracker, op->connectivity_state, - op->on_connectivity_state_change, &call_list); + op->on_connectivity_state_change, call_list); op->on_connectivity_state_change = NULL; op->connectivity_state = NULL; } @@ -588,28 +585,26 @@ static void cc_start_transport_op(grpc_channel_element *elem, if (op->disconnect && chand->resolver != NULL) { grpc_connectivity_state_set(&chand->state_tracker, GRPC_CHANNEL_FATAL_FAILURE, "disconnect", - &call_list); + call_list); destroy_resolver = chand->resolver; chand->resolver = NULL; if (chand->lb_policy != NULL) { - grpc_lb_policy_shutdown(chand->lb_policy, &call_list); - GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel"); + grpc_lb_policy_shutdown(chand->lb_policy, call_list); + GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel", call_list); chand->lb_policy = NULL; } } gpr_mu_unlock(&chand->mu_config); if (destroy_resolver) { - grpc_resolver_shutdown(destroy_resolver); - GRPC_RESOLVER_UNREF(destroy_resolver, "channel"); + grpc_resolver_shutdown(destroy_resolver, call_list); + GRPC_RESOLVER_UNREF(destroy_resolver, "channel", call_list); } if (lb_policy) { - grpc_lb_policy_broadcast(lb_policy, op, &call_list); - GRPC_LB_POLICY_UNREF(lb_policy, "broadcast"); + grpc_lb_policy_broadcast(lb_policy, op, call_list); + GRPC_LB_POLICY_UNREF(lb_policy, "broadcast", call_list); } - - grpc_call_list_run(call_list); } /* Constructor for call_data */ @@ -630,7 +625,8 @@ static void init_call_elem(grpc_call_element *elem, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_call_element *elem) { +static void destroy_call_elem(grpc_call_element *elem, + grpc_call_list *call_list) { call_data *calld = elem->call_data; grpc_subchannel_call *subchannel_call; @@ -642,7 +638,7 @@ static void destroy_call_elem(grpc_call_element *elem) { case CALL_ACTIVE: subchannel_call = calld->subchannel_call; gpr_mu_unlock(&calld->mu_state); - GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call, "client_channel"); + GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call, "client_channel", call_list); break; case CALL_CREATED: case CALL_CANCELLED: @@ -662,7 +658,7 @@ static void destroy_call_elem(grpc_call_element *elem) { static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *metadata_context, int is_first, - int is_last) { + int is_last, grpc_call_list *call_list) { channel_data *chand = elem->channel_data; memset(chand, 0, sizeof(*chand)); @@ -681,15 +677,16 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, } /* Destructor for channel_data */ -static void destroy_channel_elem(grpc_channel_element *elem) { +static void destroy_channel_elem(grpc_channel_element *elem, + grpc_call_list *call_list) { channel_data *chand = elem->channel_data; if (chand->resolver != NULL) { - grpc_resolver_shutdown(chand->resolver); - GRPC_RESOLVER_UNREF(chand->resolver, "channel"); + grpc_resolver_shutdown(chand->resolver, call_list); + GRPC_RESOLVER_UNREF(chand->resolver, "channel", call_list); } if (chand->lb_policy != NULL) { - GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel"); + GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel", call_list); } grpc_connectivity_state_destroy(&chand->state_tracker); grpc_pollset_set_destroy(&chand->pollset_set); @@ -710,7 +707,8 @@ const grpc_channel_filter grpc_client_channel_filter = { }; void grpc_client_channel_set_resolver(grpc_channel_stack *channel_stack, - grpc_resolver *resolver) { + grpc_resolver *resolver, + grpc_call_list *call_list) { /* post construction initialization: set the transport setup pointer */ grpc_channel_element *elem = grpc_channel_stack_last_element(channel_stack); channel_data *chand = elem->channel_data; @@ -723,7 +721,7 @@ void grpc_client_channel_set_resolver(grpc_channel_stack *channel_stack, chand->started_resolving = 1; GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver"); grpc_resolver_next(resolver, &chand->incoming_configuration, - &chand->on_config_changed); + &chand->on_config_changed, call_list); } gpr_mu_unlock(&chand->mu_config); } @@ -743,7 +741,7 @@ grpc_connectivity_state grpc_client_channel_check_connectivity_state( GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver"); chand->started_resolving = 1; grpc_resolver_next(chand->resolver, &chand->incoming_configuration, - &chand->on_config_changed); + &chand->on_config_changed, call_list); } } } @@ -768,13 +766,15 @@ grpc_pollset_set *grpc_client_channel_get_connecting_pollset_set( } void grpc_client_channel_add_interested_party(grpc_channel_element *elem, - grpc_pollset *pollset) { + grpc_pollset *pollset, + grpc_call_list *call_list) { channel_data *chand = elem->channel_data; - grpc_pollset_set_add_pollset(&chand->pollset_set, pollset); + grpc_pollset_set_add_pollset(&chand->pollset_set, pollset, call_list); } void grpc_client_channel_del_interested_party(grpc_channel_element *elem, - grpc_pollset *pollset) { + grpc_pollset *pollset, + grpc_call_list *call_list) { channel_data *chand = elem->channel_data; - grpc_pollset_set_del_pollset(&chand->pollset_set, pollset); + grpc_pollset_set_del_pollset(&chand->pollset_set, pollset, call_list); } diff --git a/src/core/channel/client_channel.h b/src/core/channel/client_channel.h index bd84dad6b7..7afc7f8544 100644 --- a/src/core/channel/client_channel.h +++ b/src/core/channel/client_channel.h @@ -50,7 +50,8 @@ extern const grpc_channel_filter grpc_client_channel_filter; transport setup it should cancel upon destruction, or initiate when it needs a connection */ void grpc_client_channel_set_resolver(grpc_channel_stack *channel_stack, - grpc_resolver *resolver); + grpc_resolver *resolver, + grpc_call_list *call_list); grpc_connectivity_state grpc_client_channel_check_connectivity_state( grpc_channel_element *elem, int try_to_connect, grpc_call_list *call_list); @@ -63,8 +64,10 @@ grpc_pollset_set *grpc_client_channel_get_connecting_pollset_set( grpc_channel_element *elem); void grpc_client_channel_add_interested_party(grpc_channel_element *channel, - grpc_pollset *pollset); + grpc_pollset *pollset, + grpc_call_list *call_list); void grpc_client_channel_del_interested_party(grpc_channel_element *channel, - grpc_pollset *pollset); + grpc_pollset *pollset, + grpc_call_list *call_list); #endif /* GRPC_INTERNAL_CORE_CHANNEL_CLIENT_CHANNEL_H */ diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c index 7959603102..5a7403ccfd 100644 --- a/src/core/channel/compress_filter.c +++ b/src/core/channel/compress_filter.c @@ -269,13 +269,14 @@ static void process_send_ops(grpc_call_element *elem, op contains type and call direction information, in addition to the data that is being sent or received. */ static void compress_start_transport_stream_op(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { if (op->send_ops && op->send_ops->nops > 0) { process_send_ops(elem, op->send_ops); } /* pass control down the stack */ - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); } /* Constructor for call_data */ @@ -298,7 +299,8 @@ static void init_call_elem(grpc_call_element *elem, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_call_element *elem) { +static void destroy_call_elem(grpc_call_element *elem, + grpc_call_list *call_list) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; gpr_slice_buffer_destroy(&calld->slices); @@ -307,7 +309,8 @@ static void destroy_call_elem(grpc_call_element *elem) { /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *mdctx, - int is_first, int is_last) { + int is_first, int is_last, + grpc_call_list *call_list) { channel_data *channeld = elem->channel_data; grpc_compression_algorithm algo_idx; const char *supported_algorithms_names[GRPC_COMPRESS_ALGORITHMS_COUNT - 1]; @@ -369,7 +372,8 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, } /* Destructor for channel data */ -static void destroy_channel_elem(grpc_channel_element *elem) { +static void destroy_channel_elem(grpc_channel_element *elem, + grpc_call_list *call_list) { channel_data *channeld = elem->channel_data; grpc_compression_algorithm algo_idx; diff --git a/src/core/channel/connected_channel.c b/src/core/channel/connected_channel.c index b95ed06f2b..00a4d61afe 100644 --- a/src/core/channel/connected_channel.c +++ b/src/core/channel/connected_channel.c @@ -62,7 +62,8 @@ typedef struct connected_channel_call_data { void *unused; } call_data; /* Intercept a call operation and either push it directly up or translate it into transport stream operations */ static void con_start_transport_stream_op(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; GPR_ASSERT(elem->filter == &grpc_connected_channel_filter); @@ -73,7 +74,8 @@ static void con_start_transport_stream_op(grpc_call_element *elem, } static void con_start_transport_op(grpc_channel_element *elem, - grpc_transport_op *op) { + grpc_transport_op *op, + grpc_call_list *call_list) { channel_data *chand = elem->channel_data; grpc_transport_perform_op(chand->transport, op); } @@ -94,7 +96,8 @@ static void init_call_elem(grpc_call_element *elem, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_call_element *elem) { +static void destroy_call_elem(grpc_call_element *elem, + grpc_call_list *call_list) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; GPR_ASSERT(elem->filter == &grpc_connected_channel_filter); @@ -105,7 +108,8 @@ static void destroy_call_elem(grpc_call_element *elem) { /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *mdctx, - int is_first, int is_last) { + int is_first, int is_last, + grpc_call_list *call_list) { channel_data *cd = (channel_data *)elem->channel_data; GPR_ASSERT(is_last); GPR_ASSERT(elem->filter == &grpc_connected_channel_filter); @@ -113,13 +117,14 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, } /* Destructor for channel_data */ -static void destroy_channel_elem(grpc_channel_element *elem) { +static void destroy_channel_elem(grpc_channel_element *elem, + grpc_call_list *call_list) { channel_data *cd = (channel_data *)elem->channel_data; GPR_ASSERT(elem->filter == &grpc_connected_channel_filter); grpc_transport_destroy(cd->transport); } -static char *con_get_peer(grpc_call_element *elem) { +static char *con_get_peer(grpc_call_element *elem, grpc_call_list *call_list) { channel_data *chand = elem->channel_data; return grpc_transport_get_peer(chand->transport); } diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c index bcaaca0f7d..4e9eac72ba 100644 --- a/src/core/channel/http_client_filter.c +++ b/src/core/channel/http_client_filter.c @@ -67,22 +67,26 @@ typedef struct channel_data { grpc_mdelem *user_agent; } channel_data; -/* used to silence 'variable not used' warnings */ -static void ignore_unused(void *ignored) {} +typedef struct { + grpc_call_element *elem; + grpc_call_list *call_list; +} client_filter_args; static grpc_mdelem *client_filter(void *user_data, grpc_mdelem *md) { - grpc_call_element *elem = user_data; + client_filter_args *a = user_data; + grpc_call_element *elem = a->elem; channel_data *channeld = elem->channel_data; if (md == channeld->status) { return NULL; } else if (md->key == channeld->status->key) { - grpc_call_element_send_cancel(elem); + grpc_call_element_send_cancel(elem, a->call_list); return NULL; } return md; } -static void hc_on_recv(void *user_data, int success) { +static void hc_on_recv(void *user_data, int success, + grpc_call_list *call_list) { grpc_call_element *elem = user_data; call_data *calld = elem->call_data; size_t i; @@ -90,11 +94,14 @@ static void hc_on_recv(void *user_data, int success) { grpc_stream_op *ops = calld->recv_ops->ops; for (i = 0; i < nops; i++) { grpc_stream_op *op = &ops[i]; + client_filter_args a; if (op->type != GRPC_OP_METADATA) continue; calld->got_initial_metadata = 1; - grpc_metadata_batch_filter(&op->data.metadata, client_filter, elem); + a.elem = elem; + a.call_list = call_list; + grpc_metadata_batch_filter(&op->data.metadata, client_filter, &a); } - calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); + calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success, call_list); } static grpc_mdelem *client_strip_filter(void *user_data, grpc_mdelem *md) { @@ -148,10 +155,11 @@ static void hc_mutate_op(grpc_call_element *elem, } static void hc_start_transport_op(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); hc_mutate_op(elem, op); - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); } /* Constructor for call_data */ @@ -167,14 +175,8 @@ static void init_call_elem(grpc_call_element *elem, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_call_element *elem) { - /* grab pointers to our data from the call element */ - call_data *calld = elem->call_data; - channel_data *channeld = elem->channel_data; - - ignore_unused(calld); - ignore_unused(channeld); -} +static void destroy_call_elem(grpc_call_element *elem, + grpc_call_list *call_list) {} static const char *scheme_from_args(const grpc_channel_args *args) { unsigned i; @@ -241,7 +243,8 @@ static grpc_mdstr *user_agent_from_args(grpc_mdctx *mdctx, /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *channel_args, - grpc_mdctx *mdctx, int is_first, int is_last) { + grpc_mdctx *mdctx, int is_first, int is_last, + grpc_call_list *call_list) { /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; @@ -264,7 +267,8 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, } /* Destructor for channel data */ -static void destroy_channel_elem(grpc_channel_element *elem) { +static void destroy_channel_elem(grpc_channel_element *elem, + grpc_call_list *call_list) { /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c index 7a9f17d057..201adf4f35 100644 --- a/src/core/channel/http_server_filter.c +++ b/src/core/channel/http_server_filter.c @@ -74,8 +74,14 @@ typedef struct channel_data { grpc_mdctx *mdctx; } channel_data; +typedef struct { + grpc_call_element *elem; + grpc_call_list *call_list; +} server_filter_args; + static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { - grpc_call_element *elem = user_data; + server_filter_args *a = user_data; + grpc_call_element *elem = a->elem; channel_data *channeld = elem->channel_data; call_data *calld = elem->call_data; @@ -118,7 +124,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { /* swallow it and error everything out. */ /* TODO(klempner): We ought to generate more descriptive error messages on the wire here. */ - grpc_call_element_send_cancel(elem); + grpc_call_element_send_cancel(elem, a->call_list); return NULL; } else if (md->key == channeld->path_key) { if (calld->seen_path) { @@ -144,7 +150,8 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { } } -static void hs_on_recv(void *user_data, int success) { +static void hs_on_recv(void *user_data, int success, + grpc_call_list *call_list) { grpc_call_element *elem = user_data; call_data *calld = elem->call_data; if (success) { @@ -153,9 +160,12 @@ static void hs_on_recv(void *user_data, int success) { grpc_stream_op *ops = calld->recv_ops->ops; for (i = 0; i < nops; i++) { grpc_stream_op *op = &ops[i]; + server_filter_args a; if (op->type != GRPC_OP_METADATA) continue; calld->got_initial_metadata = 1; - grpc_metadata_batch_filter(&op->data.metadata, server_filter, elem); + a.elem = elem; + a.call_list = call_list; + grpc_metadata_batch_filter(&op->data.metadata, server_filter, &a); /* Have we seen the required http2 transport headers? (:method, :scheme, content-type, with :path and :authority covered at the channel level right now) */ @@ -180,11 +190,11 @@ static void hs_on_recv(void *user_data, int success) { } /* Error this call out */ success = 0; - grpc_call_element_send_cancel(elem); + grpc_call_element_send_cancel(elem, call_list); } } } - calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); + calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success, call_list); } static void hs_mutate_op(grpc_call_element *elem, @@ -218,10 +228,11 @@ static void hs_mutate_op(grpc_call_element *elem, } static void hs_start_transport_op(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); hs_mutate_op(elem, op); - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); } /* Constructor for call_data */ @@ -237,12 +248,14 @@ static void init_call_elem(grpc_call_element *elem, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_call_element *elem) {} +static void destroy_call_elem(grpc_call_element *elem, + grpc_call_list *call_list) {} /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *mdctx, - int is_first, int is_last) { + int is_first, int is_last, + grpc_call_list *call_list) { /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; @@ -271,7 +284,8 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, } /* Destructor for channel data */ -static void destroy_channel_elem(grpc_channel_element *elem) { +static void destroy_channel_elem(grpc_channel_element *elem, + grpc_call_list *call_list) { /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; diff --git a/src/core/channel/noop_filter.c b/src/core/channel/noop_filter.c index d631885aaf..d4f1acb4c5 100644 --- a/src/core/channel/noop_filter.c +++ b/src/core/channel/noop_filter.c @@ -63,11 +63,12 @@ static void noop_mutate_op(grpc_call_element *elem, op contains type and call direction information, in addition to the data that is being sent or received. */ static void noop_start_transport_stream_op(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { noop_mutate_op(elem, op); /* pass control down the stack */ - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); } /* Constructor for call_data */ @@ -85,19 +86,14 @@ static void init_call_elem(grpc_call_element *elem, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_call_element *elem) { - /* grab pointers to our data from the call element */ - call_data *calld = elem->call_data; - channel_data *channeld = elem->channel_data; - - ignore_unused(calld); - ignore_unused(channeld); -} +static void destroy_call_elem(grpc_call_element *elem, + grpc_call_list *call_list) {} /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *mdctx, - int is_first, int is_last) { + int is_first, int is_last, + grpc_call_list *call_list) { /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; @@ -112,7 +108,8 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, } /* Destructor for channel data */ -static void destroy_channel_elem(grpc_channel_element *elem) { +static void destroy_channel_elem(grpc_channel_element *elem, + grpc_call_list *call_list) { /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; diff --git a/src/core/client_config/client_config.c b/src/core/client_config/client_config.c index 4453824148..0780880473 100644 --- a/src/core/client_config/client_config.c +++ b/src/core/client_config/client_config.c @@ -51,21 +51,20 @@ grpc_client_config *grpc_client_config_create() { void grpc_client_config_ref(grpc_client_config *c) { gpr_ref(&c->refs); } -void grpc_client_config_unref(grpc_client_config *c) { +void grpc_client_config_unref(grpc_client_config *c, + grpc_call_list *call_list) { if (gpr_unref(&c->refs)) { - GRPC_LB_POLICY_UNREF(c->lb_policy, "client_config"); + GRPC_LB_POLICY_UNREF(c->lb_policy, "client_config", call_list); gpr_free(c); } } void grpc_client_config_set_lb_policy(grpc_client_config *c, grpc_lb_policy *lb_policy) { + GPR_ASSERT(c->lb_policy == NULL); if (lb_policy) { GRPC_LB_POLICY_REF(lb_policy, "client_config"); } - if (c->lb_policy) { - GRPC_LB_POLICY_UNREF(c->lb_policy, "client_config"); - } c->lb_policy = lb_policy; } diff --git a/src/core/client_config/client_config.h b/src/core/client_config/client_config.h index 47612da42c..76a5c66594 100644 --- a/src/core/client_config/client_config.h +++ b/src/core/client_config/client_config.h @@ -42,7 +42,8 @@ typedef struct grpc_client_config grpc_client_config; grpc_client_config *grpc_client_config_create(); void grpc_client_config_ref(grpc_client_config *client_config); -void grpc_client_config_unref(grpc_client_config *client_config); +void grpc_client_config_unref(grpc_client_config *client_config, + grpc_call_list *call_list); void grpc_client_config_set_lb_policy(grpc_client_config *client_config, grpc_lb_policy *lb_policy); diff --git a/src/core/client_config/connector.c b/src/core/client_config/connector.c index 6252d57271..31f0b84efe 100644 --- a/src/core/client_config/connector.c +++ b/src/core/client_config/connector.c @@ -44,10 +44,11 @@ void grpc_connector_unref(grpc_connector *connector) { void grpc_connector_connect(grpc_connector *connector, const grpc_connect_in_args *in_args, grpc_connect_out_args *out_args, - grpc_closure *notify) { - connector->vtable->connect(connector, in_args, out_args, notify); + grpc_closure *notify, grpc_call_list *call_list) { + connector->vtable->connect(connector, in_args, out_args, notify, call_list); } -void grpc_connector_shutdown(grpc_connector *connector) { - connector->vtable->shutdown(connector); +void grpc_connector_shutdown(grpc_connector *connector, + grpc_call_list *call_list) { + connector->vtable->shutdown(connector, call_list); } diff --git a/src/core/client_config/connector.h b/src/core/client_config/connector.h index fc1f6708de..388f656c44 100644 --- a/src/core/client_config/connector.h +++ b/src/core/client_config/connector.h @@ -55,10 +55,6 @@ typedef struct { gpr_timespec deadline; /** channel arguments (to be passed to transport) */ const grpc_channel_args *channel_args; - /** metadata context */ - grpc_mdctx *metadata_context; - /** workqueue */ - grpc_workqueue *workqueue; } grpc_connect_in_args; typedef struct { @@ -71,23 +67,25 @@ typedef struct { struct grpc_connector_vtable { void (*ref)(grpc_connector *connector); - void (*unref)(grpc_connector *connector); + void (*unref)(grpc_connector *connector, grpc_call_list *call_list); /** Implementation of grpc_connector_shutdown */ - void (*shutdown)(grpc_connector *connector); + void (*shutdown)(grpc_connector *connector, grpc_call_list *call_list); /** Implementation of grpc_connector_connect */ void (*connect)(grpc_connector *connector, const grpc_connect_in_args *in_args, - grpc_connect_out_args *out_args, grpc_closure *notify); + grpc_connect_out_args *out_args, grpc_closure *notify, + grpc_call_list *call_list); }; void grpc_connector_ref(grpc_connector *connector); -void grpc_connector_unref(grpc_connector *connector); +void grpc_connector_unref(grpc_connector *connector, grpc_call_list *call_list); /** Connect using the connector: max one outstanding call at a time */ void grpc_connector_connect(grpc_connector *connector, const grpc_connect_in_args *in_args, grpc_connect_out_args *out_args, - grpc_closure *notify); + grpc_closure *notify, grpc_call_list *call_list); /** Cancel any pending connection */ -void grpc_connector_shutdown(grpc_connector *connector); +void grpc_connector_shutdown(grpc_connector *connector, + grpc_call_list *call_list); #endif diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c index 852eed310d..6dc52f43ce 100644 --- a/src/core/client_config/lb_policies/pick_first.c +++ b/src/core/client_config/lb_policies/pick_first.c @@ -52,8 +52,6 @@ typedef struct { /** all our subchannels */ grpc_subchannel **subchannels; size_t num_subchannels; - /** workqueue for async work */ - grpc_workqueue *workqueue; grpc_closure connectivity_changed; @@ -78,33 +76,34 @@ typedef struct { grpc_connectivity_state_tracker state_tracker; } pick_first_lb_policy; -static void del_interested_parties_locked(pick_first_lb_policy *p) { +static void del_interested_parties_locked(pick_first_lb_policy *p, + grpc_call_list *call_list) { pending_pick *pp; for (pp = p->pending_picks; pp; pp = pp->next) { grpc_subchannel_del_interested_party(p->subchannels[p->checking_subchannel], - pp->pollset); + pp->pollset, call_list); } } -static void add_interested_parties_locked(pick_first_lb_policy *p) { +static void add_interested_parties_locked(pick_first_lb_policy *p, + grpc_call_list *call_list) { pending_pick *pp; for (pp = p->pending_picks; pp; pp = pp->next) { grpc_subchannel_add_interested_party(p->subchannels[p->checking_subchannel], - pp->pollset); + pp->pollset, call_list); } } -void pf_destroy(grpc_lb_policy *pol) { +void pf_destroy(grpc_lb_policy *pol, grpc_call_list *call_list) { pick_first_lb_policy *p = (pick_first_lb_policy *)pol; size_t i; - del_interested_parties_locked(p); + GPR_ASSERT(p->shutdown); for (i = 0; i < p->num_subchannels; i++) { - GRPC_SUBCHANNEL_UNREF(p->subchannels[i], "pick_first"); + GRPC_SUBCHANNEL_UNREF(p->subchannels[i], "pick_first", call_list); } grpc_connectivity_state_destroy(&p->state_tracker); gpr_free(p->subchannels); gpr_mu_destroy(&p->mu); - GRPC_WORKQUEUE_UNREF(p->workqueue, "pick_first"); gpr_free(p); } @@ -112,7 +111,7 @@ void pf_shutdown(grpc_lb_policy *pol, grpc_call_list *call_list) { pick_first_lb_policy *p = (pick_first_lb_policy *)pol; pending_pick *pp; gpr_mu_lock(&p->mu); - del_interested_parties_locked(p); + del_interested_parties_locked(p, call_list); p->shutdown = 1; pp = p->pending_picks; p->pending_picks = NULL; @@ -156,13 +155,13 @@ void pf_pick(grpc_lb_policy *pol, grpc_pollset *pollset, if (p->selected) { gpr_mu_unlock(&p->mu); *target = p->selected; - on_complete->cb(on_complete->cb_arg, 1); + grpc_call_list_add(call_list, on_complete, 1); } else { if (!p->started_picking) { start_picking(p, call_list); } grpc_subchannel_add_interested_party(p->subchannels[p->checking_subchannel], - pollset); + pollset, call_list); pp = gpr_malloc(sizeof(*pp)); pp->next = p->pending_picks; pp->pollset = pollset; @@ -173,58 +172,58 @@ void pf_pick(grpc_lb_policy *pol, grpc_pollset *pollset, } } -static void pf_connectivity_changed(void *arg, int iomgr_success) { +static void pf_connectivity_changed(void *arg, int iomgr_success, + grpc_call_list *call_list) { pick_first_lb_policy *p = arg; pending_pick *pp; - int unref = 0; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; gpr_mu_lock(&p->mu); if (p->shutdown) { - unref = 1; + GRPC_LB_POLICY_UNREF(&p->base, "pick_first_connectivity", call_list); } else if (p->selected != NULL) { grpc_connectivity_state_set(&p->state_tracker, p->checking_connectivity, - "selected_changed", &call_list); + "selected_changed", call_list); if (p->checking_connectivity != GRPC_CHANNEL_FATAL_FAILURE) { grpc_subchannel_notify_on_state_change( p->selected, &p->checking_connectivity, &p->connectivity_changed, - &call_list); + call_list); } else { - unref = 1; + GRPC_LB_POLICY_UNREF(&p->base, "pick_first_connectivity", call_list); } } else { loop: switch (p->checking_connectivity) { case GRPC_CHANNEL_READY: grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_READY, - "connecting_ready", &call_list); + "connecting_ready", call_list); p->selected = p->subchannels[p->checking_subchannel]; while ((pp = p->pending_picks)) { p->pending_picks = pp->next; *pp->target = p->selected; - grpc_subchannel_del_interested_party(p->selected, pp->pollset); - grpc_call_list_add(&call_list, pp->on_complete, 1); + grpc_subchannel_del_interested_party(p->selected, pp->pollset, + call_list); + grpc_call_list_add(call_list, pp->on_complete, 1); gpr_free(pp); } grpc_subchannel_notify_on_state_change( p->selected, &p->checking_connectivity, &p->connectivity_changed, - &call_list); + call_list); break; case GRPC_CHANNEL_TRANSIENT_FAILURE: grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, - "connecting_transient_failure", &call_list); - del_interested_parties_locked(p); + "connecting_transient_failure", call_list); + del_interested_parties_locked(p, call_list); p->checking_subchannel = (p->checking_subchannel + 1) % p->num_subchannels; p->checking_connectivity = grpc_subchannel_check_connectivity( p->subchannels[p->checking_subchannel]); - add_interested_parties_locked(p); + add_interested_parties_locked(p, call_list); if (p->checking_connectivity == GRPC_CHANNEL_TRANSIENT_FAILURE) { grpc_subchannel_notify_on_state_change( p->subchannels[p->checking_subchannel], &p->checking_connectivity, - &p->connectivity_changed, &call_list); + &p->connectivity_changed, call_list); } else { goto loop; } @@ -232,48 +231,43 @@ static void pf_connectivity_changed(void *arg, int iomgr_success) { case GRPC_CHANNEL_CONNECTING: case GRPC_CHANNEL_IDLE: grpc_connectivity_state_set(&p->state_tracker, p->checking_connectivity, - "connecting_changed", &call_list); + "connecting_changed", call_list); grpc_subchannel_notify_on_state_change( p->subchannels[p->checking_subchannel], &p->checking_connectivity, - &p->connectivity_changed, &call_list); + &p->connectivity_changed, call_list); break; case GRPC_CHANNEL_FATAL_FAILURE: - del_interested_parties_locked(p); + del_interested_parties_locked(p, call_list); GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel], p->subchannels[p->num_subchannels - 1]); p->num_subchannels--; - GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first"); + GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first", + call_list); if (p->num_subchannels == 0) { grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_FATAL_FAILURE, - "no_more_channels", &call_list); + "no_more_channels", call_list); while ((pp = p->pending_picks)) { p->pending_picks = pp->next; *pp->target = NULL; - grpc_call_list_add(&call_list, pp->on_complete, 1); + grpc_call_list_add(call_list, pp->on_complete, 1); gpr_free(pp); } - unref = 1; + GRPC_LB_POLICY_UNREF(&p->base, "pick_first_connectivity", call_list); } else { grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, - "subchannel_failed", &call_list); + "subchannel_failed", call_list); p->checking_subchannel %= p->num_subchannels; p->checking_connectivity = grpc_subchannel_check_connectivity( p->subchannels[p->checking_subchannel]); - add_interested_parties_locked(p); + add_interested_parties_locked(p, call_list); goto loop; } } } gpr_mu_unlock(&p->mu); - - grpc_call_list_run(call_list); - - if (unref) { - GRPC_LB_POLICY_UNREF(&p->base, "pick_first_connectivity"); - } } static void pf_broadcast(grpc_lb_policy *pol, grpc_transport_op *op, @@ -293,8 +287,8 @@ static void pf_broadcast(grpc_lb_policy *pol, grpc_transport_op *op, gpr_mu_unlock(&p->mu); for (i = 0; i < n; i++) { - grpc_subchannel_process_transport_op(subchannels[i], op); - GRPC_SUBCHANNEL_UNREF(subchannels[i], "pf_broadcast"); + grpc_subchannel_process_transport_op(subchannels[i], op, call_list); + GRPC_SUBCHANNEL_UNREF(subchannels[i], "pf_broadcast", call_list); } gpr_free(subchannels); } @@ -341,8 +335,6 @@ static grpc_lb_policy *create_pick_first(grpc_lb_policy_factory *factory, grpc_lb_policy_init(&p->base, &pick_first_lb_policy_vtable); p->subchannels = gpr_malloc(sizeof(grpc_subchannel *) * args->num_subchannels); p->num_subchannels = args->num_subchannels; - p->workqueue = args->workqueue; - GRPC_WORKQUEUE_REF(p->workqueue, "pick_first"); grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE, "pick_first"); memcpy(p->subchannels, args->subchannels, diff --git a/src/core/client_config/lb_policy.c b/src/core/client_config/lb_policy.c index 95ab29cd88..84a033d583 100644 --- a/src/core/client_config/lb_policy.c +++ b/src/core/client_config/lb_policy.c @@ -51,15 +51,15 @@ void grpc_lb_policy_ref(grpc_lb_policy *policy) { } #ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG -void grpc_lb_policy_unref(grpc_lb_policy *policy, const char *file, int line, - const char *reason) { +void grpc_lb_policy_unref(grpc_lb_policy *policy, grpc_call_list *call_list, + const char *file, int line, const char *reason) { gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "LB_POLICY:%p unref %d -> %d %s", policy, (int)policy->refs.count, (int)policy->refs.count - 1, reason); #else -void grpc_lb_policy_unref(grpc_lb_policy *policy) { +void grpc_lb_policy_unref(grpc_lb_policy *policy, grpc_call_list *call_list) { #endif if (gpr_unref(&policy->refs)) { - policy->vtable->destroy(policy); + policy->vtable->destroy(policy, call_list); } } diff --git a/src/core/client_config/lb_policy.h b/src/core/client_config/lb_policy.h index e445d2cded..f9430b4250 100644 --- a/src/core/client_config/lb_policy.h +++ b/src/core/client_config/lb_policy.h @@ -51,7 +51,7 @@ struct grpc_lb_policy { }; struct grpc_lb_policy_vtable { - void (*destroy)(grpc_lb_policy *policy); + void (*destroy)(grpc_lb_policy *policy, grpc_call_list *call_list); void (*shutdown)(grpc_lb_policy *policy, grpc_call_list *call_list); @@ -82,17 +82,17 @@ struct grpc_lb_policy_vtable { #ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG #define GRPC_LB_POLICY_REF(p, r) \ grpc_lb_policy_ref((p), __FILE__, __LINE__, (r)) -#define GRPC_LB_POLICY_UNREF(p, r) \ - grpc_lb_policy_unref((p), __FILE__, __LINE__, (r)) +#define GRPC_LB_POLICY_UNREF(p, r, cl) \ + grpc_lb_policy_unref((p), (cl), __FILE__, __LINE__, (r)) void grpc_lb_policy_ref(grpc_lb_policy *policy, const char *file, int line, const char *reason); -void grpc_lb_policy_unref(grpc_lb_policy *policy, const char *file, int line, - const char *reason); +void grpc_lb_policy_unref(grpc_lb_policy *policy, grpc_call_list *call_list, + const char *file, int line, const char *reason); #else #define GRPC_LB_POLICY_REF(p, r) grpc_lb_policy_ref((p)) -#define GRPC_LB_POLICY_UNREF(p, r) grpc_lb_policy_unref((p)) +#define GRPC_LB_POLICY_UNREF(p, r, cl) grpc_lb_policy_unref((p), (cl)) void grpc_lb_policy_ref(grpc_lb_policy *policy); -void grpc_lb_policy_unref(grpc_lb_policy *policy); +void grpc_lb_policy_unref(grpc_lb_policy *policy, grpc_call_list *call_list); #endif /** called by concrete implementations to initialize the base struct */ diff --git a/src/core/client_config/lb_policy_factory.h b/src/core/client_config/lb_policy_factory.h index 755faf8813..04610316ee 100644 --- a/src/core/client_config/lb_policy_factory.h +++ b/src/core/client_config/lb_policy_factory.h @@ -36,7 +36,6 @@ #include "src/core/client_config/lb_policy.h" #include "src/core/client_config/subchannel.h" -#include "src/core/iomgr/workqueue.h" typedef struct grpc_lb_policy_factory grpc_lb_policy_factory; typedef struct grpc_lb_policy_factory_vtable grpc_lb_policy_factory_vtable; @@ -50,7 +49,6 @@ struct grpc_lb_policy_factory { typedef struct grpc_lb_policy_args { grpc_subchannel **subchannels; size_t num_subchannels; - grpc_workqueue *workqueue; } grpc_lb_policy_args; struct grpc_lb_policy_factory_vtable { diff --git a/src/core/client_config/resolver.c b/src/core/client_config/resolver.c index 6a7d73c54f..eae8ec6ddf 100644 --- a/src/core/client_config/resolver.c +++ b/src/core/client_config/resolver.c @@ -40,8 +40,8 @@ void grpc_resolver_init(grpc_resolver *resolver, } #ifdef GRPC_RESOLVER_REFCOUNT_DEBUG -void grpc_resolver_ref(grpc_resolver *resolver, const char *file, int line, - const char *reason) { +void grpc_resolver_ref(grpc_resolver *resolver, grpc_call_list *call_list, + const char *file, int line, const char *reason) { gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "RESOLVER:%p ref %d -> %d %s", resolver, (int)resolver->refs.count, (int)resolver->refs.count + 1, reason); @@ -52,32 +52,34 @@ void grpc_resolver_ref(grpc_resolver *resolver) { } #ifdef GRPC_RESOLVER_REFCOUNT_DEBUG -void grpc_resolver_unref(grpc_resolver *resolver, const char *file, int line, - const char *reason) { +void grpc_resolver_unref(grpc_resolver *resolver, grpc_call_list *call_list, + const char *file, int line, const char *reason) { gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "RESOLVER:%p unref %d -> %d %s", resolver, (int)resolver->refs.count, (int)resolver->refs.count - 1, reason); #else -void grpc_resolver_unref(grpc_resolver *resolver) { +void grpc_resolver_unref(grpc_resolver *resolver, grpc_call_list *call_list) { #endif if (gpr_unref(&resolver->refs)) { - resolver->vtable->destroy(resolver); + resolver->vtable->destroy(resolver, call_list); } } -void grpc_resolver_shutdown(grpc_resolver *resolver) { - resolver->vtable->shutdown(resolver); +void grpc_resolver_shutdown(grpc_resolver *resolver, + grpc_call_list *call_list) { + resolver->vtable->shutdown(resolver, call_list); } void grpc_resolver_channel_saw_error(grpc_resolver *resolver, struct sockaddr *failing_address, - int failing_address_len) { + int failing_address_len, + grpc_call_list *call_list) { resolver->vtable->channel_saw_error(resolver, failing_address, - failing_address_len); + failing_address_len, call_list); } void grpc_resolver_next(grpc_resolver *resolver, grpc_client_config **target_config, - grpc_closure *on_complete) { - resolver->vtable->next(resolver, target_config, on_complete); + grpc_closure *on_complete, grpc_call_list *call_list) { + resolver->vtable->next(resolver, target_config, on_complete, call_list); } diff --git a/src/core/client_config/resolver.h b/src/core/client_config/resolver.h index 217d061abe..7960bcb92f 100644 --- a/src/core/client_config/resolver.h +++ b/src/core/client_config/resolver.h @@ -49,40 +49,41 @@ struct grpc_resolver { }; struct grpc_resolver_vtable { - void (*destroy)(grpc_resolver *resolver); - void (*shutdown)(grpc_resolver *resolver); + void (*destroy)(grpc_resolver *resolver, grpc_call_list *call_list); + void (*shutdown)(grpc_resolver *resolver, grpc_call_list *call_list); void (*channel_saw_error)(grpc_resolver *resolver, struct sockaddr *failing_address, - int failing_address_len); + int failing_address_len, grpc_call_list *call_list); void (*next)(grpc_resolver *resolver, grpc_client_config **target_config, - grpc_closure *on_complete); + grpc_closure *on_complete, grpc_call_list *call_list); }; #ifdef GRPC_RESOLVER_REFCOUNT_DEBUG #define GRPC_RESOLVER_REF(p, r) grpc_resolver_ref((p), __FILE__, __LINE__, (r)) -#define GRPC_RESOLVER_UNREF(p, r) \ - grpc_resolver_unref((p), __FILE__, __LINE__, (r)) +#define GRPC_RESOLVER_UNREF(p, r, cl) \ + grpc_resolver_unref((p), (cl), __FILE__, __LINE__, (r)) void grpc_resolver_ref(grpc_resolver *policy, const char *file, int line, const char *reason); -void grpc_resolver_unref(grpc_resolver *policy, const char *file, int line, - const char *reason); +void grpc_resolver_unref(grpc_resolver *policy, grpc_call_list *call_list, + const char *file, int line, const char *reason); #else #define GRPC_RESOLVER_REF(p, r) grpc_resolver_ref((p)) -#define GRPC_RESOLVER_UNREF(p, r) grpc_resolver_unref((p)) +#define GRPC_RESOLVER_UNREF(p, r, cl) grpc_resolver_unref((p), (cl)) void grpc_resolver_ref(grpc_resolver *policy); -void grpc_resolver_unref(grpc_resolver *policy); +void grpc_resolver_unref(grpc_resolver *policy, grpc_call_list *call_list); #endif void grpc_resolver_init(grpc_resolver *resolver, const grpc_resolver_vtable *vtable); -void grpc_resolver_shutdown(grpc_resolver *resolver); +void grpc_resolver_shutdown(grpc_resolver *resolver, grpc_call_list *call_list); /** Notification that the channel has seen an error on some address. Can be used as a hint that re-resolution is desirable soon. */ void grpc_resolver_channel_saw_error(grpc_resolver *resolver, struct sockaddr *failing_address, - int failing_address_len); + int failing_address_len, + grpc_call_list *call_list); /** Get the next client config. Called by the channel to fetch a new configuration. Expected to set *target_config with a new configuration, @@ -92,6 +93,6 @@ void grpc_resolver_channel_saw_error(grpc_resolver *resolver, schedule on_complete. */ void grpc_resolver_next(grpc_resolver *resolver, grpc_client_config **target_config, - grpc_closure *on_complete); + grpc_closure *on_complete, grpc_call_list *call_list); #endif /* GRPC_INTERNAL_CORE_CONFIG_RESOLVER_H */ diff --git a/src/core/client_config/resolver_factory.h b/src/core/client_config/resolver_factory.h index 24da84d89d..4c4df353f7 100644 --- a/src/core/client_config/resolver_factory.h +++ b/src/core/client_config/resolver_factory.h @@ -37,7 +37,6 @@ #include "src/core/client_config/resolver.h" #include "src/core/client_config/subchannel_factory.h" #include "src/core/client_config/uri_parser.h" -#include "src/core/iomgr/workqueue.h" typedef struct grpc_resolver_factory grpc_resolver_factory; typedef struct grpc_resolver_factory_vtable grpc_resolver_factory_vtable; @@ -51,7 +50,6 @@ struct grpc_resolver_factory { typedef struct grpc_resolver_args { grpc_uri *uri; grpc_subchannel_factory *subchannel_factory; - grpc_workqueue *workqueue; } grpc_resolver_args; struct grpc_resolver_factory_vtable { diff --git a/src/core/client_config/resolver_registry.c b/src/core/client_config/resolver_registry.c index 3611252bda..89a945c2d3 100644 --- a/src/core/client_config/resolver_registry.c +++ b/src/core/client_config/resolver_registry.c @@ -114,9 +114,8 @@ static grpc_resolver_factory *resolve_factory(const char *target, return factory; } -grpc_resolver *grpc_resolver_create(const char *target, - grpc_subchannel_factory *subchannel_factory, - grpc_workqueue *workqueue) { +grpc_resolver *grpc_resolver_create( + const char *target, grpc_subchannel_factory *subchannel_factory) { grpc_uri *uri = NULL; grpc_resolver_factory *factory = resolve_factory(target, &uri); grpc_resolver *resolver; @@ -124,7 +123,6 @@ grpc_resolver *grpc_resolver_create(const char *target, memset(&args, 0, sizeof(args)); args.uri = uri; args.subchannel_factory = subchannel_factory; - args.workqueue = workqueue; resolver = grpc_resolver_factory_create_resolver(factory, &args); grpc_uri_destroy(uri); return resolver; diff --git a/src/core/client_config/resolver_registry.h b/src/core/client_config/resolver_registry.h index 9a9a628262..5a7193b7ae 100644 --- a/src/core/client_config/resolver_registry.h +++ b/src/core/client_config/resolver_registry.h @@ -55,9 +55,8 @@ void grpc_register_resolver_type(grpc_resolver_factory *factory); If a resolver factory was found, use it to instantiate a resolver and return it. If a resolver factory was not found, return NULL. */ -grpc_resolver *grpc_resolver_create(const char *target, - grpc_subchannel_factory *subchannel_factory, - grpc_workqueue *workqueue); +grpc_resolver *grpc_resolver_create( + const char *target, grpc_subchannel_factory *subchannel_factory); /** Given a target, return a (freshly allocated with gpr_malloc) string representing the default authority to pass from a client. */ diff --git a/src/core/client_config/resolvers/dns_resolver.c b/src/core/client_config/resolvers/dns_resolver.c index e251664e03..9e9b3d4917 100644 --- a/src/core/client_config/resolvers/dns_resolver.c +++ b/src/core/client_config/resolvers/dns_resolver.c @@ -49,8 +49,6 @@ typedef struct { grpc_resolver base; /** refcount */ gpr_refcount refs; - /** workqueue */ - grpc_workqueue *workqueue; /** name to resolve */ char *name; /** default port to use */ @@ -76,37 +74,36 @@ typedef struct { grpc_client_config *resolved_config; } dns_resolver; -static void dns_destroy(grpc_resolver *r); +static void dns_destroy(grpc_resolver *r, grpc_call_list *call_list); static void dns_start_resolving_locked(dns_resolver *r); -static grpc_closure *dns_maybe_finish_next_locked(dns_resolver *r) - GRPC_MUST_USE_RESULT; +static void dns_maybe_finish_next_locked(dns_resolver *r, + grpc_call_list *call_list); -static void dns_shutdown(grpc_resolver *r); +static void dns_shutdown(grpc_resolver *r, grpc_call_list *call_list); static void dns_channel_saw_error(grpc_resolver *r, struct sockaddr *failing_address, - int failing_address_len); + int failing_address_len, + grpc_call_list *call_list); static void dns_next(grpc_resolver *r, grpc_client_config **target_config, - grpc_closure *on_complete); + grpc_closure *on_complete, grpc_call_list *call_list); static const grpc_resolver_vtable dns_resolver_vtable = { dns_destroy, dns_shutdown, dns_channel_saw_error, dns_next}; -static void dns_shutdown(grpc_resolver *resolver) { +static void dns_shutdown(grpc_resolver *resolver, grpc_call_list *call_list) { dns_resolver *r = (dns_resolver *)resolver; - grpc_closure *next_completion; gpr_mu_lock(&r->mu); - next_completion = r->next_completion; - r->next_completion = NULL; - gpr_mu_unlock(&r->mu); - if (next_completion != NULL) { + if (r->next_completion != NULL) { *r->target_config = NULL; - next_completion->cb(next_completion->cb_arg, 1); + grpc_call_list_add(call_list, r->next_completion, 1); + r->next_completion = NULL; } + gpr_mu_unlock(&r->mu); } static void dns_channel_saw_error(grpc_resolver *resolver, struct sockaddr *sa, - int len) { + int len, grpc_call_list *call_list) { dns_resolver *r = (dns_resolver *)resolver; gpr_mu_lock(&r->mu); if (!r->resolving) { @@ -117,9 +114,8 @@ static void dns_channel_saw_error(grpc_resolver *resolver, struct sockaddr *sa, static void dns_next(grpc_resolver *resolver, grpc_client_config **target_config, - grpc_closure *on_complete) { + grpc_closure *on_complete, grpc_call_list *call_list) { dns_resolver *r = (dns_resolver *)resolver; - grpc_closure *call = NULL; gpr_mu_lock(&r->mu); GPR_ASSERT(!r->next_completion); r->next_completion = on_complete; @@ -127,21 +123,18 @@ static void dns_next(grpc_resolver *resolver, if (r->resolved_version == 0 && !r->resolving) { dns_start_resolving_locked(r); } else { - call = dns_maybe_finish_next_locked(r); + dns_maybe_finish_next_locked(r, call_list); } gpr_mu_unlock(&r->mu); - if (call) { - call->cb(call->cb_arg, 1); - } } -static void dns_on_resolved(void *arg, grpc_resolved_addresses *addresses) { +static void dns_on_resolved(void *arg, grpc_resolved_addresses *addresses, + grpc_call_list *call_list) { dns_resolver *r = arg; grpc_client_config *config = NULL; grpc_subchannel **subchannels; grpc_subchannel_args args; grpc_lb_policy *lb_policy; - grpc_closure *call; size_t i; if (addresses) { grpc_lb_policy_args lb_policy_args; @@ -157,10 +150,9 @@ static void dns_on_resolved(void *arg, grpc_resolved_addresses *addresses) { memset(&lb_policy_args, 0, sizeof(lb_policy_args)); lb_policy_args.subchannels = subchannels; lb_policy_args.num_subchannels = addresses->naddrs; - lb_policy_args.workqueue = r->workqueue; lb_policy = grpc_lb_policy_create(r->lb_policy_name, &lb_policy_args); grpc_client_config_set_lb_policy(config, lb_policy); - GRPC_LB_POLICY_UNREF(lb_policy, "construction"); + GRPC_LB_POLICY_UNREF(lb_policy, "construction", call_list); grpc_resolved_addresses_destroy(addresses); gpr_free(subchannels); } @@ -168,17 +160,14 @@ static void dns_on_resolved(void *arg, grpc_resolved_addresses *addresses) { GPR_ASSERT(r->resolving); r->resolving = 0; if (r->resolved_config) { - grpc_client_config_unref(r->resolved_config); + grpc_client_config_unref(r->resolved_config, call_list); } r->resolved_config = config; r->resolved_version++; - call = dns_maybe_finish_next_locked(r); + dns_maybe_finish_next_locked(r, call_list); gpr_mu_unlock(&r->mu); - if (call) { - call->cb(call->cb_arg, 1); - } - GRPC_RESOLVER_UNREF(&r->base, "dns-resolving"); + GRPC_RESOLVER_UNREF(&r->base, "dns-resolving", call_list); } static void dns_start_resolving_locked(dns_resolver *r) { @@ -188,29 +177,27 @@ static void dns_start_resolving_locked(dns_resolver *r) { grpc_resolve_address(r->name, r->default_port, dns_on_resolved, r); } -static grpc_closure *dns_maybe_finish_next_locked(dns_resolver *r) { - grpc_closure *ret = NULL; +static void dns_maybe_finish_next_locked(dns_resolver *r, + grpc_call_list *call_list) { if (r->next_completion != NULL && r->resolved_version != r->published_version) { *r->target_config = r->resolved_config; if (r->resolved_config) { grpc_client_config_ref(r->resolved_config); } - ret = r->next_completion; + grpc_call_list_add(call_list, r->next_completion, 1); r->next_completion = NULL; r->published_version = r->resolved_version; } - return ret; } -static void dns_destroy(grpc_resolver *gr) { +static void dns_destroy(grpc_resolver *gr, grpc_call_list *call_list) { dns_resolver *r = (dns_resolver *)gr; gpr_mu_destroy(&r->mu); if (r->resolved_config) { - grpc_client_config_unref(r->resolved_config); + grpc_client_config_unref(r->resolved_config, call_list); } grpc_subchannel_factory_unref(r->subchannel_factory); - GRPC_WORKQUEUE_UNREF(r->workqueue, "dns"); gpr_free(r->name); gpr_free(r->default_port); gpr_free(r->lb_policy_name); @@ -239,8 +226,6 @@ static grpc_resolver *dns_create(grpc_resolver_args *args, r->default_port = gpr_strdup(default_port); r->subchannel_factory = args->subchannel_factory; grpc_subchannel_factory_ref(r->subchannel_factory); - r->workqueue = args->workqueue; - GRPC_WORKQUEUE_REF(r->workqueue, "dns"); r->lb_policy_name = gpr_strdup(lb_policy_name); return &r->base; } diff --git a/src/core/client_config/resolvers/sockaddr_resolver.c b/src/core/client_config/resolvers/sockaddr_resolver.c index 47fe5fad4a..15eb60b93a 100644 --- a/src/core/client_config/resolvers/sockaddr_resolver.c +++ b/src/core/client_config/resolvers/sockaddr_resolver.c @@ -56,8 +56,6 @@ typedef struct { gpr_refcount refs; /** subchannel factory */ grpc_subchannel_factory *subchannel_factory; - /** workqueue */ - grpc_workqueue *workqueue; /** load balancing policy name */ char *lb_policy_name; @@ -78,61 +76,59 @@ typedef struct { grpc_client_config **target_config; } sockaddr_resolver; -static void sockaddr_destroy(grpc_resolver *r); +static void sockaddr_destroy(grpc_resolver *r, grpc_call_list *call_list); -static grpc_closure *sockaddr_maybe_finish_next_locked(sockaddr_resolver *r) - GRPC_MUST_USE_RESULT; +static void sockaddr_maybe_finish_next_locked(sockaddr_resolver *r, + grpc_call_list *call_list); -static void sockaddr_shutdown(grpc_resolver *r); +static void sockaddr_shutdown(grpc_resolver *r, grpc_call_list *call_list); static void sockaddr_channel_saw_error(grpc_resolver *r, struct sockaddr *failing_address, - int failing_address_len); + int failing_address_len, + grpc_call_list *call_list); static void sockaddr_next(grpc_resolver *r, grpc_client_config **target_config, - grpc_closure *on_complete); + grpc_closure *on_complete, grpc_call_list *call_list); static const grpc_resolver_vtable sockaddr_resolver_vtable = { sockaddr_destroy, sockaddr_shutdown, sockaddr_channel_saw_error, sockaddr_next}; -static void sockaddr_shutdown(grpc_resolver *resolver) { +static void sockaddr_shutdown(grpc_resolver *resolver, + grpc_call_list *call_list) { sockaddr_resolver *r = (sockaddr_resolver *)resolver; - grpc_closure *call = NULL; gpr_mu_lock(&r->mu); if (r->next_completion != NULL) { *r->target_config = NULL; - call = r->next_completion; + grpc_call_list_add(call_list, r->next_completion, 1); r->next_completion = NULL; } gpr_mu_unlock(&r->mu); - if (call) { - call->cb(call->cb_arg, 1); - } } static void sockaddr_channel_saw_error(grpc_resolver *resolver, - struct sockaddr *sa, int len) {} + struct sockaddr *sa, int len, + grpc_call_list *call_list) {} static void sockaddr_next(grpc_resolver *resolver, grpc_client_config **target_config, - grpc_closure *on_complete) { + grpc_closure *on_complete, + grpc_call_list *call_list) { sockaddr_resolver *r = (sockaddr_resolver *)resolver; - grpc_closure *call = NULL; gpr_mu_lock(&r->mu); GPR_ASSERT(!r->next_completion); r->next_completion = on_complete; r->target_config = target_config; - call = sockaddr_maybe_finish_next_locked(r); + sockaddr_maybe_finish_next_locked(r, call_list); gpr_mu_unlock(&r->mu); - if (call) call->cb(call->cb_arg, 1); } -static grpc_closure *sockaddr_maybe_finish_next_locked(sockaddr_resolver *r) { +static void sockaddr_maybe_finish_next_locked(sockaddr_resolver *r, + grpc_call_list *call_list) { grpc_client_config *cfg; grpc_lb_policy *lb_policy; grpc_lb_policy_args lb_policy_args; grpc_subchannel **subchannels; grpc_subchannel_args args; - grpc_closure *call = NULL; if (r->next_completion != NULL && !r->published) { size_t i; @@ -148,26 +144,22 @@ static grpc_closure *sockaddr_maybe_finish_next_locked(sockaddr_resolver *r) { memset(&lb_policy_args, 0, sizeof(lb_policy_args)); lb_policy_args.subchannels = subchannels; lb_policy_args.num_subchannels = r->num_addrs; - lb_policy_args.workqueue = r->workqueue; lb_policy = grpc_lb_policy_create(r->lb_policy_name, &lb_policy_args); gpr_free(subchannels); grpc_client_config_set_lb_policy(cfg, lb_policy); - GRPC_LB_POLICY_UNREF(lb_policy, "unix"); + GRPC_LB_POLICY_UNREF(lb_policy, "sockaddr", call_list); r->published = 1; *r->target_config = cfg; - call = r->next_completion; + grpc_call_list_add(call_list, r->next_completion, 1); r->next_completion = NULL; } - - return call; } -static void sockaddr_destroy(grpc_resolver *gr) { +static void sockaddr_destroy(grpc_resolver *gr, grpc_call_list *call_list) { sockaddr_resolver *r = (sockaddr_resolver *)gr; gpr_mu_destroy(&r->mu); grpc_subchannel_factory_unref(r->subchannel_factory); - GRPC_WORKQUEUE_UNREF(r->workqueue, "sockaddr"); gpr_free(r->addrs); gpr_free(r->addrs_len); gpr_free(r->lb_policy_name); @@ -340,8 +332,6 @@ static grpc_resolver *sockaddr_create( grpc_resolver_init(&r->base, &sockaddr_resolver_vtable); r->subchannel_factory = args->subchannel_factory; grpc_subchannel_factory_ref(r->subchannel_factory); - r->workqueue = args->workqueue; - GRPC_WORKQUEUE_REF(r->workqueue, "sockaddr"); r->lb_policy_name = gpr_strdup(lb_policy_name); return &r->base; diff --git a/src/core/client_config/resolvers/zookeeper_resolver.c b/src/core/client_config/resolvers/zookeeper_resolver.c index 5632f0425e..c785e76939 100644 --- a/src/core/client_config/resolvers/zookeeper_resolver.c +++ b/src/core/client_config/resolvers/zookeeper_resolver.c @@ -61,8 +61,6 @@ typedef struct { grpc_subchannel_factory *subchannel_factory; /** load balancing policy name */ char *lb_policy_name; - /** work queue */ - grpc_workqueue *workqueue; /** mutex guarding the rest of the state */ gpr_mu mu; @@ -436,7 +434,6 @@ static void zookeeper_destroy(grpc_resolver *gr) { grpc_client_config_unref(r->resolved_config); } grpc_subchannel_factory_unref(r->subchannel_factory); - grpc_workqueue_unref(r->workqueue); gpr_free(r->name); gpr_free(r->lb_policy_name); gpr_free(r); @@ -466,9 +463,6 @@ static grpc_resolver *zookeeper_create(grpc_resolver_args *args, grpc_resolver_init(&r->base, &zookeeper_resolver_vtable); r->name = gpr_strdup(path); - r->workqueue = args->workqueue; - grpc_workqueue_ref(r->workqueue); - r->subchannel_factory = args->subchannel_factory; grpc_subchannel_factory_ref(r->subchannel_factory); diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c index 99310e02e5..937b3cd71c 100644 --- a/src/core/client_config/subchannel.c +++ b/src/core/client_config/subchannel.c @@ -76,7 +76,6 @@ typedef struct waiting_for_connect { struct grpc_subchannel { grpc_connector *connector; - grpc_workqueue *workqueue; /** non-transport related channel filters */ const grpc_channel_filter **filters; @@ -150,7 +149,8 @@ static void connectivity_state_changed_locked(grpc_subchannel *c, grpc_call_list *call_list); static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c); static gpr_timespec compute_connect_deadline(grpc_subchannel *c); -static void subchannel_connected(void *subchannel, int iomgr_success); +static void subchannel_connected(void *subchannel, int iomgr_success, + grpc_call_list *call_list); static void subchannel_ref_locked( grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS); @@ -158,8 +158,9 @@ static int subchannel_unref_locked( grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT; static void connection_ref_locked(connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS); static grpc_subchannel *connection_unref_locked( - connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT; -static void subchannel_destroy(grpc_subchannel *c); + connection *c, grpc_call_list *call_list GRPC_SUBCHANNEL_REF_EXTRA_ARGS) + GRPC_MUST_USE_RESULT; +static void subchannel_destroy(grpc_subchannel *c, grpc_call_list *call_list); #ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG #define SUBCHANNEL_REF_LOCKED(p, r) \ @@ -168,8 +169,8 @@ static void subchannel_destroy(grpc_subchannel *c); subchannel_unref_locked((p), __FILE__, __LINE__, (r)) #define CONNECTION_REF_LOCKED(p, r) \ connection_ref_locked((p), __FILE__, __LINE__, (r)) -#define CONNECTION_UNREF_LOCKED(p, r) \ - connection_unref_locked((p), __FILE__, __LINE__, (r)) +#define CONNECTION_UNREF_LOCKED(p, r, cl) \ + connection_unref_locked((p), (cl), __FILE__, __LINE__, (r)) #define REF_PASS_ARGS , file, line, reason #define REF_LOG(name, p) \ gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p ref %d -> %d %s", \ @@ -181,7 +182,7 @@ static void subchannel_destroy(grpc_subchannel *c); #define SUBCHANNEL_REF_LOCKED(p, r) subchannel_ref_locked((p)) #define SUBCHANNEL_UNREF_LOCKED(p, r) subchannel_unref_locked((p)) #define CONNECTION_REF_LOCKED(p, r) connection_ref_locked((p)) -#define CONNECTION_UNREF_LOCKED(p, r) connection_unref_locked((p)) +#define CONNECTION_UNREF_LOCKED(p, r, cl) connection_unref_locked((p), (cl)) #define REF_PASS_ARGS #define REF_LOG(name, p) \ do { \ @@ -195,9 +196,9 @@ static void subchannel_destroy(grpc_subchannel *c); * connection implementation */ -static void connection_destroy(connection *c) { +static void connection_destroy(connection *c, grpc_call_list *call_list) { GPR_ASSERT(c->refs == 0); - grpc_channel_stack_destroy(CHANNEL_STACK_FROM_CONNECTION(c)); + grpc_channel_stack_destroy(CHANNEL_STACK_FROM_CONNECTION(c), call_list); gpr_free(c); } @@ -209,14 +210,14 @@ static void connection_ref_locked( } static grpc_subchannel *connection_unref_locked( - connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + connection *c, grpc_call_list *call_list GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { grpc_subchannel *destroy = NULL; UNREF_LOG("CONNECTION", c); if (subchannel_unref_locked(c->subchannel REF_PASS_ARGS)) { destroy = c->subchannel; } if (--c->refs == 0 && c->subchannel->active != c) { - connection_destroy(c); + connection_destroy(c, call_list); } return destroy; } @@ -243,17 +244,19 @@ void grpc_subchannel_ref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { gpr_mu_unlock(&c->mu); } -void grpc_subchannel_unref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { +void grpc_subchannel_unref(grpc_subchannel *c, + grpc_call_list *call_list + GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { int destroy; gpr_mu_lock(&c->mu); destroy = subchannel_unref_locked(c REF_PASS_ARGS); gpr_mu_unlock(&c->mu); - if (destroy) subchannel_destroy(c); + if (destroy) subchannel_destroy(c, call_list); } -static void subchannel_destroy(grpc_subchannel *c) { +static void subchannel_destroy(grpc_subchannel *c, grpc_call_list *call_list) { if (c->active != NULL) { - connection_destroy(c->active); + connection_destroy(c->active, call_list); } gpr_free(c->filters); grpc_channel_args_destroy(c->args); @@ -261,18 +264,19 @@ static void subchannel_destroy(grpc_subchannel *c) { grpc_mdctx_unref(c->mdctx); grpc_connectivity_state_destroy(&c->state_tracker); grpc_connector_unref(c->connector); - GRPC_WORKQUEUE_UNREF(c->workqueue, "subchannel"); gpr_free(c); } void grpc_subchannel_add_interested_party(grpc_subchannel *c, - grpc_pollset *pollset) { - grpc_pollset_set_add_pollset(c->pollset_set, pollset); + grpc_pollset *pollset, + grpc_call_list *call_list) { + grpc_pollset_set_add_pollset(c->pollset_set, pollset, call_list); } void grpc_subchannel_del_interested_party(grpc_subchannel *c, - grpc_pollset *pollset) { - grpc_pollset_set_del_pollset(c->pollset_set, pollset); + grpc_pollset *pollset, + grpc_call_list *call_list) { + grpc_pollset_set_del_pollset(c->pollset_set, pollset, call_list); } static gpr_uint32 random_seed() { @@ -298,8 +302,6 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector, c->args = grpc_channel_args_copy(args->args); c->mdctx = args->mdctx; c->master = args->master; - c->workqueue = grpc_channel_get_workqueue(c->master); - GRPC_WORKQUEUE_REF(c->workqueue, "subchannel"); c->pollset_set = grpc_client_channel_get_connecting_pollset_set(parent_elem); c->random = random_seed(); grpc_mdctx_ref(c->mdctx); @@ -310,7 +312,7 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector, return c; } -static void continue_connect(grpc_subchannel *c) { +static void continue_connect(grpc_subchannel *c, grpc_call_list *call_list) { grpc_connect_in_args args; args.interested_parties = c->pollset_set; @@ -321,24 +323,25 @@ static void continue_connect(grpc_subchannel *c) { args.metadata_context = c->mdctx; grpc_connector_connect(c->connector, &args, &c->connecting_result, - &c->connected); + &c->connected, call_list); } -static void start_connect(grpc_subchannel *c) { +static void start_connect(grpc_subchannel *c, grpc_call_list *call_list) { c->backoff_delta = gpr_time_from_seconds( GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS, GPR_TIMESPAN); c->next_attempt = gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), c->backoff_delta); - continue_connect(c); + continue_connect(c, call_list); } -static void continue_creating_call(void *arg, int iomgr_success) { +static void continue_creating_call(void *arg, int iomgr_success, + grpc_call_list *call_list) { waiting_for_connect *w4c = arg; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; - grpc_subchannel_del_interested_party(w4c->subchannel, w4c->pollset); + grpc_subchannel_del_interested_party(w4c->subchannel, w4c->pollset, + call_list); grpc_subchannel_create_call(w4c->subchannel, w4c->pollset, w4c->target, - w4c->notify, &call_list); - GRPC_SUBCHANNEL_UNREF(w4c->subchannel, "waiting_for_connect"); + w4c->notify, call_list); + GRPC_SUBCHANNEL_UNREF(w4c->subchannel, "waiting_for_connect", call_list); gpr_free(w4c); } @@ -354,7 +357,7 @@ void grpc_subchannel_create_call(grpc_subchannel *c, grpc_pollset *pollset, gpr_mu_unlock(&c->mu); *target = create_call(con); - notify->cb(notify->cb_arg, 1); + notify->cb(notify->cb_arg, 1, call_list); } else { waiting_for_connect *w4c = gpr_malloc(sizeof(*w4c)); w4c->next = c->waiting; @@ -366,7 +369,7 @@ void grpc_subchannel_create_call(grpc_subchannel *c, grpc_pollset *pollset, SUBCHANNEL_REF_LOCKED(c, "waiting_for_connect"); grpc_closure_init(&w4c->continuation, continue_creating_call, w4c); c->waiting = w4c; - grpc_subchannel_add_interested_party(c, pollset); + grpc_subchannel_add_interested_party(c, pollset, call_list); if (!c->connecting) { c->connecting = 1; connectivity_state_changed_locked(c, "create_call", call_list); @@ -375,7 +378,7 @@ void grpc_subchannel_create_call(grpc_subchannel *c, grpc_pollset *pollset, GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting"); gpr_mu_unlock(&c->mu); - start_connect(c); + start_connect(c, call_list); } else { gpr_mu_unlock(&c->mu); } @@ -408,16 +411,16 @@ void grpc_subchannel_notify_on_state_change(grpc_subchannel *c, gpr_mu_unlock(&c->mu); if (do_connect) { - start_connect(c); + start_connect(c, call_list); } } void grpc_subchannel_process_transport_op(grpc_subchannel *c, - grpc_transport_op *op) { + grpc_transport_op *op, + grpc_call_list *call_list) { connection *con = NULL; grpc_subchannel *destroy; int cancel_alarm = 0; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; gpr_mu_lock(&c->mu); if (c->active != NULL) { con = c->active; @@ -425,7 +428,7 @@ void grpc_subchannel_process_transport_op(grpc_subchannel *c, } if (op->disconnect) { c->disconnected = 1; - connectivity_state_changed_locked(c, "disconnect", &call_list); + connectivity_state_changed_locked(c, "disconnect", call_list); if (c->have_alarm) { cancel_alarm = 1; } @@ -436,28 +439,27 @@ void grpc_subchannel_process_transport_op(grpc_subchannel *c, grpc_channel_stack *channel_stack = CHANNEL_STACK_FROM_CONNECTION(con); grpc_channel_element *top_elem = grpc_channel_stack_element(channel_stack, 0); - top_elem->filter->start_transport_op(top_elem, op); + top_elem->filter->start_transport_op(top_elem, op, call_list); gpr_mu_lock(&c->mu); - destroy = CONNECTION_UNREF_LOCKED(con, "transport-op"); + destroy = CONNECTION_UNREF_LOCKED(con, "transport-op", call_list); gpr_mu_unlock(&c->mu); if (destroy) { - subchannel_destroy(destroy); + subchannel_destroy(destroy, call_list); } } if (cancel_alarm) { - grpc_alarm_cancel(&c->alarm); + grpc_alarm_cancel(&c->alarm, call_list); } if (op->disconnect) { - grpc_connector_shutdown(c->connector); + grpc_connector_shutdown(c->connector, call_list); } - - grpc_call_list_run(call_list); } -static void on_state_changed(void *p, int iomgr_success) { +static void on_state_changed(void *p, int iomgr_success, + grpc_call_list *call_list) { state_watcher *sw = p; grpc_subchannel *c = sw->subchannel; gpr_mu *mu = &c->mu; @@ -465,7 +467,6 @@ static void on_state_changed(void *p, int iomgr_success) { grpc_transport_op op; grpc_channel_element *elem; connection *destroy_connection = NULL; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; gpr_mu_lock(mu); @@ -485,7 +486,7 @@ static void on_state_changed(void *p, int iomgr_success) { op.on_connectivity_state_change = &sw->closure; elem = grpc_channel_stack_element( CHANNEL_STACK_FROM_CONNECTION(c->active), 0); - elem->filter->start_transport_op(elem, &op); + elem->filter->start_transport_op(elem, &op, call_list); /* early out */ gpr_mu_unlock(mu); return; @@ -499,22 +500,21 @@ static void on_state_changed(void *p, int iomgr_success) { grpc_connectivity_state_set( &c->state_tracker, c->disconnected ? GRPC_CHANNEL_FATAL_FAILURE : GRPC_CHANNEL_TRANSIENT_FAILURE, - "connection_failed", &call_list); + "connection_failed", call_list); break; } done: - connectivity_state_changed_locked(c, "transport_state_changed", &call_list); + connectivity_state_changed_locked(c, "transport_state_changed", call_list); destroy = SUBCHANNEL_UNREF_LOCKED(c, "state_watcher"); gpr_free(sw); gpr_mu_unlock(mu); if (destroy) { - subchannel_destroy(c); + subchannel_destroy(c, call_list); } if (destroy_connection != NULL) { - connection_destroy(destroy_connection); + connection_destroy(destroy_connection, call_list); } - grpc_call_list_run(call_list); } static void publish_transport(grpc_subchannel *c, grpc_call_list *call_list) { @@ -544,7 +544,7 @@ static void publish_transport(grpc_subchannel *c, grpc_call_list *call_list) { con->refs = 0; con->subchannel = c; grpc_channel_stack_init(filters, num_filters, c->master, c->args, c->mdctx, - stk); + stk, call_list); grpc_connected_channel_bind_transport(stk, c->connecting_result.transport); gpr_free(c->connecting_result.filters); memset(&c->connecting_result, 0, sizeof(c->connecting_result)); @@ -561,9 +561,9 @@ static void publish_transport(grpc_subchannel *c, grpc_call_list *call_list) { gpr_mu_unlock(&c->mu); gpr_free(sw); gpr_free(filters); - grpc_channel_stack_destroy(stk); + grpc_channel_stack_destroy(stk, call_list); GRPC_CHANNEL_INTERNAL_UNREF(c->master, "connecting"); - GRPC_SUBCHANNEL_UNREF(c, "connecting"); + GRPC_SUBCHANNEL_UNREF(c, "connecting", call_list); return; } @@ -587,7 +587,7 @@ static void publish_transport(grpc_subchannel *c, grpc_call_list *call_list) { GPR_ASSERT(!SUBCHANNEL_UNREF_LOCKED(c, "connecting")); elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(c->active), 0); - elem->filter->start_transport_op(elem, &op); + elem->filter->start_transport_op(elem, &op, call_list); /* signal completion */ connectivity_state_changed_locked(c, "connected", call_list); @@ -605,7 +605,7 @@ static void publish_transport(grpc_subchannel *c, grpc_call_list *call_list) { gpr_free(filters); if (destroy_connection != NULL) { - connection_destroy(destroy_connection); + connection_destroy(destroy_connection, call_list); } } @@ -638,41 +638,38 @@ static void update_reconnect_parameters(grpc_subchannel *c) { gpr_time_add(c->next_attempt, gpr_time_from_millis(jitter, GPR_TIMESPAN)); } -static void on_alarm(void *arg, int iomgr_success) { +static void on_alarm(void *arg, int iomgr_success, grpc_call_list *call_list) { grpc_subchannel *c = arg; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; gpr_mu_lock(&c->mu); c->have_alarm = 0; if (c->disconnected) { iomgr_success = 0; } - connectivity_state_changed_locked(c, "alarm", &call_list); + connectivity_state_changed_locked(c, "alarm", call_list); gpr_mu_unlock(&c->mu); if (iomgr_success) { update_reconnect_parameters(c); - continue_connect(c); + continue_connect(c, call_list); } else { GRPC_CHANNEL_INTERNAL_UNREF(c->master, "connecting"); - GRPC_SUBCHANNEL_UNREF(c, "connecting"); + GRPC_SUBCHANNEL_UNREF(c, "connecting", call_list); } - grpc_call_list_run(call_list); } -static void subchannel_connected(void *arg, int iomgr_success) { +static void subchannel_connected(void *arg, int iomgr_success, + grpc_call_list *call_list) { grpc_subchannel *c = arg; - grpc_call_list call_list = GRPC_CALL_LIST_INIT; if (c->connecting_result.transport != NULL) { - publish_transport(c, &call_list); + publish_transport(c, call_list); } else { gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); gpr_mu_lock(&c->mu); GPR_ASSERT(!c->have_alarm); c->have_alarm = 1; - connectivity_state_changed_locked(c, "connect_failed", &call_list); - grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, now); + connectivity_state_changed_locked(c, "connect_failed", call_list); + grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, now, call_list); gpr_mu_unlock(&c->mu); } - grpc_call_list_run(call_list); } static gpr_timespec compute_connect_deadline(grpc_subchannel *c) { @@ -718,33 +715,36 @@ void grpc_subchannel_call_ref( gpr_ref(&c->refs); } -void grpc_subchannel_call_unref( - grpc_subchannel_call *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { +void grpc_subchannel_call_unref(grpc_subchannel_call *c, + grpc_call_list *call_list + GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { if (gpr_unref(&c->refs)) { gpr_mu *mu = &c->connection->subchannel->mu; grpc_subchannel *destroy; - grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(c)); + grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(c), call_list); gpr_mu_lock(mu); - destroy = CONNECTION_UNREF_LOCKED(c->connection, "call"); + destroy = CONNECTION_UNREF_LOCKED(c->connection, "call", call_list); gpr_mu_unlock(mu); gpr_free(c); if (destroy != NULL) { - subchannel_destroy(destroy); + subchannel_destroy(destroy, call_list); } } } -char *grpc_subchannel_call_get_peer(grpc_subchannel_call *call) { +char *grpc_subchannel_call_get_peer(grpc_subchannel_call *call, + grpc_call_list *call_list) { grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call); grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0); - return top_elem->filter->get_peer(top_elem); + return top_elem->filter->get_peer(top_elem, call_list); } void grpc_subchannel_call_process_op(grpc_subchannel_call *call, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call); grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0); - top_elem->filter->start_transport_stream_op(top_elem, op); + top_elem->filter->start_transport_stream_op(top_elem, op, call_list); } grpc_subchannel_call *create_call(connection *con) { diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h index 189e531d35..0ebfc7c29a 100644 --- a/src/core/client_config/subchannel.h +++ b/src/core/client_config/subchannel.h @@ -47,30 +47,33 @@ typedef struct grpc_subchannel_args grpc_subchannel_args; #ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG #define GRPC_SUBCHANNEL_REF(p, r) \ grpc_subchannel_ref((p), __FILE__, __LINE__, (r)) -#define GRPC_SUBCHANNEL_UNREF(p, r) \ - grpc_subchannel_unref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_UNREF(p, r, cl) \ + grpc_subchannel_unref((p), (cl), __FILE__, __LINE__, (r)) #define GRPC_SUBCHANNEL_CALL_REF(p, r) \ grpc_subchannel_call_ref((p), __FILE__, __LINE__, (r)) -#define GRPC_SUBCHANNEL_CALL_UNREF(p, r) \ - grpc_subchannel_call_unref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_CALL_UNREF(p, r, cl) \ + grpc_subchannel_call_unref((p), (cl), __FILE__, __LINE__, (r)) #define GRPC_SUBCHANNEL_REF_EXTRA_ARGS \ , const char *file, int line, const char *reason #else #define GRPC_SUBCHANNEL_REF(p, r) grpc_subchannel_ref((p)) -#define GRPC_SUBCHANNEL_UNREF(p, r) grpc_subchannel_unref((p)) +#define GRPC_SUBCHANNEL_UNREF(p, r, cl) grpc_subchannel_unref((p), (cl)) #define GRPC_SUBCHANNEL_CALL_REF(p, r) grpc_subchannel_call_ref((p)) -#define GRPC_SUBCHANNEL_CALL_UNREF(p, r) grpc_subchannel_call_unref((p)) +#define GRPC_SUBCHANNEL_CALL_UNREF(p, r, cl) \ + grpc_subchannel_call_unref((p), (cl)) #define GRPC_SUBCHANNEL_REF_EXTRA_ARGS #endif void grpc_subchannel_ref( grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS); -void grpc_subchannel_unref( - grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_unref(grpc_subchannel *channel, + grpc_call_list *call_list + GRPC_SUBCHANNEL_REF_EXTRA_ARGS); void grpc_subchannel_call_ref( grpc_subchannel_call *call GRPC_SUBCHANNEL_REF_EXTRA_ARGS); -void grpc_subchannel_call_unref( - grpc_subchannel_call *call GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_call_unref(grpc_subchannel_call *call, + grpc_call_list *call_list + GRPC_SUBCHANNEL_REF_EXTRA_ARGS); /** construct a call (possibly asynchronously) */ void grpc_subchannel_create_call(grpc_subchannel *subchannel, @@ -81,7 +84,8 @@ void grpc_subchannel_create_call(grpc_subchannel *subchannel, /** process a transport level op */ void grpc_subchannel_process_transport_op(grpc_subchannel *subchannel, - grpc_transport_op *op); + grpc_transport_op *op, + grpc_call_list *call_list); /** poll the current connectivity state of a channel */ grpc_connectivity_state grpc_subchannel_check_connectivity( @@ -96,17 +100,21 @@ void grpc_subchannel_notify_on_state_change(grpc_subchannel *channel, /** express interest in \a channel's activities through \a pollset. */ void grpc_subchannel_add_interested_party(grpc_subchannel *channel, - grpc_pollset *pollset); + grpc_pollset *pollset, + grpc_call_list *call_list); /** stop following \a channel's activity through \a pollset. */ void grpc_subchannel_del_interested_party(grpc_subchannel *channel, - grpc_pollset *pollset); + grpc_pollset *pollset, + grpc_call_list *call_list); /** continue processing a transport op */ void grpc_subchannel_call_process_op(grpc_subchannel_call *subchannel_call, - grpc_transport_stream_op *op); + grpc_transport_stream_op *op, + grpc_call_list *call_list); /** continue querying for peer */ -char *grpc_subchannel_call_get_peer(grpc_subchannel_call *subchannel_call); +char *grpc_subchannel_call_get_peer(grpc_subchannel_call *subchannel_call, + grpc_call_list *call_list); struct grpc_subchannel_args { /** Channel filters for this channel - wrapped factories will likely diff --git a/src/core/client_config/subchannel_factory.h b/src/core/client_config/subchannel_factory.h index d7eae1c964..b588580edb 100644 --- a/src/core/client_config/subchannel_factory.h +++ b/src/core/client_config/subchannel_factory.h @@ -50,7 +50,8 @@ struct grpc_subchannel_factory_vtable { void (*ref)(grpc_subchannel_factory *factory); void (*unref)(grpc_subchannel_factory *factory); grpc_subchannel *(*create_subchannel)(grpc_subchannel_factory *factory, - grpc_subchannel_args *args); + grpc_subchannel_args *args, + grpc_call_list *call_list); }; void grpc_subchannel_factory_ref(grpc_subchannel_factory *factory); @@ -58,6 +59,7 @@ void grpc_subchannel_factory_unref(grpc_subchannel_factory *factory); /** Create a new grpc_subchannel */ grpc_subchannel *grpc_subchannel_factory_create_subchannel( - grpc_subchannel_factory *factory, grpc_subchannel_args *args); + grpc_subchannel_factory *factory, grpc_subchannel_args *args, + grpc_call_list *call_list); #endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_H */ diff --git a/src/core/httpcli/httpcli.c b/src/core/httpcli/httpcli.c index c484f4c7b3..e35fa20953 100644 --- a/src/core/httpcli/httpcli.c +++ b/src/core/httpcli/httpcli.c @@ -65,6 +65,7 @@ typedef struct { gpr_slice_buffer outgoing; grpc_closure on_read; grpc_closure done_write; + grpc_closure connected; grpc_workqueue *workqueue; } internal_request; @@ -74,8 +75,10 @@ static grpc_httpcli_post_override g_post_override = NULL; static void plaintext_handshake(void *arg, grpc_endpoint *endpoint, const char *host, void (*on_done)(void *arg, - grpc_endpoint *endpoint)) { - on_done(arg, endpoint); + grpc_endpoint *endpoint, + grpc_call_list *call_list), + grpc_call_list *call_list) { + on_done(arg, endpoint, call_list); } const grpc_httpcli_handshaker grpc_httpcli_plaintext = {"http", @@ -89,17 +92,19 @@ void grpc_httpcli_context_destroy(grpc_httpcli_context *context) { grpc_pollset_set_destroy(&context->pollset_set); } -static void next_address(internal_request *req); +static void next_address(internal_request *req, grpc_call_list *call_list); -static void finish(internal_request *req, int success) { - grpc_pollset_set_del_pollset(&req->context->pollset_set, req->pollset); - req->on_response(req->user_data, success ? &req->parser.r : NULL); +static void finish(internal_request *req, int success, + grpc_call_list *call_list) { + grpc_pollset_set_del_pollset(&req->context->pollset_set, req->pollset, + call_list); + req->on_response(req->user_data, success ? &req->parser.r : NULL, call_list); grpc_httpcli_parser_destroy(&req->parser); if (req->addresses != NULL) { grpc_resolved_addresses_destroy(req->addresses); } if (req->ep != NULL) { - grpc_endpoint_destroy(req->ep); + grpc_endpoint_destroy(req->ep, call_list); } gpr_slice_unref(req->request_text); gpr_free(req->host); @@ -110,22 +115,13 @@ static void finish(internal_request *req, int success) { gpr_free(req); } -static void on_read(void *user_data, int success); +static void on_read(void *user_data, int success, grpc_call_list *call_list); -static void do_read(internal_request *req) { - switch (grpc_endpoint_read(req->ep, &req->incoming, &req->on_read)) { - case GRPC_ENDPOINT_DONE: - on_read(req, 1); - break; - case GRPC_ENDPOINT_PENDING: - break; - case GRPC_ENDPOINT_ERROR: - on_read(req, 0); - break; - } +static void do_read(internal_request *req, grpc_call_list *call_list) { + grpc_endpoint_read(req->ep, &req->incoming, &req->on_read, call_list); } -static void on_read(void *user_data, int success) { +static void on_read(void *user_data, int success, grpc_call_list *call_list) { internal_request *req = user_data; size_t i; @@ -133,99 +129,94 @@ static void on_read(void *user_data, int success) { if (GPR_SLICE_LENGTH(req->incoming.slices[i])) { req->have_read_byte = 1; if (!grpc_httpcli_parser_parse(&req->parser, req->incoming.slices[i])) { - finish(req, 0); + finish(req, 0, call_list); return; } } } if (success) { - do_read(req); + do_read(req, call_list); } else if (!req->have_read_byte) { - next_address(req); + next_address(req, call_list); } else { - finish(req, grpc_httpcli_parser_eof(&req->parser)); + finish(req, grpc_httpcli_parser_eof(&req->parser), call_list); } } -static void on_written(internal_request *req) { do_read(req); } +static void on_written(internal_request *req, grpc_call_list *call_list) { + do_read(req, call_list); +} -static void done_write(void *arg, int success) { +static void done_write(void *arg, int success, grpc_call_list *call_list) { internal_request *req = arg; if (success) { - on_written(req); + on_written(req, call_list); } else { - next_address(req); + next_address(req, call_list); } } -static void start_write(internal_request *req) { +static void start_write(internal_request *req, grpc_call_list *call_list) { gpr_slice_ref(req->request_text); gpr_slice_buffer_add(&req->outgoing, req->request_text); - switch (grpc_endpoint_write(req->ep, &req->outgoing, &req->done_write)) { - case GRPC_ENDPOINT_DONE: - on_written(req); - break; - case GRPC_ENDPOINT_PENDING: - break; - case GRPC_ENDPOINT_ERROR: - finish(req, 0); - break; - } + grpc_endpoint_write(req->ep, &req->outgoing, &req->done_write, call_list); } -static void on_handshake_done(void *arg, grpc_endpoint *ep) { +static void on_handshake_done(void *arg, grpc_endpoint *ep, + grpc_call_list *call_list) { internal_request *req = arg; if (!ep) { - next_address(req); + next_address(req, call_list); return; } req->ep = ep; - start_write(req); + start_write(req, call_list); } -static void on_connected(void *arg, grpc_endpoint *tcp) { +static void on_connected(void *arg, int success, grpc_call_list *call_list) { internal_request *req = arg; - if (!tcp) { - next_address(req); + if (!req->ep) { + next_address(req, call_list); return; } - req->handshaker->handshake(req, tcp, req->host, on_handshake_done); + req->handshaker->handshake(req, req->ep, req->host, on_handshake_done, + call_list); } -static void next_address(internal_request *req) { +static void next_address(internal_request *req, grpc_call_list *call_list) { grpc_resolved_address *addr; if (req->next_address == req->addresses->naddrs) { - finish(req, 0); + finish(req, 0, call_list); return; } addr = &req->addresses->addrs[req->next_address++]; - grpc_tcp_client_connect(on_connected, req, &req->context->pollset_set, + grpc_closure_init(&req->connected, on_connected, req); + grpc_tcp_client_connect(&req->connected, &req->ep, &req->context->pollset_set, req->workqueue, (struct sockaddr *)&addr->addr, - addr->len, req->deadline); + addr->len, req->deadline, call_list); } -static void on_resolved(void *arg, grpc_resolved_addresses *addresses) { +static void on_resolved(void *arg, grpc_resolved_addresses *addresses, + grpc_call_list *call_list) { internal_request *req = arg; if (!addresses) { - finish(req, 0); + finish(req, 0, call_list); return; } req->addresses = addresses; req->next_address = 0; - next_address(req); + next_address(req, call_list); } -static void internal_request_begin(grpc_httpcli_context *context, - grpc_pollset *pollset, - const grpc_httpcli_request *request, - gpr_timespec deadline, - grpc_httpcli_response_cb on_response, - void *user_data, const char *name, - gpr_slice request_text) { +static void internal_request_begin( + grpc_httpcli_context *context, grpc_pollset *pollset, + const grpc_httpcli_request *request, gpr_timespec deadline, + grpc_httpcli_response_cb on_response, void *user_data, const char *name, + gpr_slice request_text, grpc_call_list *call_list) { internal_request *req = gpr_malloc(sizeof(internal_request)); memset(req, 0, sizeof(*req)); req->request_text = request_text; @@ -243,10 +234,11 @@ static void internal_request_begin(grpc_httpcli_context *context, gpr_slice_buffer_init(&req->outgoing); grpc_iomgr_register_object(&req->iomgr_obj, name); req->host = gpr_strdup(request->host); - req->workqueue = grpc_workqueue_create(); + req->workqueue = grpc_workqueue_create(call_list); grpc_workqueue_add_to_pollset(req->workqueue, pollset); - grpc_pollset_set_add_pollset(&req->context->pollset_set, req->pollset); + grpc_pollset_set_add_pollset(&req->context->pollset_set, req->pollset, + call_list); grpc_resolve_address(request->host, req->handshaker->default_port, on_resolved, req); } @@ -254,7 +246,8 @@ static void internal_request_begin(grpc_httpcli_context *context, void grpc_httpcli_get(grpc_httpcli_context *context, grpc_pollset *pollset, const grpc_httpcli_request *request, gpr_timespec deadline, - grpc_httpcli_response_cb on_response, void *user_data) { + grpc_httpcli_response_cb on_response, void *user_data, + grpc_call_list *call_list) { char *name; if (g_get_override && g_get_override(request, deadline, on_response, user_data)) { @@ -263,7 +256,7 @@ void grpc_httpcli_get(grpc_httpcli_context *context, grpc_pollset *pollset, gpr_asprintf(&name, "HTTP:GET:%s:%s", request->host, request->path); internal_request_begin(context, pollset, request, deadline, on_response, user_data, name, - grpc_httpcli_format_get_request(request)); + grpc_httpcli_format_get_request(request), call_list); gpr_free(name); } @@ -271,7 +264,8 @@ void grpc_httpcli_post(grpc_httpcli_context *context, grpc_pollset *pollset, 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) { + grpc_httpcli_response_cb on_response, void *user_data, + grpc_call_list *call_list) { char *name; if (g_post_override && g_post_override(request, body_bytes, body_size, deadline, on_response, user_data)) { @@ -280,7 +274,8 @@ void grpc_httpcli_post(grpc_httpcli_context *context, grpc_pollset *pollset, gpr_asprintf(&name, "HTTP:POST:%s:%s", request->host, request->path); internal_request_begin( context, pollset, request, deadline, on_response, user_data, name, - grpc_httpcli_format_post_request(request, body_bytes, body_size)); + grpc_httpcli_format_post_request(request, body_bytes, body_size), + call_list); gpr_free(name); } diff --git a/src/core/httpcli/httpcli.h b/src/core/httpcli/httpcli.h index c45966714c..74bb123042 100644 --- a/src/core/httpcli/httpcli.h +++ b/src/core/httpcli/httpcli.h @@ -62,7 +62,9 @@ typedef struct grpc_httpcli_context { typedef struct { const char *default_port; void (*handshake)(void *arg, grpc_endpoint *endpoint, const char *host, - void (*on_done)(void *arg, grpc_endpoint *endpoint)); + void (*on_done)(void *arg, grpc_endpoint *endpoint, + grpc_call_list *call_list), + grpc_call_list *call_list); } grpc_httpcli_handshaker; extern const grpc_httpcli_handshaker grpc_httpcli_plaintext; @@ -97,7 +99,8 @@ typedef struct grpc_httpcli_response { /* Callback for grpc_httpcli_get and grpc_httpcli_post. */ typedef void (*grpc_httpcli_response_cb)(void *user_data, - const grpc_httpcli_response *response); + const grpc_httpcli_response *response, + grpc_call_list *call_list); void grpc_httpcli_context_init(grpc_httpcli_context *context); void grpc_httpcli_context_destroy(grpc_httpcli_context *context); @@ -115,7 +118,8 @@ void grpc_httpcli_context_destroy(grpc_httpcli_context *context); void grpc_httpcli_get(grpc_httpcli_context *context, grpc_pollset *pollset, const grpc_httpcli_request *request, gpr_timespec deadline, - grpc_httpcli_response_cb on_response, void *user_data); + grpc_httpcli_response_cb on_response, void *user_data, + grpc_call_list *call_list); /* Asynchronously perform a HTTP POST. 'context' specifies the http context under which to do the post @@ -136,7 +140,8 @@ void grpc_httpcli_post(grpc_httpcli_context *context, grpc_pollset *pollset, 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); + grpc_httpcli_response_cb on_response, void *user_data, + grpc_call_list *call_list); /* override functions return 1 if they handled the request, 0 otherwise */ typedef int (*grpc_httpcli_get_override)(const grpc_httpcli_request *request, diff --git a/src/core/httpcli/httpcli_security_connector.c b/src/core/httpcli/httpcli_security_connector.c index 7887f9d530..a11fcd3e62 100644 --- a/src/core/httpcli/httpcli_security_connector.c +++ b/src/core/httpcli/httpcli_security_connector.c @@ -134,33 +134,36 @@ static grpc_security_status httpcli_ssl_channel_security_connector_create( /* handshaker */ typedef struct { - void (*func)(void *arg, grpc_endpoint *endpoint); + void (*func)(void *arg, grpc_endpoint *endpoint, grpc_call_list *call_list); void *arg; } on_done_closure; static void on_secure_transport_setup_done(void *rp, grpc_security_status status, grpc_endpoint *wrapped_endpoint, - grpc_endpoint *secure_endpoint) { + grpc_endpoint *secure_endpoint, + grpc_call_list *call_list) { on_done_closure *c = rp; if (status != GRPC_SECURITY_OK) { gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status); - c->func(c->arg, NULL); + c->func(c->arg, NULL, call_list); } else { - c->func(c->arg, secure_endpoint); + c->func(c->arg, secure_endpoint, call_list); } gpr_free(c); } static void ssl_handshake(void *arg, grpc_endpoint *tcp, const char *host, - void (*on_done)(void *arg, grpc_endpoint *endpoint)) { + void (*on_done)(void *arg, grpc_endpoint *endpoint, + grpc_call_list *call_list), + grpc_call_list *call_list) { grpc_channel_security_connector *sc = NULL; const unsigned char *pem_root_certs = NULL; on_done_closure *c = gpr_malloc(sizeof(*c)); size_t pem_root_certs_size = grpc_get_default_ssl_roots(&pem_root_certs); if (pem_root_certs == NULL || pem_root_certs_size == 0) { gpr_log(GPR_ERROR, "Could not get default pem root certs."); - on_done(arg, NULL); + on_done(arg, NULL, call_list); gpr_free(c); return; } @@ -169,8 +172,8 @@ static void ssl_handshake(void *arg, grpc_endpoint *tcp, const char *host, GPR_ASSERT(httpcli_ssl_channel_security_connector_create( pem_root_certs, pem_root_certs_size, host, &sc) == GRPC_SECURITY_OK); - grpc_setup_secure_transport(&sc->base, tcp, on_secure_transport_setup_done, - c); + grpc_setup_secure_transport(&sc->base, tcp, on_secure_transport_setup_done, c, + call_list); GRPC_SECURITY_CONNECTOR_UNREF(&sc->base, "httpcli"); } diff --git a/src/core/iomgr/alarm.c b/src/core/iomgr/alarm.c index 7b67fe3b1d..6e0d516f0c 100644 --- a/src/core/iomgr/alarm.c +++ b/src/core/iomgr/alarm.c @@ -44,7 +44,6 @@ #define LOG2_NUM_SHARDS 5 #define NUM_SHARDS (1 << LOG2_NUM_SHARDS) -#define MAX_ALARMS_PER_CHECK 128 #define ADD_DEADLINE_SCALE 0.33 #define MIN_QUEUE_WINDOW_DURATION 0.01 #define MAX_QUEUE_WINDOW_DURATION 1 @@ -73,8 +72,8 @@ static shard_type g_shards[NUM_SHARDS]; /* Protected by g_mu */ static shard_type *g_shard_queue[NUM_SHARDS]; -static int run_some_expired_alarms(gpr_mu *drop_mu, gpr_timespec now, - gpr_timespec *next, int success); +static int run_some_expired_alarms(gpr_timespec now, gpr_timespec *next, + int success, grpc_call_list *call_list); static gpr_timespec compute_min_deadline(shard_type *shard) { return grpc_alarm_heap_is_empty(&shard->heap) @@ -103,10 +102,9 @@ void grpc_alarm_list_init(gpr_timespec now) { } } -void grpc_alarm_list_shutdown(void) { +void grpc_alarm_list_shutdown(grpc_call_list *call_list) { int i; - while (run_some_expired_alarms(NULL, gpr_inf_future(g_clock_type), NULL, 0)) - ; + run_some_expired_alarms(gpr_inf_future(g_clock_type), NULL, 0, call_list); for (i = 0; i < NUM_SHARDS; i++) { shard_type *shard = &g_shards[i]; gpr_mu_destroy(&shard->mu); @@ -174,13 +172,12 @@ static void note_deadline_change(shard_type *shard) { void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline, grpc_iomgr_cb_func alarm_cb, void *alarm_cb_arg, - gpr_timespec now) { + gpr_timespec now, grpc_call_list *call_list) { int is_first_alarm = 0; shard_type *shard = &g_shards[shard_idx(alarm)]; GPR_ASSERT(deadline.clock_type == g_clock_type); GPR_ASSERT(now.clock_type == g_clock_type); - alarm->cb = alarm_cb; - alarm->cb_arg = alarm_cb_arg; + grpc_closure_init(&alarm->closure, alarm_cb, alarm_cb_arg); alarm->deadline = deadline; alarm->triggered = 0; @@ -223,12 +220,11 @@ void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline, } } -void grpc_alarm_cancel(grpc_alarm *alarm) { +void grpc_alarm_cancel(grpc_alarm *alarm, grpc_call_list *call_list) { shard_type *shard = &g_shards[shard_idx(alarm)]; - int triggered = 0; gpr_mu_lock(&shard->mu); if (!alarm->triggered) { - triggered = 1; + grpc_call_list_add(call_list, &alarm->closure, 1); alarm->triggered = 1; if (alarm->heap_index == INVALID_HEAP_INDEX) { list_remove(alarm); @@ -237,10 +233,6 @@ void grpc_alarm_cancel(grpc_alarm *alarm) { } } gpr_mu_unlock(&shard->mu); - - if (triggered) { - alarm->cb(alarm->cb_arg, 0); - } } /* This is called when the queue is empty and "now" has reached the @@ -292,39 +284,36 @@ static grpc_alarm *pop_one(shard_type *shard, gpr_timespec now) { /* REQUIRES: shard->mu unlocked */ static size_t pop_alarms(shard_type *shard, gpr_timespec now, - grpc_alarm **alarms, size_t max_alarms, - gpr_timespec *new_min_deadline) { + gpr_timespec *new_min_deadline, int success, + grpc_call_list *call_list) { size_t n = 0; grpc_alarm *alarm; gpr_mu_lock(&shard->mu); - while (n < max_alarms && (alarm = pop_one(shard, now))) { - alarms[n++] = alarm; + while ((alarm = pop_one(shard, now))) { + grpc_call_list_add(call_list, &alarm->closure, success); } *new_min_deadline = compute_min_deadline(shard); gpr_mu_unlock(&shard->mu); return n; } -static int run_some_expired_alarms(gpr_mu *drop_mu, gpr_timespec now, - gpr_timespec *next, int success) { +static int run_some_expired_alarms(gpr_timespec now, gpr_timespec *next, + int success, grpc_call_list *call_list) { size_t n = 0; - size_t i; - grpc_alarm *alarms[MAX_ALARMS_PER_CHECK]; /* TODO(ctiller): verify that there are any alarms (atomically) here */ if (gpr_mu_trylock(&g_checker_mu)) { gpr_mu_lock(&g_mu); - while (n < MAX_ALARMS_PER_CHECK && - gpr_time_cmp(g_shard_queue[0]->min_deadline, now) < 0) { + while (gpr_time_cmp(g_shard_queue[0]->min_deadline, now) < 0) { gpr_timespec new_min_deadline; /* For efficiency, we pop as many available alarms as we can from the shard. This may violate perfect alarm deadline ordering, but that shouldn't be a big deal because we don't make ordering guarantees. */ - n += pop_alarms(g_shard_queue[0], now, alarms + n, - MAX_ALARMS_PER_CHECK - n, &new_min_deadline); + n += pop_alarms(g_shard_queue[0], now, &new_min_deadline, success, + call_list); /* An grpc_alarm_init() on the shard could intervene here, adding a new alarm that is earlier than new_min_deadline. However, @@ -343,26 +332,15 @@ static int run_some_expired_alarms(gpr_mu *drop_mu, gpr_timespec now, gpr_mu_unlock(&g_checker_mu); } - if (n && drop_mu) { - gpr_mu_unlock(drop_mu); - } - - for (i = 0; i < n; i++) { - alarms[i]->cb(alarms[i]->cb_arg, success); - } - - if (n && drop_mu) { - gpr_mu_lock(drop_mu); - } - - return (int)n; + return n > 0; } -int grpc_alarm_check(gpr_mu *drop_mu, gpr_timespec now, gpr_timespec *next) { +int grpc_alarm_check(gpr_timespec now, gpr_timespec *next, + grpc_call_list *call_list) { GPR_ASSERT(now.clock_type == g_clock_type); return run_some_expired_alarms( - drop_mu, now, next, - gpr_time_cmp(now, gpr_inf_future(now.clock_type)) != 0); + now, next, gpr_time_cmp(now, gpr_inf_future(now.clock_type)) != 0, + call_list); } gpr_timespec grpc_alarm_list_next_timeout(void) { diff --git a/src/core/iomgr/alarm.h b/src/core/iomgr/alarm.h index 4a13527e64..05ca7c27bb 100644 --- a/src/core/iomgr/alarm.h +++ b/src/core/iomgr/alarm.h @@ -44,8 +44,7 @@ typedef struct grpc_alarm { int triggered; struct grpc_alarm *next; struct grpc_alarm *prev; - grpc_iomgr_cb_func cb; - void *cb_arg; + grpc_closure closure; } grpc_alarm; /* Initialize *alarm. When expired or canceled, alarm_cb will be called with @@ -56,7 +55,7 @@ typedef struct grpc_alarm { information about when to free up any user-level state. */ void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline, grpc_iomgr_cb_func alarm_cb, void *alarm_cb_arg, - gpr_timespec now); + gpr_timespec now, grpc_call_list *call_list); /* Note that there is no alarm destroy function. This is because the alarm is a one-time occurrence with a guarantee that the callback will @@ -84,6 +83,6 @@ void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline, matches this aim. Requires: cancel() must happen after add() on a given alarm */ -void grpc_alarm_cancel(grpc_alarm *alarm); +void grpc_alarm_cancel(grpc_alarm *alarm, grpc_call_list *call_list); #endif /* GRPC_INTERNAL_CORE_IOMGR_ALARM_H */ diff --git a/src/core/iomgr/alarm_internal.h b/src/core/iomgr/alarm_internal.h index e9f98a3444..a2e9946f4f 100644 --- a/src/core/iomgr/alarm_internal.h +++ b/src/core/iomgr/alarm_internal.h @@ -48,10 +48,10 @@ with high probability at least one thread in the system will see an update at any time slice. */ -int grpc_alarm_check(gpr_mu *drop_mu, gpr_timespec now, gpr_timespec *next); - +int grpc_alarm_check(gpr_timespec now, gpr_timespec *next, + grpc_call_list *call_list); void grpc_alarm_list_init(gpr_timespec now); -void grpc_alarm_list_shutdown(void); +void grpc_alarm_list_shutdown(grpc_call_list *call_list); gpr_timespec grpc_alarm_list_next_timeout(void); diff --git a/src/core/iomgr/endpoint.c b/src/core/iomgr/endpoint.c index 1955f74b9a..769f155a63 100644 --- a/src/core/iomgr/endpoint.c +++ b/src/core/iomgr/endpoint.c @@ -33,30 +33,34 @@ #include "src/core/iomgr/endpoint.h" -grpc_endpoint_op_status grpc_endpoint_read(grpc_endpoint *ep, - gpr_slice_buffer *slices, - grpc_closure *cb) { - return ep->vtable->read(ep, slices, cb); +void grpc_endpoint_read(grpc_endpoint *ep, gpr_slice_buffer *slices, + grpc_closure *cb, grpc_call_list *call_list) { + ep->vtable->read(ep, slices, cb, call_list); } -grpc_endpoint_op_status grpc_endpoint_write(grpc_endpoint *ep, - gpr_slice_buffer *slices, - grpc_closure *cb) { - return ep->vtable->write(ep, slices, cb); +void grpc_endpoint_write(grpc_endpoint *ep, gpr_slice_buffer *slices, + grpc_closure *cb, grpc_call_list *call_list) { + ep->vtable->write(ep, slices, cb, call_list); } -void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { - ep->vtable->add_to_pollset(ep, pollset); +void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset, + grpc_call_list *call_list) { + ep->vtable->add_to_pollset(ep, pollset, call_list); } void grpc_endpoint_add_to_pollset_set(grpc_endpoint *ep, - grpc_pollset_set *pollset_set) { - ep->vtable->add_to_pollset_set(ep, pollset_set); + grpc_pollset_set *pollset_set, + grpc_call_list *call_list) { + ep->vtable->add_to_pollset_set(ep, pollset_set, call_list); } -void grpc_endpoint_shutdown(grpc_endpoint *ep) { ep->vtable->shutdown(ep); } +void grpc_endpoint_shutdown(grpc_endpoint *ep, grpc_call_list *call_list) { + ep->vtable->shutdown(ep, call_list); +} -void grpc_endpoint_destroy(grpc_endpoint *ep) { ep->vtable->destroy(ep); } +void grpc_endpoint_destroy(grpc_endpoint *ep, grpc_call_list *call_list) { + ep->vtable->destroy(ep, call_list); +} char *grpc_endpoint_get_peer(grpc_endpoint *ep) { return ep->vtable->get_peer(ep); diff --git a/src/core/iomgr/endpoint.h b/src/core/iomgr/endpoint.h index 79b7d6a78e..f9881684ff 100644 --- a/src/core/iomgr/endpoint.h +++ b/src/core/iomgr/endpoint.h @@ -46,21 +46,17 @@ typedef struct grpc_endpoint grpc_endpoint; typedef struct grpc_endpoint_vtable grpc_endpoint_vtable; -typedef enum grpc_endpoint_op_status { - GRPC_ENDPOINT_DONE, /* completed immediately, cb won't be called */ - GRPC_ENDPOINT_PENDING, /* cb will be called when completed */ - GRPC_ENDPOINT_ERROR /* write errored out, cb won't be called */ -} grpc_endpoint_op_status; - struct grpc_endpoint_vtable { - grpc_endpoint_op_status (*read)(grpc_endpoint *ep, gpr_slice_buffer *slices, - grpc_closure *cb); - grpc_endpoint_op_status (*write)(grpc_endpoint *ep, gpr_slice_buffer *slices, - grpc_closure *cb); - void (*add_to_pollset)(grpc_endpoint *ep, grpc_pollset *pollset); - void (*add_to_pollset_set)(grpc_endpoint *ep, grpc_pollset_set *pollset); - void (*shutdown)(grpc_endpoint *ep); - void (*destroy)(grpc_endpoint *ep); + void (*read)(grpc_endpoint *ep, gpr_slice_buffer *slices, grpc_closure *cb, + grpc_call_list *call_list); + void (*write)(grpc_endpoint *ep, gpr_slice_buffer *slices, grpc_closure *cb, + grpc_call_list *call_list); + void (*add_to_pollset)(grpc_endpoint *ep, grpc_pollset *pollset, + grpc_call_list *call_list); + void (*add_to_pollset_set)(grpc_endpoint *ep, grpc_pollset_set *pollset, + grpc_call_list *call_list); + void (*shutdown)(grpc_endpoint *ep, grpc_call_list *call_list); + void (*destroy)(grpc_endpoint *ep, grpc_call_list *call_list); char *(*get_peer)(grpc_endpoint *ep); }; @@ -68,9 +64,8 @@ struct grpc_endpoint_vtable { Callback success indicates that the endpoint can accept more reads, failure indicates the endpoint is closed. Valid slices may be placed into \a slices even on callback success == 0. */ -grpc_endpoint_op_status grpc_endpoint_read( - grpc_endpoint *ep, gpr_slice_buffer *slices, - grpc_closure *cb) GRPC_MUST_USE_RESULT; +void grpc_endpoint_read(grpc_endpoint *ep, gpr_slice_buffer *slices, + grpc_closure *cb, grpc_call_list *call_list); char *grpc_endpoint_get_peer(grpc_endpoint *ep); @@ -84,20 +79,21 @@ char *grpc_endpoint_get_peer(grpc_endpoint *ep); No guarantee is made to the content of slices after a write EXCEPT that it is a valid slice buffer. */ -grpc_endpoint_op_status grpc_endpoint_write( - grpc_endpoint *ep, gpr_slice_buffer *slices, - grpc_closure *cb) GRPC_MUST_USE_RESULT; +void grpc_endpoint_write(grpc_endpoint *ep, gpr_slice_buffer *slices, + grpc_closure *cb, grpc_call_list *call_list); /* Causes any pending read/write callbacks to run immediately with success==0 */ -void grpc_endpoint_shutdown(grpc_endpoint *ep); -void grpc_endpoint_destroy(grpc_endpoint *ep); +void grpc_endpoint_shutdown(grpc_endpoint *ep, grpc_call_list *call_list); +void grpc_endpoint_destroy(grpc_endpoint *ep, grpc_call_list *call_list); /* Add an endpoint to a pollset, so that when the pollset is polled, events from this endpoint are considered */ -void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset); +void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset, + grpc_call_list *call_list); void grpc_endpoint_add_to_pollset_set(grpc_endpoint *ep, - grpc_pollset_set *pollset_set); + grpc_pollset_set *pollset_set, + grpc_call_list *call_list); struct grpc_endpoint { const grpc_endpoint_vtable *vtable; diff --git a/src/core/iomgr/endpoint_pair.h b/src/core/iomgr/endpoint_pair.h index 25ef1891fb..095ec5fcc9 100644 --- a/src/core/iomgr/endpoint_pair.h +++ b/src/core/iomgr/endpoint_pair.h @@ -42,7 +42,6 @@ typedef struct { } grpc_endpoint_pair; grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name, - size_t read_slice_size, - grpc_workqueue *workqueue); + size_t read_slice_size); #endif /* GRPC_INTERNAL_CORE_IOMGR_ENDPOINT_PAIR_H */ diff --git a/src/core/iomgr/endpoint_pair_posix.c b/src/core/iomgr/endpoint_pair_posix.c index dc1f441b4b..deae9c6875 100644 --- a/src/core/iomgr/endpoint_pair_posix.c +++ b/src/core/iomgr/endpoint_pair_posix.c @@ -59,20 +59,19 @@ static void create_sockets(int sv[2]) { } grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name, - size_t read_slice_size, - grpc_workqueue *workqueue) { + size_t read_slice_size) { int sv[2]; grpc_endpoint_pair p; char *final_name; create_sockets(sv); gpr_asprintf(&final_name, "%s:client", name); - p.client = grpc_tcp_create(grpc_fd_create(sv[1], workqueue, final_name), - read_slice_size, "socketpair-server"); + p.client = grpc_tcp_create(grpc_fd_create(sv[1], final_name), read_slice_size, + "socketpair-server"); gpr_free(final_name); gpr_asprintf(&final_name, "%s:server", name); - p.server = grpc_tcp_create(grpc_fd_create(sv[0], workqueue, final_name), - read_slice_size, "socketpair-client"); + p.server = grpc_tcp_create(grpc_fd_create(sv[0], final_name), read_slice_size, + "socketpair-client"); gpr_free(final_name); return p; } diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c index dfebbcc2e1..294cb70746 100644 --- a/src/core/iomgr/fd_posix.c +++ b/src/core/iomgr/fd_posix.c @@ -71,9 +71,6 @@ static grpc_fd *fd_freelist = NULL; static gpr_mu fd_freelist_mu; static void freelist_fd(grpc_fd *fd) { - if (fd->workqueue->wakeup_read_fd != fd) { - GRPC_WORKQUEUE_UNREF(fd->workqueue, "fd"); - } gpr_mu_lock(&fd_freelist_mu); fd->freelist_next = fd_freelist; fd_freelist = fd; @@ -161,14 +158,8 @@ void grpc_fd_global_shutdown(void) { gpr_mu_destroy(&fd_freelist_mu); } -grpc_fd *grpc_fd_create(int fd, grpc_workqueue *workqueue, const char *name) { +grpc_fd *grpc_fd_create(int fd, const char *name) { grpc_fd *r = alloc_fd(fd); - r->workqueue = workqueue; - /* if the wakeup_read_fd is NULL, then the workqueue is under construction - ==> this fd will be the wakeup_read_fd, and we shouldn't take a ref */ - if (workqueue->wakeup_read_fd != NULL) { - GRPC_WORKQUEUE_REF(workqueue, "fd"); - } grpc_iomgr_register_object(&r->iomgr_object, name); return r; } @@ -218,7 +209,8 @@ static int has_watchers(grpc_fd *fd) { fd->inactive_watcher_root.next != &fd->inactive_watcher_root; } -void grpc_fd_orphan(grpc_fd *fd, grpc_closure *on_done, const char *reason) { +void grpc_fd_orphan(grpc_fd *fd, grpc_closure *on_done, const char *reason, + grpc_call_list *call_list) { fd->on_done_closure = on_done; shutdown(fd->fd, SHUT_RDWR); gpr_mu_lock(&fd->watcher_mu); @@ -226,9 +218,7 @@ void grpc_fd_orphan(grpc_fd *fd, grpc_closure *on_done, const char *reason) { if (!has_watchers(fd)) { fd->closed = 1; close(fd->fd); - if (fd->on_done_closure) { - grpc_workqueue_push(fd->workqueue, fd->on_done_closure, 1); - } + grpc_call_list_add(call_list, fd->on_done_closure, 1); } else { wake_all_watchers_locked(fd); } @@ -252,25 +242,8 @@ void grpc_fd_ref(grpc_fd *fd) { ref_by(fd, 2); } void grpc_fd_unref(grpc_fd *fd) { unref_by(fd, 2); } #endif -static void process_callback(grpc_closure *closure, int success, - grpc_workqueue *optional_workqueue) { - if (optional_workqueue == NULL) { - closure->cb(closure->cb_arg, success); - } else { - grpc_workqueue_push(optional_workqueue, closure, success); - } -} - -static void process_callbacks(grpc_closure *callbacks, size_t n, int success, - grpc_workqueue *optional_workqueue) { - size_t i; - for (i = 0; i < n; i++) { - process_callback(callbacks + i, success, optional_workqueue); - } -} - static void notify_on(grpc_fd *fd, gpr_atm *st, grpc_closure *closure, - int allow_synchronous_callback) { + grpc_call_list *call_list) { switch (gpr_atm_acq_load(st)) { case NOT_READY: /* There is no race if the descriptor is already ready, so we skip @@ -292,8 +265,7 @@ static void notify_on(grpc_fd *fd, gpr_atm *st, grpc_closure *closure, case READY: GPR_ASSERT(gpr_atm_no_barrier_load(st) == READY); gpr_atm_rel_store(st, NOT_READY); - process_callback(closure, !gpr_atm_acq_load(&fd->shutdown), - allow_synchronous_callback ? NULL : fd->workqueue); + grpc_call_list_add(call_list, closure, !gpr_atm_acq_load(&fd->shutdown)); return; default: /* WAITING */ /* upcallptr was set to a different closure. This is an error! */ @@ -306,8 +278,8 @@ static void notify_on(grpc_fd *fd, gpr_atm *st, grpc_closure *closure, abort(); } -static void set_ready_locked(gpr_atm *st, grpc_closure **callbacks, - size_t *ncallbacks) { +static void set_ready_locked(grpc_fd *fd, gpr_atm *st, + grpc_call_list *call_list) { gpr_intptr state = gpr_atm_acq_load(st); switch (state) { @@ -326,50 +298,38 @@ static void set_ready_locked(gpr_atm *st, grpc_closure **callbacks, default: /* waiting */ GPR_ASSERT(gpr_atm_no_barrier_load(st) != READY && gpr_atm_no_barrier_load(st) != NOT_READY); - callbacks[(*ncallbacks)++] = (grpc_closure *)state; + grpc_call_list_add(call_list, (grpc_closure *)state, + !gpr_atm_acq_load(&fd->shutdown)); gpr_atm_rel_store(st, NOT_READY); return; } } -static void set_ready(grpc_fd *fd, gpr_atm *st, - int allow_synchronous_callback) { +static void set_ready(grpc_fd *fd, gpr_atm *st, grpc_call_list *call_list) { /* only one set_ready can be active at once (but there may be a racing notify_on) */ - int success; - grpc_closure *closure; - size_t ncb = 0; - gpr_mu_lock(&fd->set_state_mu); - set_ready_locked(st, &closure, &ncb); + set_ready_locked(fd, st, call_list); gpr_mu_unlock(&fd->set_state_mu); - success = !gpr_atm_acq_load(&fd->shutdown); - GPR_ASSERT(ncb <= 1); - if (ncb > 0) { - process_callbacks(closure, ncb, success, - allow_synchronous_callback ? NULL : fd->workqueue); - } } -void grpc_fd_shutdown(grpc_fd *fd) { - size_t ncb = 0; +void grpc_fd_shutdown(grpc_fd *fd, grpc_call_list *call_list) { gpr_mu_lock(&fd->set_state_mu); GPR_ASSERT(!gpr_atm_no_barrier_load(&fd->shutdown)); gpr_atm_rel_store(&fd->shutdown, 1); - set_ready_locked(&fd->readst, &fd->shutdown_closures[0], &ncb); - set_ready_locked(&fd->writest, &fd->shutdown_closures[0], &ncb); + set_ready_locked(fd, &fd->readst, call_list); + set_ready_locked(fd, &fd->writest, call_list); gpr_mu_unlock(&fd->set_state_mu); - GPR_ASSERT(ncb <= 2); - process_callbacks(fd->shutdown_closures[0], ncb, 0 /* GPR_FALSE */, - 0 /* GPR_FALSE */); } -void grpc_fd_notify_on_read(grpc_fd *fd, grpc_closure *closure) { - notify_on(fd, &fd->readst, closure, 0); +void grpc_fd_notify_on_read(grpc_fd *fd, grpc_closure *closure, + grpc_call_list *call_list) { + notify_on(fd, &fd->readst, closure, call_list); } -void grpc_fd_notify_on_write(grpc_fd *fd, grpc_closure *closure) { - notify_on(fd, &fd->writest, closure, 0); +void grpc_fd_notify_on_write(grpc_fd *fd, grpc_closure *closure, + grpc_call_list *call_list) { + notify_on(fd, &fd->writest, closure, call_list); } gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset, @@ -415,7 +375,8 @@ gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset, return mask; } -void grpc_fd_end_poll(grpc_fd_watcher *watcher, int got_read, int got_write) { +void grpc_fd_end_poll(grpc_fd_watcher *watcher, int got_read, int got_write, + grpc_call_list *call_list) { int was_polling = 0; int kick = 0; grpc_fd *fd = watcher->fd; @@ -448,21 +409,19 @@ void grpc_fd_end_poll(grpc_fd_watcher *watcher, int got_read, int got_write) { if (grpc_fd_is_orphaned(fd) && !has_watchers(fd) && !fd->closed) { fd->closed = 1; close(fd->fd); - if (fd->on_done_closure != NULL) { - grpc_workqueue_push(fd->workqueue, fd->on_done_closure, 1); - } + grpc_call_list_add(call_list, fd->on_done_closure, 1); } gpr_mu_unlock(&fd->watcher_mu); GRPC_FD_UNREF(fd, "poll"); } -void grpc_fd_become_readable(grpc_fd *fd, int allow_synchronous_callback) { - set_ready(fd, &fd->readst, allow_synchronous_callback); +void grpc_fd_become_readable(grpc_fd *fd, grpc_call_list *call_list) { + set_ready(fd, &fd->readst, call_list); } -void grpc_fd_become_writable(grpc_fd *fd, int allow_synchronous_callback) { - set_ready(fd, &fd->writest, allow_synchronous_callback); +void grpc_fd_become_writable(grpc_fd *fd, grpc_call_list *call_list) { + set_ready(fd, &fd->writest, call_list); } #endif diff --git a/src/core/iomgr/fd_posix.h b/src/core/iomgr/fd_posix.h index bb85b6c16e..f435e2d3f9 100644 --- a/src/core/iomgr/fd_posix.h +++ b/src/core/iomgr/fd_posix.h @@ -58,7 +58,6 @@ struct grpc_fd { meaning that mostly we ref by two to avoid altering the orphaned bit, and just unref by 1 when we're ready to flag the object as orphaned */ gpr_atm refst; - grpc_workqueue *workqueue; gpr_mu set_state_mu; gpr_atm shutdown; @@ -105,7 +104,7 @@ struct grpc_fd { /* Create a wrapped file descriptor. Requires fd is a non-blocking file descriptor. This takes ownership of closing fd. */ -grpc_fd *grpc_fd_create(int fd, grpc_workqueue *workqueue, const char *name); +grpc_fd *grpc_fd_create(int fd, const char *name); /* Releases fd to be asynchronously destroyed. on_done is called when the underlying file descriptor is definitely close()d. @@ -113,7 +112,8 @@ grpc_fd *grpc_fd_create(int fd, grpc_workqueue *workqueue, const char *name); Requires: *fd initialized; no outstanding notify_on_read or notify_on_write. MUST NOT be called with a pollset lock taken */ -void grpc_fd_orphan(grpc_fd *fd, grpc_closure *on_done, const char *reason); +void grpc_fd_orphan(grpc_fd *fd, grpc_closure *on_done, const char *reason, + grpc_call_list *call_list); /* Begin polling on an fd. Registers that the given pollset is interested in this fd - so that if read @@ -131,13 +131,14 @@ gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset, grpc_fd_watcher *rec); /* Complete polling previously started with grpc_fd_begin_poll MUST NOT be called with a pollset lock taken */ -void grpc_fd_end_poll(grpc_fd_watcher *rec, int got_read, int got_write); +void grpc_fd_end_poll(grpc_fd_watcher *rec, int got_read, int got_write, + grpc_call_list *call_list); /* Return 1 if this fd is orphaned, 0 otherwise */ int grpc_fd_is_orphaned(grpc_fd *fd); /* Cause any current callbacks to error out with GRPC_CALLBACK_CANCELLED. */ -void grpc_fd_shutdown(grpc_fd *fd); +void grpc_fd_shutdown(grpc_fd *fd, grpc_call_list *call_list); /* Register read interest, causing read_cb to be called once when fd becomes readable, on deadline specified by deadline, or on shutdown triggered by @@ -152,17 +153,19 @@ void grpc_fd_shutdown(grpc_fd *fd); underlying platform. This means that users must drain fd in read_cb before calling notify_on_read again. Users are also expected to handle spurious events, i.e read_cb is called while nothing can be readable from fd */ -void grpc_fd_notify_on_read(grpc_fd *fd, grpc_closure *closure); +void grpc_fd_notify_on_read(grpc_fd *fd, grpc_closure *closure, + grpc_call_list *call_list); /* Exactly the same semantics as above, except based on writable events. */ -void grpc_fd_notify_on_write(grpc_fd *fd, grpc_closure *closure); +void grpc_fd_notify_on_write(grpc_fd *fd, grpc_closure *closure, + grpc_call_list *call_list); /* Notification from the poller to an fd that it has become readable or writable. If allow_synchronous_callback is 1, allow running the fd callback inline in this callstack, otherwise register an asynchronous callback and return */ -void grpc_fd_become_readable(grpc_fd *fd, int allow_synchronous_callback); -void grpc_fd_become_writable(grpc_fd *fd, int allow_synchronous_callback); +void grpc_fd_become_readable(grpc_fd *fd, grpc_call_list *call_list); +void grpc_fd_become_writable(grpc_fd *fd, grpc_call_list *call_list); /* Reference counting for fds */ #ifdef GRPC_FD_REF_COUNT_DEBUG diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index dd76044913..029c689982 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -88,6 +88,7 @@ void grpc_iomgr_shutdown(void) { gpr_timespec shutdown_deadline = gpr_time_add( gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_seconds(10, GPR_TIMESPAN)); gpr_timespec last_warning_time = gpr_now(GPR_CLOCK_REALTIME); + grpc_call_list call_list = GRPC_CALL_LIST_INIT; gpr_mu_lock(&g_mu); g_shutdown = 1; @@ -101,7 +102,11 @@ void grpc_iomgr_shutdown(void) { } last_warning_time = gpr_now(GPR_CLOCK_REALTIME); } - if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL)) { + if (grpc_alarm_check(gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL, + &call_list)) { + gpr_mu_unlock(&g_mu); + grpc_call_list_run(&call_list); + gpr_mu_lock(&g_mu); continue; } if (g_root_object.next != &g_root_object) { @@ -126,7 +131,8 @@ void grpc_iomgr_shutdown(void) { } gpr_mu_unlock(&g_mu); - grpc_alarm_list_shutdown(); + grpc_alarm_list_shutdown(&call_list); + grpc_call_list_run(&call_list); grpc_iomgr_platform_shutdown(); gpr_mu_destroy(&g_mu); @@ -171,12 +177,15 @@ void grpc_call_list_add(grpc_call_list *call_list, grpc_closure *closure, call_list->tail = closure; } -void grpc_call_list_run(grpc_call_list call_list) { - grpc_closure *c = call_list.head; - while (c) { - grpc_closure *next = c->next; - c->cb(c->cb_arg, c->success); - c = next; +void grpc_call_list_run(grpc_call_list *call_list) { + while (!grpc_call_list_empty(*call_list)) { + grpc_closure *c = call_list->head; + call_list->head = call_list->tail = NULL; + while (c) { + grpc_closure *next = c->next; + c->cb(c->cb_arg, c->success, call_list); + c = next; + } } } diff --git a/src/core/iomgr/iomgr.h b/src/core/iomgr/iomgr.h index 56f0195be9..94f3912990 100644 --- a/src/core/iomgr/iomgr.h +++ b/src/core/iomgr/iomgr.h @@ -34,15 +34,24 @@ #ifndef GRPC_INTERNAL_CORE_IOMGR_IOMGR_H #define GRPC_INTERNAL_CORE_IOMGR_IOMGR_H +struct grpc_closure; +typedef struct grpc_closure grpc_closure; + +typedef struct grpc_call_list { + grpc_closure *head; + grpc_closure *tail; +} grpc_call_list; + /** gRPC Callback definition. * * \param arg Arbitrary input. * \param success An indication on the state of the iomgr. On false, cleanup * actions should be taken (eg, shutdown). */ -typedef void (*grpc_iomgr_cb_func)(void *arg, int success); +typedef void (*grpc_iomgr_cb_func)(void *arg, int success, + grpc_call_list *call_list); /** A closure over a grpc_iomgr_cb_func. */ -typedef struct grpc_closure { +struct grpc_closure { /** Bound callback. */ grpc_iomgr_cb_func cb; @@ -56,12 +65,7 @@ typedef struct grpc_closure { /**< Internal. Do not touch */ struct grpc_closure *next; -} grpc_closure; - -typedef struct grpc_call_list { - grpc_closure *head; - grpc_closure *tail; -} grpc_call_list; +}; /** Initializes \a closure with \a cb and \a cb_arg. */ void grpc_closure_init(grpc_closure *closure, grpc_iomgr_cb_func cb, @@ -72,7 +76,7 @@ void grpc_closure_init(grpc_closure *closure, grpc_iomgr_cb_func cb, void grpc_call_list_add(grpc_call_list *list, grpc_closure *closure, int success); -void grpc_call_list_run(grpc_call_list list); +void grpc_call_list_run(grpc_call_list *list); void grpc_call_list_move(grpc_call_list *src, grpc_call_list *dst); int grpc_call_list_empty(grpc_call_list list); diff --git a/src/core/iomgr/pollset.h b/src/core/iomgr/pollset.h index 337596cb74..98653412bb 100644 --- a/src/core/iomgr/pollset.h +++ b/src/core/iomgr/pollset.h @@ -55,9 +55,8 @@ #endif void grpc_pollset_init(grpc_pollset *pollset); -void grpc_pollset_shutdown(grpc_pollset *pollset, - void (*shutdown_done)(void *arg), - void *shutdown_done_arg); +void grpc_pollset_shutdown(grpc_pollset *pollset, grpc_closure *closure, + grpc_call_list *call_list); void grpc_pollset_destroy(grpc_pollset *pollset); /* Do some work on a pollset. diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c index 5ca957b8e1..e6e0f5cdd4 100644 --- a/src/core/iomgr/pollset_multipoller_with_epoll.c +++ b/src/core/iomgr/pollset_multipoller_with_epoll.c @@ -61,7 +61,8 @@ typedef struct { wakeup_fd_hdl *free_wakeup_fds; } pollset_hdr; -static void finally_add_fd(grpc_pollset *pollset, grpc_fd *fd) { +static void finally_add_fd(grpc_pollset *pollset, grpc_fd *fd, + grpc_call_list *call_list) { pollset_hdr *h = pollset->data.ptr; struct epoll_event ev; int err; @@ -83,15 +84,15 @@ static void finally_add_fd(grpc_pollset *pollset, grpc_fd *fd) { } } } - grpc_fd_end_poll(&watcher, 0, 0); + grpc_fd_end_poll(&watcher, 0, 0, call_list); } -static void perform_delayed_add(void *arg, int iomgr_status) { +static void perform_delayed_add(void *arg, int iomgr_status, + grpc_call_list *call_list) { delayed_add *da = arg; - int do_shutdown_cb = 0; if (!grpc_fd_is_orphaned(da->fd)) { - finally_add_fd(da->pollset, da->fd); + finally_add_fd(da->pollset, da->fd, call_list); } gpr_mu_lock(&da->pollset->mu); @@ -100,26 +101,23 @@ static void perform_delayed_add(void *arg, int iomgr_status) { /* We don't care about this pollset anymore. */ if (da->pollset->in_flight_cbs == 0 && !da->pollset->called_shutdown) { da->pollset->called_shutdown = 1; - do_shutdown_cb = 1; + grpc_call_list_add(call_list, da->pollset->shutdown_done, 1); } } gpr_mu_unlock(&da->pollset->mu); GRPC_FD_UNREF(da->fd, "delayed_add"); - if (do_shutdown_cb) { - da->pollset->shutdown_done_cb(da->pollset->shutdown_done_arg); - } - gpr_free(da); } static void multipoll_with_epoll_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd, - int and_unlock_pollset) { + int and_unlock_pollset, + grpc_call_list *call_list) { if (and_unlock_pollset) { gpr_mu_unlock(&pollset->mu); - finally_add_fd(pollset, fd); + finally_add_fd(pollset, fd, call_list); } else { delayed_add *da = gpr_malloc(sizeof(*da)); da->pollset = pollset; @@ -127,13 +125,14 @@ static void multipoll_with_epoll_pollset_add_fd(grpc_pollset *pollset, GRPC_FD_REF(fd, "delayed_add"); grpc_closure_init(&da->closure, perform_delayed_add, da); pollset->in_flight_cbs++; - grpc_pollset_add_unlock_job(pollset, &da->closure); + grpc_call_list_add(call_list, &da->closure, 1); } } static void multipoll_with_epoll_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd, - int and_unlock_pollset) { + int and_unlock_pollset, + grpc_call_list *call_list) { pollset_hdr *h = pollset->data.ptr; int err; @@ -153,9 +152,9 @@ static void multipoll_with_epoll_pollset_del_fd(grpc_pollset *pollset, /* TODO(klempner): We probably want to turn this down a bit */ #define GRPC_EPOLL_MAX_EVENTS 1000 -static void multipoll_with_epoll_pollset_maybe_work( +static void multipoll_with_epoll_pollset_maybe_work_and_unlock( grpc_pollset *pollset, grpc_pollset_worker *worker, gpr_timespec deadline, - gpr_timespec now, int allow_synchronous_callback) { + gpr_timespec now, grpc_call_list *call_list) { struct epoll_event ep_ev[GRPC_EPOLL_MAX_EVENTS]; int ep_rv; int poll_rv; @@ -209,18 +208,16 @@ static void multipoll_with_epoll_pollset_maybe_work( int read = ep_ev[i].events & (EPOLLIN | EPOLLPRI); int write = ep_ev[i].events & EPOLLOUT; if (read || cancel) { - grpc_fd_become_readable(fd, allow_synchronous_callback); + grpc_fd_become_readable(fd, call_list); } if (write || cancel) { - grpc_fd_become_writable(fd, allow_synchronous_callback); + grpc_fd_become_writable(fd, call_list); } } } } while (ep_rv == GRPC_EPOLL_MAX_EVENTS); } } - - gpr_mu_lock(&pollset->mu); } static void multipoll_with_epoll_pollset_finish_shutdown( @@ -234,12 +231,12 @@ static void multipoll_with_epoll_pollset_destroy(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, + multipoll_with_epoll_pollset_maybe_work_and_unlock, 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) { + size_t nfds, grpc_call_list *call_list) { size_t i; pollset_hdr *h = gpr_malloc(sizeof(pollset_hdr)); @@ -252,7 +249,7 @@ static void epoll_become_multipoller(grpc_pollset *pollset, grpc_fd **fds, abort(); } for (i = 0; i < nfds; i++) { - multipoll_with_epoll_pollset_add_fd(pollset, fds[i], 0); + multipoll_with_epoll_pollset_add_fd(pollset, fds[i], 0, call_list); } } diff --git a/src/core/iomgr/pollset_multipoller_with_poll_posix.c b/src/core/iomgr/pollset_multipoller_with_poll_posix.c index cae260cab0..03a28894f0 100644 --- a/src/core/iomgr/pollset_multipoller_with_poll_posix.c +++ b/src/core/iomgr/pollset_multipoller_with_poll_posix.c @@ -61,7 +61,8 @@ typedef struct { static void multipoll_with_poll_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd, - int and_unlock_pollset) { + int and_unlock_pollset, + grpc_call_list *call_list) { size_t i; pollset_hdr *h = pollset->data.ptr; /* TODO(ctiller): this is O(num_fds^2); maybe switch to a hash set here */ @@ -82,7 +83,8 @@ exit: static void multipoll_with_poll_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd, - int and_unlock_pollset) { + int and_unlock_pollset, + grpc_call_list *call_list) { /* will get removed next poll cycle */ pollset_hdr *h = pollset->data.ptr; if (h->del_count == h->del_capacity) { @@ -96,9 +98,9 @@ static void multipoll_with_poll_pollset_del_fd(grpc_pollset *pollset, } } -static void multipoll_with_poll_pollset_maybe_work( +static void multipoll_with_poll_pollset_maybe_work_and_unlock( grpc_pollset *pollset, grpc_pollset_worker *worker, gpr_timespec deadline, - gpr_timespec now, int allow_synchronous_callback) { + gpr_timespec now, grpc_call_list *call_list) { int timeout; int r; size_t i, j, fd_count; @@ -149,7 +151,7 @@ static void multipoll_with_poll_pollset_maybe_work( for (i = 1; i < pfd_count; i++) { grpc_fd_end_poll(&watchers[i], pfds[i].revents & POLLIN, - pfds[i].revents & POLLOUT); + pfds[i].revents & POLLOUT, call_list); } if (r < 0) { @@ -167,18 +169,16 @@ static void multipoll_with_poll_pollset_maybe_work( continue; } if (pfds[i].revents & (POLLIN | POLLHUP | POLLERR)) { - grpc_fd_become_readable(watchers[i].fd, allow_synchronous_callback); + grpc_fd_become_readable(watchers[i].fd, call_list); } if (pfds[i].revents & (POLLOUT | POLLHUP | POLLERR)) { - grpc_fd_become_writable(watchers[i].fd, allow_synchronous_callback); + grpc_fd_become_writable(watchers[i].fd, call_list); } } } gpr_free(pfds); gpr_free(watchers); - - gpr_mu_lock(&pollset->mu); } static void multipoll_with_poll_pollset_finish_shutdown(grpc_pollset *pollset) { @@ -204,12 +204,12 @@ 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_maybe_work_and_unlock, 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) { + size_t nfds, grpc_call_list *call_list) { size_t i; pollset_hdr *h = gpr_malloc(sizeof(pollset_hdr)); pollset->vtable = &multipoll_with_poll_pollset; diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c index 0fe3f80d44..885cb29234 100644 --- a/src/core/iomgr/pollset_posix.c +++ b/src/core/iomgr/pollset_posix.c @@ -136,17 +136,14 @@ void grpc_pollset_init(grpc_pollset *pollset) { pollset->in_flight_cbs = 0; pollset->shutting_down = 0; pollset->called_shutdown = 0; - pollset->idle_jobs = NULL; - pollset->unlock_jobs = NULL; + pollset->idle_jobs.head = pollset->idle_jobs.tail = NULL; become_basic_pollset(pollset, NULL); } -void grpc_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { - if (fd->workqueue->wakeup_read_fd != fd) { - grpc_pollset_add_fd(pollset, fd->workqueue->wakeup_read_fd); - } +void grpc_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd, + grpc_call_list *call_list) { gpr_mu_lock(&pollset->mu); - pollset->vtable->add_fd(pollset, fd, 1); + pollset->vtable->add_fd(pollset, fd, 1, call_list); /* the following (enabled only in debug) will reacquire and then release our lock - meaning that if the unlocking flag passed to del_fd above is not respected, the code will deadlock (in a way that we have a chance of @@ -157,9 +154,10 @@ void grpc_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { #endif } -void grpc_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) { +void grpc_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd, + grpc_call_list *call_list) { gpr_mu_lock(&pollset->mu); - pollset->vtable->del_fd(pollset, fd, 1); + pollset->vtable->del_fd(pollset, fd, 1, call_list); /* the following (enabled only in debug) will reacquire and then release our lock - meaning that if the unlocking flag passed to del_fd above is not respected, the code will deadlock (in a way that we have a chance of @@ -170,53 +168,27 @@ void grpc_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) { #endif } -static void finish_shutdown(grpc_pollset *pollset) { +static void finish_shutdown(grpc_pollset *pollset, grpc_call_list *call_list) { pollset->vtable->finish_shutdown(pollset); - pollset->shutdown_done_cb(pollset->shutdown_done_arg); -} - -static void run_jobs(grpc_pollset *pollset, grpc_closure **root) { - grpc_closure *exec = *root; - *root = NULL; - gpr_mu_unlock(&pollset->mu); - while (exec != NULL) { - grpc_closure *next = exec->next; - exec->cb(exec->cb_arg, 1); - exec = next; - } - gpr_mu_lock(&pollset->mu); -} - -static void add_job(grpc_closure **root, grpc_closure *closure) { - closure->next = *root; - *root = closure; -} - -void grpc_pollset_add_idle_job(grpc_pollset *pollset, grpc_closure *closure) { - add_job(&pollset->idle_jobs, closure); -} - -void grpc_pollset_add_unlock_job(grpc_pollset *pollset, grpc_closure *closure) { - add_job(&pollset->unlock_jobs, closure); + grpc_call_list_add(call_list, pollset->shutdown_done, 1); } void grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker, gpr_timespec now, gpr_timespec deadline) { /* pollset->mu already held */ int added_worker = 0; + int locked = 1; + grpc_call_list call_list = GRPC_CALL_LIST_INIT; /* this must happen before we (potentially) drop pollset->mu */ worker->next = worker->prev = NULL; /* TODO(ctiller): pool these */ grpc_wakeup_fd_init(&worker->wakeup_fd); - if (!grpc_pollset_has_workers(pollset) && pollset->idle_jobs != NULL) { - run_jobs(pollset, &pollset->idle_jobs); + if (!grpc_pollset_has_workers(pollset) && + !grpc_call_list_empty(pollset->idle_jobs)) { + grpc_call_list_move(&pollset->idle_jobs, &call_list); goto done; } - if (pollset->unlock_jobs != NULL) { - run_jobs(pollset, &pollset->unlock_jobs); - goto done; - } - if (grpc_alarm_check(&pollset->mu, now, &deadline)) { + if (grpc_alarm_check(now, &deadline, &call_list)) { goto done; } if (pollset->shutting_down) { @@ -225,19 +197,32 @@ void grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker, if (pollset->in_flight_cbs) { /* Give do_promote priority so we don't starve it out */ gpr_mu_unlock(&pollset->mu); - gpr_mu_lock(&pollset->mu); + locked = 0; goto done; } if (!pollset->kicked_without_pollers) { push_front_worker(pollset, worker); added_worker = 1; gpr_tls_set(&g_current_thread_poller, (gpr_intptr)pollset); - pollset->vtable->maybe_work(pollset, worker, deadline, now, 1); + pollset->vtable->maybe_work_and_unlock(pollset, worker, deadline, now, + NULL); + locked = 0; gpr_tls_set(&g_current_thread_poller, 0); } else { pollset->kicked_without_pollers = 0; } done: + if (!grpc_call_list_empty(call_list)) { + if (locked) { + gpr_mu_unlock(&pollset->mu); + locked = 0; + } + grpc_call_list_run(&call_list); + } + if (!locked) { + gpr_mu_lock(&pollset->mu); + locked = 1; + } grpc_wakeup_fd_destroy(&worker->wakeup_fd); if (added_worker) { remove_worker(pollset, worker); @@ -248,7 +233,8 @@ done: } else if (!pollset->called_shutdown && pollset->in_flight_cbs == 0) { pollset->called_shutdown = 1; gpr_mu_unlock(&pollset->mu); - finish_shutdown(pollset); + finish_shutdown(pollset, &call_list); + grpc_call_list_run(&call_list); /* 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. @@ -258,9 +244,8 @@ done: } } -void grpc_pollset_shutdown(grpc_pollset *pollset, - void (*shutdown_done)(void *arg), - void *shutdown_done_arg) { +void grpc_pollset_shutdown(grpc_pollset *pollset, grpc_closure *closure, + grpc_call_list *call_list) { int call_shutdown = 0; gpr_mu_lock(&pollset->mu); GPR_ASSERT(!pollset->shutting_down); @@ -270,13 +255,12 @@ void grpc_pollset_shutdown(grpc_pollset *pollset, pollset->called_shutdown = 1; call_shutdown = 1; } - pollset->shutdown_done_cb = shutdown_done; - pollset->shutdown_done_arg = shutdown_done_arg; + pollset->shutdown_done = closure; grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST); gpr_mu_unlock(&pollset->mu); if (call_shutdown) { - finish_shutdown(pollset); + finish_shutdown(pollset, call_list); } } @@ -317,12 +301,12 @@ typedef struct grpc_unary_promote_args { grpc_closure promotion_closure; } grpc_unary_promote_args; -static void basic_do_promote(void *args, int success) { +static void basic_do_promote(void *args, int success, + grpc_call_list *call_list) { grpc_unary_promote_args *up_args = args; const grpc_pollset_vtable *original_vtable = up_args->original_vtable; grpc_pollset *pollset = up_args->pollset; grpc_fd *fd = up_args->fd; - int do_shutdown_cb = 0; /* * This is quite tricky. There are a number of cases to keep in mind here: @@ -349,19 +333,20 @@ static void basic_do_promote(void *args, int success) { if (pollset->in_flight_cbs == 0 && !pollset->called_shutdown) { GPR_ASSERT(!grpc_pollset_has_workers(pollset)); pollset->called_shutdown = 1; - do_shutdown_cb = 1; + grpc_call_list_add(call_list, pollset->shutdown_done, 1); } } else if (grpc_fd_is_orphaned(fd)) { /* Don't try to add it to anything, we'll drop our ref on it below */ } else if (pollset->vtable != original_vtable) { - pollset->vtable->add_fd(pollset, fd, 0); + pollset->vtable->add_fd(pollset, fd, 0, call_list); } else if (fd != pollset->data.ptr) { grpc_fd *fds[2]; fds[0] = pollset->data.ptr; fds[1] = fd; if (fds[0] && !grpc_fd_is_orphaned(fds[0])) { - grpc_platform_become_multipoller(pollset, fds, GPR_ARRAY_SIZE(fds)); + grpc_platform_become_multipoller(pollset, fds, GPR_ARRAY_SIZE(fds), + call_list); GRPC_FD_UNREF(fds[0], "basicpoll"); } else { /* old fd is orphaned and we haven't cleaned it up until now, so remain a @@ -376,16 +361,15 @@ static void basic_do_promote(void *args, int success) { gpr_mu_unlock(&pollset->mu); - if (do_shutdown_cb) { - pollset->shutdown_done_cb(pollset->shutdown_done_arg); - } - /* Matching ref in basic_pollset_add_fd */ GRPC_FD_UNREF(fd, "basicpoll_add"); + + grpc_call_list_run(call_list); } static void basic_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd, - int and_unlock_pollset) { + int and_unlock_pollset, + grpc_call_list *call_list) { grpc_unary_promote_args *up_args; GPR_ASSERT(fd); if (fd == pollset->data.ptr) goto exit; @@ -402,7 +386,8 @@ static void basic_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd, 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_platform_become_multipoller(pollset, fds, GPR_ARRAY_SIZE(fds), + call_list); GRPC_FD_UNREF(fds[0], "basicpoll"); } else { /* old fd is orphaned and we haven't cleaned it up until now, so remain a @@ -424,7 +409,7 @@ static void basic_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd, up_args->promotion_closure.cb = basic_do_promote; up_args->promotion_closure.cb_arg = up_args; - grpc_pollset_add_idle_job(pollset, &up_args->promotion_closure); + grpc_call_list_add(&pollset->idle_jobs, &up_args->promotion_closure, 1); grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST); exit: @@ -434,7 +419,8 @@ exit: } static void basic_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd, - int and_unlock_pollset) { + int and_unlock_pollset, + grpc_call_list *call_list) { GPR_ASSERT(fd); if (fd == pollset->data.ptr) { GRPC_FD_UNREF(pollset->data.ptr, "basicpoll"); @@ -446,10 +432,11 @@ static void basic_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd, } } -static void basic_pollset_maybe_work(grpc_pollset *pollset, - grpc_pollset_worker *worker, - gpr_timespec deadline, gpr_timespec now, - int allow_synchronous_callback) { +static void basic_pollset_maybe_work_and_unlock(grpc_pollset *pollset, + grpc_pollset_worker *worker, + gpr_timespec deadline, + gpr_timespec now, + grpc_call_list *call_list) { struct pollfd pfd[2]; grpc_fd *fd; grpc_fd_watcher fd_watcher; @@ -487,7 +474,7 @@ static void basic_pollset_maybe_work(grpc_pollset *pollset, if (fd) { grpc_fd_end_poll(&fd_watcher, pfd[1].revents & POLLIN, - pfd[1].revents & POLLOUT); + pfd[1].revents & POLLOUT, call_list); } if (r < 0) { @@ -502,15 +489,13 @@ static void basic_pollset_maybe_work(grpc_pollset *pollset, } if (nfds > 1) { if (pfd[1].revents & (POLLIN | POLLHUP | POLLERR)) { - grpc_fd_become_readable(fd, allow_synchronous_callback); + grpc_fd_become_readable(fd, call_list); } if (pfd[1].revents & (POLLOUT | POLLHUP | POLLERR)) { - grpc_fd_become_writable(fd, allow_synchronous_callback); + grpc_fd_become_writable(fd, call_list); } } } - - gpr_mu_lock(&pollset->mu); } static void basic_pollset_destroy(grpc_pollset *pollset) { @@ -521,8 +506,9 @@ static void basic_pollset_destroy(grpc_pollset *pollset) { } static const grpc_pollset_vtable basic_pollset = { - basic_pollset_add_fd, basic_pollset_del_fd, basic_pollset_maybe_work, - basic_pollset_destroy, basic_pollset_destroy}; + basic_pollset_add_fd, basic_pollset_del_fd, + basic_pollset_maybe_work_and_unlock, basic_pollset_destroy, + basic_pollset_destroy}; static void become_basic_pollset(grpc_pollset *pollset, grpc_fd *fd_or_null) { pollset->vtable = &basic_pollset; diff --git a/src/core/iomgr/pollset_posix.h b/src/core/iomgr/pollset_posix.h index 4f09f870f7..c65f25fd18 100644 --- a/src/core/iomgr/pollset_posix.h +++ b/src/core/iomgr/pollset_posix.h @@ -65,10 +65,8 @@ typedef struct grpc_pollset { int shutting_down; int called_shutdown; int kicked_without_pollers; - void (*shutdown_done_cb)(void *arg); - void *shutdown_done_arg; - grpc_closure *unlock_jobs; - grpc_closure *idle_jobs; + grpc_closure *shutdown_done; + grpc_call_list idle_jobs; union { int fd; void *ptr; @@ -77,12 +75,13 @@ typedef struct grpc_pollset { struct grpc_pollset_vtable { void (*add_fd)(grpc_pollset *pollset, struct grpc_fd *fd, - int and_unlock_pollset); + int and_unlock_pollset, grpc_call_list *call_list); void (*del_fd)(grpc_pollset *pollset, struct grpc_fd *fd, - int and_unlock_pollset); - void (*maybe_work)(grpc_pollset *pollset, grpc_pollset_worker *worker, - gpr_timespec deadline, gpr_timespec now, - int allow_synchronous_callback); + int and_unlock_pollset, grpc_call_list *call_list); + void (*maybe_work_and_unlock)(grpc_pollset *pollset, + grpc_pollset_worker *worker, + gpr_timespec deadline, gpr_timespec now, + grpc_call_list *call_list); void (*finish_shutdown)(grpc_pollset *pollset); void (*destroy)(grpc_pollset *pollset); }; @@ -90,10 +89,12 @@ struct grpc_pollset_vtable { #define GRPC_POLLSET_MU(pollset) (&(pollset)->mu) /* Add an fd to a pollset */ -void grpc_pollset_add_fd(grpc_pollset *pollset, struct grpc_fd *fd); +void grpc_pollset_add_fd(grpc_pollset *pollset, struct grpc_fd *fd, + grpc_call_list *call_list); /* Force remove an fd from a pollset (normally they are removed on the next poll after an fd is orphaned) */ -void grpc_pollset_del_fd(grpc_pollset *pollset, struct grpc_fd *fd); +void grpc_pollset_del_fd(grpc_pollset *pollset, struct grpc_fd *fd, + grpc_call_list *call_list); /* Returns the fd to listen on for kicks */ int grpc_kick_read_fd(grpc_pollset *p); @@ -111,13 +112,13 @@ int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline, gpr_timespec now); /* turn a pollset into a multipoller: platform specific */ -typedef void (*grpc_platform_become_multipoller_type)(grpc_pollset *pollset, - struct grpc_fd **fds, - size_t fd_count); +typedef void (*grpc_platform_become_multipoller_type)( + grpc_pollset *pollset, struct grpc_fd **fds, size_t fd_count, + grpc_call_list *call_list); extern grpc_platform_become_multipoller_type grpc_platform_become_multipoller; void grpc_poll_become_multipoller(grpc_pollset *pollset, struct grpc_fd **fds, - size_t fd_count); + size_t fd_count, grpc_call_list *call_list); /* Return 1 if the pollset has active threads in grpc_pollset_work (pollset must * be locked) */ @@ -127,9 +128,4 @@ int grpc_pollset_has_workers(grpc_pollset *pollset); typedef int (*grpc_poll_function_type)(struct pollfd *, nfds_t, int); extern grpc_poll_function_type grpc_poll_function; -/** schedule a closure to be run next time there are no active workers */ -void grpc_pollset_add_idle_job(grpc_pollset *pollset, grpc_closure *closure); -/** schedule a closure to be run next time the pollset is unlocked */ -void grpc_pollset_add_unlock_job(grpc_pollset *pollset, grpc_closure *closure); - #endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_POSIX_H */ diff --git a/src/core/iomgr/pollset_set.h b/src/core/iomgr/pollset_set.h index 6d73951c70..ca667bf34e 100644 --- a/src/core/iomgr/pollset_set.h +++ b/src/core/iomgr/pollset_set.h @@ -52,8 +52,10 @@ 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); + grpc_pollset *pollset, + grpc_call_list *call_list); void grpc_pollset_set_del_pollset(grpc_pollset_set *pollset_set, - grpc_pollset *pollset); + grpc_pollset *pollset, + grpc_call_list *call_list); #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 index 2076ac70ef..a2c1f47629 100644 --- a/src/core/iomgr/pollset_set_posix.c +++ b/src/core/iomgr/pollset_set_posix.c @@ -59,7 +59,8 @@ void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set) { } void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, - grpc_pollset *pollset) { + grpc_pollset *pollset, + grpc_call_list *call_list) { size_t i, j; gpr_mu_lock(&pollset_set->mu); if (pollset_set->pollset_count == pollset_set->pollset_capacity) { @@ -74,7 +75,7 @@ void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, if (grpc_fd_is_orphaned(pollset_set->fds[i])) { GRPC_FD_UNREF(pollset_set->fds[i], "pollset"); } else { - grpc_pollset_add_fd(pollset, pollset_set->fds[i]); + grpc_pollset_add_fd(pollset, pollset_set->fds[i], call_list); pollset_set->fds[j++] = pollset_set->fds[i]; } } @@ -83,7 +84,8 @@ void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, } void grpc_pollset_set_del_pollset(grpc_pollset_set *pollset_set, - grpc_pollset *pollset) { + grpc_pollset *pollset, + grpc_call_list *call_list) { size_t i; gpr_mu_lock(&pollset_set->mu); for (i = 0; i < pollset_set->pollset_count; i++) { @@ -97,7 +99,8 @@ void grpc_pollset_set_del_pollset(grpc_pollset_set *pollset_set, gpr_mu_unlock(&pollset_set->mu); } -void grpc_pollset_set_add_fd(grpc_pollset_set *pollset_set, grpc_fd *fd) { +void grpc_pollset_set_add_fd(grpc_pollset_set *pollset_set, grpc_fd *fd, + grpc_call_list *call_list) { size_t i; gpr_mu_lock(&pollset_set->mu); if (pollset_set->fd_count == pollset_set->fd_capacity) { @@ -108,12 +111,13 @@ void grpc_pollset_set_add_fd(grpc_pollset_set *pollset_set, grpc_fd *fd) { GRPC_FD_REF(fd, "pollset_set"); 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); + grpc_pollset_add_fd(pollset_set->pollsets[i], fd, call_list); } gpr_mu_unlock(&pollset_set->mu); } -void grpc_pollset_set_del_fd(grpc_pollset_set *pollset_set, grpc_fd *fd) { +void grpc_pollset_set_del_fd(grpc_pollset_set *pollset_set, grpc_fd *fd, + grpc_call_list *call_list) { size_t i; gpr_mu_lock(&pollset_set->mu); for (i = 0; i < pollset_set->fd_count; i++) { diff --git a/src/core/iomgr/pollset_set_posix.h b/src/core/iomgr/pollset_set_posix.h index e88740bde1..e9969d2c0f 100644 --- a/src/core/iomgr/pollset_set_posix.h +++ b/src/core/iomgr/pollset_set_posix.h @@ -49,7 +49,9 @@ typedef struct grpc_pollset_set { 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); +void grpc_pollset_set_add_fd(grpc_pollset_set *pollset_set, grpc_fd *fd, + grpc_call_list *call_list); +void grpc_pollset_set_del_fd(grpc_pollset_set *pollset_set, grpc_fd *fd, + grpc_call_list *call_list); #endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_WINDOWS_H */ diff --git a/src/core/iomgr/resolve_address.h b/src/core/iomgr/resolve_address.h index 9f361cb892..ec8d83fffa 100644 --- a/src/core/iomgr/resolve_address.h +++ b/src/core/iomgr/resolve_address.h @@ -34,7 +34,7 @@ #ifndef GRPC_INTERNAL_CORE_IOMGR_RESOLVE_ADDRESS_H #define GRPC_INTERNAL_CORE_IOMGR_RESOLVE_ADDRESS_H -#include <stddef.h> +#include "src/core/iomgr/iomgr.h" #define GRPC_MAX_SOCKADDR_SIZE 128 @@ -52,7 +52,8 @@ typedef struct { On success: addresses is the result, and the callee must call grpc_resolved_addresses_destroy when it's done with them On failure: addresses is NULL */ -typedef void (*grpc_resolve_cb)(void *arg, grpc_resolved_addresses *addresses); +typedef void (*grpc_resolve_cb)(void *arg, grpc_resolved_addresses *addresses, + grpc_call_list *call_list); /* Asynchronously resolve addr. Use default_port if a port isn't designated in addr, otherwise use the port in addr. */ /* TODO(ctiller): add a timeout here */ diff --git a/src/core/iomgr/resolve_address_posix.c b/src/core/iomgr/resolve_address_posix.c index ce6972b797..ea64f8176f 100644 --- a/src/core/iomgr/resolve_address_posix.c +++ b/src/core/iomgr/resolve_address_posix.c @@ -144,17 +144,19 @@ done: } /* Thread function to asynch-ify grpc_blocking_resolve_address */ -static void do_request(void *rp) { +static void do_request_thread(void *rp) { request *r = rp; + grpc_call_list call_list = GRPC_CALL_LIST_INIT; grpc_resolved_addresses *resolved = grpc_blocking_resolve_address(r->name, r->default_port); void *arg = r->arg; grpc_resolve_cb cb = r->cb; gpr_free(r->name); gpr_free(r->default_port); - cb(arg, resolved); + cb(arg, resolved, &call_list); grpc_iomgr_unregister_object(&r->iomgr_object); gpr_free(r); + grpc_call_list_run(&call_list); } void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addrs) { @@ -175,7 +177,7 @@ void grpc_resolve_address(const char *name, const char *default_port, 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_thread, r, NULL); } #endif diff --git a/src/core/iomgr/tcp_client.h b/src/core/iomgr/tcp_client.h index 57f80016c2..8fb5bfda1d 100644 --- a/src/core/iomgr/tcp_client.h +++ b/src/core/iomgr/tcp_client.h @@ -44,10 +44,9 @@ NULL on failure). interested_parties points to a set of pollsets that would be interested in this connection being established (in order to continue their work) */ -void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), - void *arg, grpc_pollset_set *interested_parties, - grpc_workqueue *workqueue, +void grpc_tcp_client_connect(grpc_closure *on_connect, grpc_endpoint **endpoint, + grpc_pollset_set *interested_parties, const struct sockaddr *addr, size_t addr_len, - gpr_timespec deadline); + gpr_timespec deadline, grpc_call_list *call_list); #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 1ea2155060..a4828201c7 100644 --- a/src/core/iomgr/tcp_client_posix.c +++ b/src/core/iomgr/tcp_client_posix.c @@ -57,8 +57,6 @@ extern int grpc_tcp_trace; typedef struct { - void (*cb)(void *arg, grpc_endpoint *tcp); - void *cb_arg; gpr_mu mu; grpc_fd *fd; gpr_timespec deadline; @@ -67,6 +65,8 @@ typedef struct { grpc_closure write_closure; grpc_pollset_set *interested_parties; char *addr_str; + grpc_endpoint **ep; + grpc_closure *closure; } async_connect; static int prepare_socket(const struct sockaddr *addr, int fd) { @@ -91,7 +91,7 @@ error: return 0; } -static void tc_on_alarm(void *acp, int success) { +static void tc_on_alarm(void *acp, int success, grpc_call_list *call_list) { int done; async_connect *ac = acp; if (grpc_tcp_trace) { @@ -100,7 +100,7 @@ static void tc_on_alarm(void *acp, int success) { } gpr_mu_lock(&ac->mu); if (ac->fd != NULL) { - grpc_fd_shutdown(ac->fd); + grpc_fd_shutdown(ac->fd, call_list); } done = (--ac->refs == 0); gpr_mu_unlock(&ac->mu); @@ -111,15 +111,14 @@ static void tc_on_alarm(void *acp, int success) { } } -static void on_writable(void *acp, int success) { +static void on_writable(void *acp, int success, grpc_call_list *call_list) { async_connect *ac = acp; int so_error = 0; socklen_t so_error_size; int err; int done; - grpc_endpoint *ep = NULL; - void (*cb)(void *arg, grpc_endpoint *tcp) = ac->cb; - void *cb_arg = ac->cb_arg; + grpc_endpoint **ep = ac->ep; + grpc_closure *closure = ac->closure; grpc_fd *fd; if (grpc_tcp_trace) { @@ -133,7 +132,7 @@ static void on_writable(void *acp, int success) { ac->fd = NULL; gpr_mu_unlock(&ac->mu); - grpc_alarm_cancel(&ac->alarm); + grpc_alarm_cancel(&ac->alarm, call_list); gpr_mu_lock(&ac->mu); if (success) { @@ -162,7 +161,7 @@ static void on_writable(void *acp, int success) { don't do that! */ gpr_log(GPR_ERROR, "kernel out of buffers"); gpr_mu_unlock(&ac->mu); - grpc_fd_notify_on_write(fd, &ac->write_closure); + grpc_fd_notify_on_write(fd, &ac->write_closure, call_list); return; } else { switch (so_error) { @@ -176,8 +175,8 @@ static void on_writable(void *acp, int success) { goto finish; } } else { - grpc_pollset_set_del_fd(ac->interested_parties, fd); - ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str); + grpc_pollset_set_del_fd(ac->interested_parties, fd, call_list); + *ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str); fd = NULL; goto finish; } @@ -190,8 +189,8 @@ static void on_writable(void *acp, int success) { finish: if (fd != NULL) { - grpc_pollset_set_del_fd(ac->interested_parties, fd); - grpc_fd_orphan(fd, NULL, "tcp_client_orphan"); + grpc_pollset_set_del_fd(ac->interested_parties, fd, call_list); + grpc_fd_orphan(fd, NULL, "tcp_client_orphan", call_list); fd = NULL; } done = (--ac->refs == 0); @@ -201,14 +200,14 @@ finish: gpr_free(ac->addr_str); gpr_free(ac); } - cb(cb_arg, ep); + grpc_call_list_add(call_list, closure, *ep != NULL); } -void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), - void *arg, grpc_pollset_set *interested_parties, +void grpc_tcp_client_connect(grpc_closure *closure, grpc_endpoint **ep, + grpc_pollset_set *interested_parties, grpc_workqueue *workqueue, const struct sockaddr *addr, size_t addr_len, - gpr_timespec deadline) { + gpr_timespec deadline, grpc_call_list *call_list) { int fd; grpc_dualstack_mode dsmode; int err; @@ -219,6 +218,8 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), char *name; char *addr_str; + *ep = NULL; + /* Use dualstack sockets where available. */ if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) { addr = (const struct sockaddr *)&addr6_v4mapped; @@ -236,7 +237,7 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), addr_len = sizeof(addr4_copy); } if (!prepare_socket(addr, fd)) { - cb(arg, NULL); + grpc_call_list_add(call_list, closure, 0); return; } @@ -248,25 +249,26 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), addr_str = grpc_sockaddr_to_uri(addr); gpr_asprintf(&name, "tcp-client:%s", addr_str); - fdobj = grpc_fd_create(fd, workqueue, name); + fdobj = grpc_fd_create(fd, name); if (err >= 0) { - cb(arg, grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str)); + *ep = grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str); + grpc_call_list_add(call_list, closure, 1); goto done; } if (errno != EWOULDBLOCK && errno != EINPROGRESS) { gpr_log(GPR_ERROR, "connect error to '%s': %s", addr_str, strerror(errno)); - grpc_fd_orphan(fdobj, NULL, "tcp_client_connect_error"); - cb(arg, NULL); + grpc_fd_orphan(fdobj, NULL, "tcp_client_connect_error", call_list); + grpc_call_list_add(call_list, closure, 0); goto done; } - grpc_pollset_set_add_fd(interested_parties, fdobj); + grpc_pollset_set_add_fd(interested_parties, fdobj, call_list); ac = gpr_malloc(sizeof(async_connect)); - ac->cb = cb; - ac->cb_arg = arg; + ac->closure = closure; + ac->ep = ep; ac->fd = fdobj; ac->interested_parties = interested_parties; ac->addr_str = addr_str; @@ -284,8 +286,8 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), gpr_mu_lock(&ac->mu); grpc_alarm_init(&ac->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), - tc_on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC)); - grpc_fd_notify_on_write(ac->fd, &ac->write_closure); + tc_on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC), call_list); + grpc_fd_notify_on_write(ac->fd, &ac->write_closure, call_list); gpr_mu_unlock(&ac->mu); done: diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c index 374d2f3a40..89a85f26dc 100644 --- a/src/core/iomgr/tcp_posix.c +++ b/src/core/iomgr/tcp_posix.c @@ -94,30 +94,33 @@ typedef struct { char *peer_string; } grpc_tcp; -static void tcp_handle_read(void *arg /* grpc_tcp */, int success); -static void tcp_handle_write(void *arg /* grpc_tcp */, int success); +static void tcp_handle_read(void *arg /* grpc_tcp */, int success, + grpc_call_list *call_list); +static void tcp_handle_write(void *arg /* grpc_tcp */, int success, + grpc_call_list *call_list); -static void tcp_shutdown(grpc_endpoint *ep) { +static void tcp_shutdown(grpc_endpoint *ep, grpc_call_list *call_list) { grpc_tcp *tcp = (grpc_tcp *)ep; - grpc_fd_shutdown(tcp->em_fd); + grpc_fd_shutdown(tcp->em_fd, call_list); } -static void tcp_free(grpc_tcp *tcp) { - grpc_fd_orphan(tcp->em_fd, NULL, "tcp_unref_orphan"); +static void tcp_free(grpc_tcp *tcp, grpc_call_list *call_list) { + grpc_fd_orphan(tcp->em_fd, NULL, "tcp_unref_orphan", call_list); gpr_free(tcp->peer_string); gpr_free(tcp); } /*#define GRPC_TCP_REFCOUNT_DEBUG*/ #ifdef GRPC_TCP_REFCOUNT_DEBUG -#define TCP_UNREF(tcp, reason) tcp_unref((tcp), (reason), __FILE__, __LINE__) +#define TCP_UNREF(tcp, reason, cl) \ + tcp_unref((tcp), (cl), (reason), __FILE__, __LINE__) #define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__) -static void tcp_unref(grpc_tcp *tcp, const char *reason, const char *file, - int line) { +static void tcp_unref(grpc_tcp *tcp, grpc_call_list *call_list, + const char *reason, const char *file, int line) { gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %d -> %d", tcp, reason, tcp->refcount.count, tcp->refcount.count - 1); if (gpr_unref(&tcp->refcount)) { - tcp_free(tcp); + tcp_free(tcp, call_list); } } @@ -128,23 +131,24 @@ static void tcp_ref(grpc_tcp *tcp, const char *reason, const char *file, gpr_ref(&tcp->refcount); } #else -#define TCP_UNREF(tcp, reason) tcp_unref((tcp)) +#define TCP_UNREF(tcp, reason, cl) tcp_unref((tcp), (cl)) #define TCP_REF(tcp, reason) tcp_ref((tcp)) -static void tcp_unref(grpc_tcp *tcp) { +static void tcp_unref(grpc_tcp *tcp, grpc_call_list *call_list) { if (gpr_unref(&tcp->refcount)) { - tcp_free(tcp); + tcp_free(tcp, call_list); } } static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); } #endif -static void tcp_destroy(grpc_endpoint *ep) { +static void tcp_destroy(grpc_endpoint *ep, grpc_call_list *call_list) { grpc_tcp *tcp = (grpc_tcp *)ep; - TCP_UNREF(tcp, "destroy"); + TCP_UNREF(tcp, "destroy", call_list); } -static void call_read_cb(grpc_tcp *tcp, int success) { +static void call_read_cb(grpc_tcp *tcp, int success, + grpc_call_list *call_list) { grpc_closure *cb = tcp->read_cb; if (grpc_tcp_trace) { @@ -160,11 +164,11 @@ static void call_read_cb(grpc_tcp *tcp, int success) { tcp->read_cb = NULL; tcp->incoming_buffer = NULL; - cb->cb(cb->cb_arg, success); + cb->cb(cb->cb_arg, success, call_list); } #define MAX_READ_IOVEC 4 -static void tcp_continue_read(grpc_tcp *tcp) { +static void tcp_continue_read(grpc_tcp *tcp, grpc_call_list *call_list) { struct msghdr msg; struct iovec iov[MAX_READ_IOVEC]; ssize_t read_bytes; @@ -206,18 +210,18 @@ static void tcp_continue_read(grpc_tcp *tcp) { tcp->iov_size /= 2; } /* We've consumed the edge, request a new one */ - grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure); + grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure, call_list); } else { /* TODO(klempner): Log interesting errors */ gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer); - call_read_cb(tcp, 0); - TCP_UNREF(tcp, "read"); + call_read_cb(tcp, 0, call_list); + TCP_UNREF(tcp, "read", call_list); } } else if (read_bytes == 0) { /* 0 read size ==> end of stream */ gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer); - call_read_cb(tcp, 0); - TCP_UNREF(tcp, "read"); + call_read_cb(tcp, 0, call_list); + TCP_UNREF(tcp, "read", call_list); } else { GPR_ASSERT((size_t)read_bytes <= tcp->incoming_buffer->length); if ((size_t)read_bytes < tcp->incoming_buffer->length) { @@ -228,29 +232,29 @@ static void tcp_continue_read(grpc_tcp *tcp) { ++tcp->iov_size; } GPR_ASSERT((size_t)read_bytes == tcp->incoming_buffer->length); - call_read_cb(tcp, 1); - TCP_UNREF(tcp, "read"); + call_read_cb(tcp, 1, call_list); + TCP_UNREF(tcp, "read", call_list); } GRPC_TIMER_END(GRPC_PTAG_HANDLE_READ, 0); } -static void tcp_handle_read(void *arg /* grpc_tcp */, int success) { +static void tcp_handle_read(void *arg /* grpc_tcp */, int success, + grpc_call_list *call_list) { grpc_tcp *tcp = (grpc_tcp *)arg; GPR_ASSERT(!tcp->finished_edge); if (!success) { gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer); - call_read_cb(tcp, 0); - TCP_UNREF(tcp, "read"); + call_read_cb(tcp, 0, call_list); + TCP_UNREF(tcp, "read", call_list); } else { - tcp_continue_read(tcp); + tcp_continue_read(tcp, call_list); } } -static grpc_endpoint_op_status tcp_read(grpc_endpoint *ep, - gpr_slice_buffer *incoming_buffer, - grpc_closure *cb) { +static void tcp_read(grpc_endpoint *ep, gpr_slice_buffer *incoming_buffer, + grpc_closure *cb, grpc_call_list *call_list) { grpc_tcp *tcp = (grpc_tcp *)ep; GPR_ASSERT(tcp->read_cb == NULL); tcp->read_cb = cb; @@ -259,16 +263,16 @@ static grpc_endpoint_op_status tcp_read(grpc_endpoint *ep, TCP_REF(tcp, "read"); if (tcp->finished_edge) { tcp->finished_edge = 0; - grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure); + grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure, call_list); } else { - grpc_workqueue_push(tcp->em_fd->workqueue, &tcp->read_closure, 1); + grpc_call_list_add(call_list, &tcp->read_closure, 1); } - /* TODO(ctiller): immediate return */ - return GRPC_ENDPOINT_PENDING; } +typedef enum { FLUSH_DONE, FLUSH_PENDING, FLUSH_ERROR } flush_result; + #define MAX_WRITE_IOVEC 16 -static grpc_endpoint_op_status tcp_flush(grpc_tcp *tcp) { +static flush_result tcp_flush(grpc_tcp *tcp) { struct msghdr msg; struct iovec iov[MAX_WRITE_IOVEC]; msg_iovlen_type iov_size; @@ -318,10 +322,10 @@ static grpc_endpoint_op_status tcp_flush(grpc_tcp *tcp) { if (errno == EAGAIN) { tcp->outgoing_slice_idx = unwind_slice_idx; tcp->outgoing_byte_idx = unwind_byte_idx; - return GRPC_ENDPOINT_PENDING; + return FLUSH_PENDING; } else { /* TODO(klempner): Log some of these */ - return GRPC_ENDPOINT_ERROR; + return FLUSH_ERROR; } } @@ -342,42 +346,42 @@ static grpc_endpoint_op_status tcp_flush(grpc_tcp *tcp) { } if (tcp->outgoing_slice_idx == tcp->outgoing_buffer->count) { - return GRPC_ENDPOINT_DONE; + return FLUSH_DONE; } }; } -static void tcp_handle_write(void *arg /* grpc_tcp */, int success) { +static void tcp_handle_write(void *arg /* grpc_tcp */, int success, + grpc_call_list *call_list) { grpc_tcp *tcp = (grpc_tcp *)arg; - grpc_endpoint_op_status status; + flush_result status; grpc_closure *cb; if (!success) { cb = tcp->write_cb; tcp->write_cb = NULL; - cb->cb(cb->cb_arg, 0); - TCP_UNREF(tcp, "write"); + cb->cb(cb->cb_arg, 0, call_list); + TCP_UNREF(tcp, "write", call_list); return; } GRPC_TIMER_BEGIN(GRPC_PTAG_TCP_CB_WRITE, 0); status = tcp_flush(tcp); - if (status == GRPC_ENDPOINT_PENDING) { - grpc_fd_notify_on_write(tcp->em_fd, &tcp->write_closure); + if (status == FLUSH_PENDING) { + grpc_fd_notify_on_write(tcp->em_fd, &tcp->write_closure, call_list); } else { cb = tcp->write_cb; tcp->write_cb = NULL; - cb->cb(cb->cb_arg, status == GRPC_ENDPOINT_DONE); - TCP_UNREF(tcp, "write"); + cb->cb(cb->cb_arg, status == FLUSH_DONE, call_list); + TCP_UNREF(tcp, "write", call_list); } GRPC_TIMER_END(GRPC_PTAG_TCP_CB_WRITE, 0); } -static grpc_endpoint_op_status tcp_write(grpc_endpoint *ep, - gpr_slice_buffer *buf, - grpc_closure *cb) { +static void tcp_write(grpc_endpoint *ep, gpr_slice_buffer *buf, + grpc_closure *cb, grpc_call_list *call_list) { grpc_tcp *tcp = (grpc_tcp *)ep; - grpc_endpoint_op_status status; + flush_result status; if (grpc_tcp_trace) { size_t i; @@ -395,32 +399,36 @@ static grpc_endpoint_op_status tcp_write(grpc_endpoint *ep, if (buf->length == 0) { GRPC_TIMER_END(GRPC_PTAG_TCP_WRITE, 0); - return GRPC_ENDPOINT_DONE; + grpc_call_list_add(call_list, cb, 1); + return; } tcp->outgoing_buffer = buf; tcp->outgoing_slice_idx = 0; tcp->outgoing_byte_idx = 0; status = tcp_flush(tcp); - if (status == GRPC_ENDPOINT_PENDING) { + if (status == FLUSH_PENDING) { TCP_REF(tcp, "write"); tcp->write_cb = cb; - grpc_fd_notify_on_write(tcp->em_fd, &tcp->write_closure); + grpc_fd_notify_on_write(tcp->em_fd, &tcp->write_closure, call_list); + } else { + grpc_call_list_add(call_list, cb, status == FLUSH_DONE); } GRPC_TIMER_END(GRPC_PTAG_TCP_WRITE, 0); - return status; } -static void tcp_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { +static void tcp_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset, + grpc_call_list *call_list) { grpc_tcp *tcp = (grpc_tcp *)ep; - grpc_pollset_add_fd(pollset, tcp->em_fd); + grpc_pollset_add_fd(pollset, tcp->em_fd, call_list); } static void tcp_add_to_pollset_set(grpc_endpoint *ep, - grpc_pollset_set *pollset_set) { + grpc_pollset_set *pollset_set, + grpc_call_list *call_list) { grpc_tcp *tcp = (grpc_tcp *)ep; - grpc_pollset_set_add_fd(pollset_set, tcp->em_fd); + grpc_pollset_set_add_fd(pollset_set, tcp->em_fd, call_list); } static char *tcp_get_peer(grpc_endpoint *ep) { diff --git a/src/core/iomgr/tcp_server.h b/src/core/iomgr/tcp_server.h index 5165f5c5ca..19e58bc294 100644 --- a/src/core/iomgr/tcp_server.h +++ b/src/core/iomgr/tcp_server.h @@ -40,7 +40,8 @@ typedef struct grpc_tcp_server grpc_tcp_server; /* New server callback: tcp is the newly connected tcp connection */ -typedef void (*grpc_tcp_server_cb)(void *arg, grpc_endpoint *ep); +typedef void (*grpc_tcp_server_cb)(void *arg, grpc_endpoint *ep, + grpc_call_list *call_list); /* Create a server, initially not bound to any ports */ grpc_tcp_server *grpc_tcp_server_create(void); @@ -48,7 +49,7 @@ grpc_tcp_server *grpc_tcp_server_create(void); /* Start listening to bound ports */ void grpc_tcp_server_start(grpc_tcp_server *server, grpc_pollset **pollsets, size_t pollset_count, grpc_tcp_server_cb cb, - void *cb_arg); + void *cb_arg, grpc_call_list *call_list); /* Add a port to the server, returning port number on success, or negative on failure. @@ -71,8 +72,7 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, up when grpc_tcp_server_destroy is called. */ int grpc_tcp_server_get_fd(grpc_tcp_server *s, unsigned index); -void grpc_tcp_server_destroy(grpc_tcp_server *server, - void (*shutdown_done)(void *shutdown_done_arg), - void *shutdown_done_arg); +void grpc_tcp_server_destroy(grpc_tcp_server *server, grpc_closure *closure, + grpc_call_list *call_list); #endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_SERVER_H */ diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c index 213b2e1113..0c5e0053dd 100644 --- a/src/core/iomgr/tcp_server_posix.c +++ b/src/core/iomgr/tcp_server_posix.c @@ -117,16 +117,12 @@ struct grpc_tcp_server { size_t port_capacity; /* shutdown callback */ - void (*shutdown_complete)(void *); - void *shutdown_complete_arg; + grpc_closure *shutdown_complete; /* all pollsets interested in new connections */ grpc_pollset **pollsets; /* number of pollsets in the pollsets array */ size_t pollset_count; - - /** workqueue for interally created async work */ - grpc_workqueue *workqueue; }; grpc_tcp_server *grpc_tcp_server_create(void) { @@ -140,40 +136,37 @@ grpc_tcp_server *grpc_tcp_server_create(void) { s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP); s->nports = 0; s->port_capacity = INIT_PORT_CAP; - s->workqueue = grpc_workqueue_create(); return s; } -static void finish_shutdown(grpc_tcp_server *s) { - s->shutdown_complete(s->shutdown_complete_arg); - s->shutdown_complete = NULL; +static void finish_shutdown(grpc_tcp_server *s, grpc_call_list *call_list) { + grpc_call_list_add(call_list, s->shutdown_complete, 1); gpr_mu_destroy(&s->mu); gpr_free(s->ports); - GRPC_WORKQUEUE_UNREF(s->workqueue, "destroy"); gpr_free(s); } -static void destroyed_port(void *server, int success) { +static void destroyed_port(void *server, int success, + grpc_call_list *call_list) { grpc_tcp_server *s = server; gpr_mu_lock(&s->mu); s->destroyed_ports++; if (s->destroyed_ports == s->nports) { gpr_mu_unlock(&s->mu); - finish_shutdown(s); + finish_shutdown(s, call_list); } else { GPR_ASSERT(s->destroyed_ports < s->nports); gpr_mu_unlock(&s->mu); } } -static void dont_care_about_shutdown_completion(void *ignored) {} - /* called when all listening endpoints have been shutdown, so no further events will be received on them - at this point it's safe to destroy things */ -static void deactivated_all_ports(grpc_tcp_server *s) { +static void deactivated_all_ports(grpc_tcp_server *s, + grpc_call_list *call_list) { size_t i; /* delete ALL the things */ @@ -192,38 +185,35 @@ static void deactivated_all_ports(grpc_tcp_server *s) { } sp->destroyed_closure.cb = destroyed_port; sp->destroyed_closure.cb_arg = s; - grpc_fd_orphan(sp->emfd, &sp->destroyed_closure, "tcp_listener_shutdown"); + grpc_fd_orphan(sp->emfd, &sp->destroyed_closure, "tcp_listener_shutdown", + call_list); } gpr_mu_unlock(&s->mu); } else { gpr_mu_unlock(&s->mu); - finish_shutdown(s); + finish_shutdown(s, call_list); } } -void grpc_tcp_server_destroy( - grpc_tcp_server *s, void (*shutdown_complete)(void *shutdown_complete_arg), - void *shutdown_complete_arg) { +void grpc_tcp_server_destroy(grpc_tcp_server *s, grpc_closure *closure, + grpc_call_list *call_list) { 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; + s->shutdown_complete = closure; /* shutdown all fd's */ if (s->active_ports) { for (i = 0; i < s->nports; i++) { - grpc_fd_shutdown(s->ports[i].emfd); + grpc_fd_shutdown(s->ports[i].emfd, call_list); } gpr_mu_unlock(&s->mu); } else { gpr_mu_unlock(&s->mu); - deactivated_all_ports(s); + deactivated_all_ports(s, call_list); } } @@ -308,7 +298,7 @@ error: } /* event manager callback when reads are ready */ -static void on_read(void *arg, int success) { +static void on_read(void *arg, int success, grpc_call_list *call_list) { server_port *sp = arg; grpc_fd *fdobj; size_t i; @@ -331,7 +321,7 @@ static void on_read(void *arg, int success) { case EINTR: continue; case EAGAIN: - grpc_fd_notify_on_read(sp->emfd, &sp->read_closure); + grpc_fd_notify_on_read(sp->emfd, &sp->read_closure, call_list); return; default: gpr_log(GPR_ERROR, "Failed accept4: %s", strerror(errno)); @@ -348,16 +338,17 @@ static void on_read(void *arg, int success) { gpr_log(GPR_DEBUG, "SERVER_CONNECT: incoming connection: %s", addr_str); } - fdobj = grpc_fd_create(fd, sp->server->workqueue, name); + fdobj = grpc_fd_create(fd, name); /* 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); + grpc_pollset_add_fd(sp->server->pollsets[i], fdobj, call_list); } sp->server->cb( sp->server->cb_arg, - grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str)); + grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str), + call_list); gpr_free(name); gpr_free(addr_str); @@ -369,7 +360,7 @@ error: gpr_mu_lock(&sp->server->mu); if (0 == --sp->server->active_ports) { gpr_mu_unlock(&sp->server->mu); - deactivated_all_ports(sp->server); + deactivated_all_ports(sp->server, call_list); } else { gpr_mu_unlock(&sp->server->mu); } @@ -396,7 +387,7 @@ static int add_socket_to_server(grpc_tcp_server *s, int fd, sp = &s->ports[s->nports++]; sp->server = s; sp->fd = fd; - sp->emfd = grpc_fd_create(fd, s->workqueue, name); + sp->emfd = grpc_fd_create(fd, name); memcpy(sp->addr.untyped, addr, addr_len); sp->addr_len = addr_len; GPR_ASSERT(sp->emfd); @@ -495,7 +486,7 @@ int grpc_tcp_server_get_fd(grpc_tcp_server *s, unsigned index) { void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollsets, size_t pollset_count, grpc_tcp_server_cb cb, - void *cb_arg) { + void *cb_arg, grpc_call_list *call_list) { size_t i, j; GPR_ASSERT(cb); gpr_mu_lock(&s->mu); @@ -507,11 +498,12 @@ void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **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); + grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd, call_list); } s->ports[i].read_closure.cb = on_read; s->ports[i].read_closure.cb_arg = &s->ports[i]; - grpc_fd_notify_on_read(s->ports[i].emfd, &s->ports[i].read_closure); + grpc_fd_notify_on_read(s->ports[i].emfd, &s->ports[i].read_closure, + call_list); s->active_ports++; } gpr_mu_unlock(&s->mu); diff --git a/src/core/iomgr/udp_server.c b/src/core/iomgr/udp_server.c index 1a1d812050..e1c2ae95fd 100644 --- a/src/core/iomgr/udp_server.c +++ b/src/core/iomgr/udp_server.c @@ -111,15 +111,12 @@ struct grpc_udp_server { size_t port_capacity; /* shutdown callback */ - void (*shutdown_complete)(void *); - void *shutdown_complete_arg; + grpc_closure *shutdown_complete; /* all pollsets interested in new connections */ grpc_pollset **pollsets; /* number of pollsets in the pollsets array */ size_t pollset_count; - - grpc_workqueue *workqueue; }; grpc_udp_server *grpc_udp_server_create(void) { @@ -132,40 +129,38 @@ grpc_udp_server *grpc_udp_server_create(void) { s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP); s->nports = 0; s->port_capacity = INIT_PORT_CAP; - s->workqueue = grpc_workqueue_create(); return s; } -static void finish_shutdown(grpc_udp_server *s) { - s->shutdown_complete(s->shutdown_complete_arg); +static void finish_shutdown(grpc_udp_server *s, grpc_call_list *call_list) { + grpc_call_list_add(call_list, s->shutdown_complete, 1); gpr_mu_destroy(&s->mu); gpr_cv_destroy(&s->cv); gpr_free(s->ports); - GRPC_WORKQUEUE_UNREF(s->workqueue, "workqueue"); gpr_free(s); } -static void destroyed_port(void *server, int success) { +static void destroyed_port(void *server, int success, + grpc_call_list *call_list) { grpc_udp_server *s = server; gpr_mu_lock(&s->mu); s->destroyed_ports++; if (s->destroyed_ports == s->nports) { gpr_mu_unlock(&s->mu); - finish_shutdown(s); + finish_shutdown(s, call_list); } else { gpr_mu_unlock(&s->mu); } } -static void dont_care_about_shutdown_completion(void *ignored) {} - /* called when all listening endpoints have been shutdown, so no further events will be received on them - at this point it's safe to destroy things */ -static void deactivated_all_ports(grpc_udp_server *s) { +static void deactivated_all_ports(grpc_udp_server *s, + grpc_call_list *call_list) { size_t i; /* delete ALL the things */ @@ -184,38 +179,35 @@ static void deactivated_all_ports(grpc_udp_server *s) { } sp->destroyed_closure.cb = destroyed_port; sp->destroyed_closure.cb_arg = s; - grpc_fd_orphan(sp->emfd, &sp->destroyed_closure, "udp_listener_shutdown"); + grpc_fd_orphan(sp->emfd, &sp->destroyed_closure, "udp_listener_shutdown", + call_list); } gpr_mu_unlock(&s->mu); } else { gpr_mu_unlock(&s->mu); - finish_shutdown(s); + finish_shutdown(s, call_list); } } -void grpc_udp_server_destroy( - grpc_udp_server *s, void (*shutdown_complete)(void *shutdown_complete_arg), - void *shutdown_complete_arg) { +void grpc_udp_server_destroy(grpc_udp_server *s, grpc_closure *on_done, + grpc_call_list *call_list) { 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; + s->shutdown_complete = on_done; /* shutdown all fd's */ if (s->active_ports) { for (i = 0; i < s->nports; i++) { - grpc_fd_shutdown(s->ports[i].emfd); + grpc_fd_shutdown(s->ports[i].emfd, call_list); } gpr_mu_unlock(&s->mu); } else { gpr_mu_unlock(&s->mu); - deactivated_all_ports(s); + deactivated_all_ports(s, call_list); } } @@ -270,14 +262,14 @@ error: } /* event manager callback when reads are ready */ -static void on_read(void *arg, int success) { +static void on_read(void *arg, int success, grpc_call_list *call_list) { server_port *sp = arg; if (success == 0) { gpr_mu_lock(&sp->server->mu); if (0 == --sp->server->active_ports) { gpr_mu_unlock(&sp->server->mu); - deactivated_all_ports(sp->server); + deactivated_all_ports(sp->server, call_list); } else { gpr_mu_unlock(&sp->server->mu); } @@ -289,7 +281,7 @@ static void on_read(void *arg, int success) { sp->read_cb(sp->fd); /* Re-arm the notification event so we get another chance to read. */ - grpc_fd_notify_on_read(sp->emfd, &sp->read_closure); + grpc_fd_notify_on_read(sp->emfd, &sp->read_closure, call_list); } static int add_socket_to_server(grpc_udp_server *s, int fd, @@ -313,7 +305,7 @@ static int add_socket_to_server(grpc_udp_server *s, int fd, sp = &s->ports[s->nports++]; sp->server = s; sp->fd = fd; - sp->emfd = grpc_fd_create(fd, s->workqueue, name); + sp->emfd = grpc_fd_create(fd, name); memcpy(sp->addr.untyped, addr, addr_len); sp->addr_len = addr_len; sp->read_cb = read_cb; @@ -410,18 +402,19 @@ int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned index) { } void grpc_udp_server_start(grpc_udp_server *s, grpc_pollset **pollsets, - size_t pollset_count) { + size_t pollset_count, grpc_call_list *call_list) { size_t i, j; gpr_mu_lock(&s->mu); GPR_ASSERT(s->active_ports == 0); s->pollsets = pollsets; for (i = 0; i < s->nports; i++) { for (j = 0; j < pollset_count; j++) { - grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd); + grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd, call_list); } s->ports[i].read_closure.cb = on_read; s->ports[i].read_closure.cb_arg = &s->ports[i]; - grpc_fd_notify_on_read(s->ports[i].emfd, &s->ports[i].read_closure); + grpc_fd_notify_on_read(s->ports[i].emfd, &s->ports[i].read_closure, + call_list); s->active_ports++; } gpr_mu_unlock(&s->mu); diff --git a/src/core/iomgr/udp_server.h b/src/core/iomgr/udp_server.h index c930e81cbc..fa4d2147b4 100644 --- a/src/core/iomgr/udp_server.h +++ b/src/core/iomgr/udp_server.h @@ -47,7 +47,7 @@ grpc_udp_server *grpc_udp_server_create(void); /* Start listening to bound ports */ void grpc_udp_server_start(grpc_udp_server *udp_server, grpc_pollset **pollsets, - size_t pollset_count); + size_t pollset_count, grpc_call_list *call_list); int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned index); @@ -64,9 +64,8 @@ int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned index); int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr, size_t addr_len, grpc_udp_server_read_cb read_cb); -void grpc_udp_server_destroy(grpc_udp_server *server, - void (*shutdown_done)(void *shutdown_done_arg), - void *shutdown_done_arg); +void grpc_udp_server_destroy(grpc_udp_server *server, grpc_closure *on_done, + grpc_call_list *call_list); /* Write the contents of buffer to the underlying UDP socket. */ /* diff --git a/src/core/iomgr/workqueue.h b/src/core/iomgr/workqueue.h index 6f09399b55..b9d2a87dca 100644 --- a/src/core/iomgr/workqueue.h +++ b/src/core/iomgr/workqueue.h @@ -50,25 +50,25 @@ struct grpc_workqueue; typedef struct grpc_workqueue grpc_workqueue; /** Create a work queue */ -grpc_workqueue *grpc_workqueue_create(void); +grpc_workqueue *grpc_workqueue_create(grpc_call_list *call_list); -void grpc_workqueue_flush(grpc_workqueue *workqueue); +void grpc_workqueue_flush(grpc_workqueue *workqueue, grpc_call_list *call_list); #define GRPC_WORKQUEUE_REFCOUNT_DEBUG #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG #define GRPC_WORKQUEUE_REF(p, r) \ grpc_workqueue_ref((p), __FILE__, __LINE__, (r)) -#define GRPC_WORKQUEUE_UNREF(p, r) \ - grpc_workqueue_unref((p), __FILE__, __LINE__, (r)) +#define GRPC_WORKQUEUE_UNREF(p, r, cl) \ + grpc_workqueue_unref((p), (cl), __FILE__, __LINE__, (r)) void grpc_workqueue_ref(grpc_workqueue *workqueue, const char *file, int line, const char *reason); -void grpc_workqueue_unref(grpc_workqueue *workqueue, const char *file, int line, - const char *reason); +void grpc_workqueue_unref(grpc_workqueue *workqueue, grpc_call_list *call_list, + const char *file, int line, const char *reason); #else #define GRPC_WORKQUEUE_REF(p, r) grpc_workqueue_ref((p)) -#define GRPC_WORKQUEUE_UNREF(p, r) grpc_workqueue_unref((p)) +#define GRPC_WORKQUEUE_UNREF(p, r, cl) grpc_workqueue_unref((p), (cl)) void grpc_workqueue_ref(grpc_workqueue *workqueue); -void grpc_workqueue_unref(grpc_workqueue *workqueue); +void grpc_workqueue_unref(grpc_workqueue *workqueue, grpc_call_list *call_list); #endif /** Bind this workqueue to a pollset */ diff --git a/src/core/iomgr/workqueue_posix.c b/src/core/iomgr/workqueue_posix.c index 85b541f4d2..83249c583c 100644 --- a/src/core/iomgr/workqueue_posix.c +++ b/src/core/iomgr/workqueue_posix.c @@ -45,9 +45,9 @@ #include "src/core/iomgr/fd_posix.h" -static void on_readable(void *arg, int success); +static void on_readable(void *arg, int success, grpc_call_list *call_list); -grpc_workqueue *grpc_workqueue_create(void) { +grpc_workqueue *grpc_workqueue_create(grpc_call_list *call_list) { char name[32]; grpc_workqueue *workqueue = gpr_malloc(sizeof(grpc_workqueue)); gpr_ref_init(&workqueue->refs, 1); @@ -55,17 +55,18 @@ grpc_workqueue *grpc_workqueue_create(void) { workqueue->call_list.head = workqueue->call_list.tail = NULL; grpc_wakeup_fd_init(&workqueue->wakeup_fd); sprintf(name, "workqueue:%p", (void *)workqueue); - workqueue->wakeup_read_fd = NULL; /* inspected during grpc_fd_create below */ - workqueue->wakeup_read_fd = grpc_fd_create( - GRPC_WAKEUP_FD_GET_READ_FD(&workqueue->wakeup_fd), workqueue, name); + workqueue->wakeup_read_fd = + grpc_fd_create(GRPC_WAKEUP_FD_GET_READ_FD(&workqueue->wakeup_fd), name); grpc_closure_init(&workqueue->read_closure, on_readable, workqueue); - grpc_fd_notify_on_read(workqueue->wakeup_read_fd, &workqueue->read_closure); + grpc_fd_notify_on_read(workqueue->wakeup_read_fd, &workqueue->read_closure, + call_list); return workqueue; } -static void workqueue_destroy(grpc_workqueue *workqueue) { +static void workqueue_destroy(grpc_workqueue *workqueue, + grpc_call_list *call_list) { GPR_ASSERT(grpc_call_list_empty(workqueue->call_list)); - grpc_fd_shutdown(workqueue->wakeup_read_fd); + grpc_fd_shutdown(workqueue->wakeup_read_fd, call_list); } #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG @@ -81,33 +82,34 @@ void grpc_workqueue_ref(grpc_workqueue *workqueue) { } #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG -void grpc_workqueue_unref(grpc_workqueue *workqueue, const char *file, int line, - const char *reason) { +void grpc_workqueue_unref(grpc_workqueue *workqueue, grpc_call_list *call_list, + const char *file, int line, const char *reason) { gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "WORKQUEUE:%p unref %d -> %d %s", workqueue, (int)workqueue->refs.count, (int)workqueue->refs.count - 1, reason); #else -void grpc_workqueue_unref(grpc_workqueue *workqueue) { +void grpc_workqueue_unref(grpc_workqueue *workqueue, + grpc_call_list *call_list) { #endif if (gpr_unref(&workqueue->refs)) { - workqueue_destroy(workqueue); + workqueue_destroy(workqueue, call_list); } } void grpc_workqueue_add_to_pollset(grpc_workqueue *workqueue, - grpc_pollset *pollset) { - grpc_pollset_add_fd(pollset, workqueue->wakeup_read_fd); + grpc_pollset *pollset, + grpc_call_list *call_list) { + grpc_pollset_add_fd(pollset, workqueue->wakeup_read_fd, call_list); } -void grpc_workqueue_flush(grpc_workqueue *workqueue) { - grpc_call_list todo = GRPC_CALL_LIST_INIT; +void grpc_workqueue_flush(grpc_workqueue *workqueue, + grpc_call_list *call_list) { gpr_mu_lock(&workqueue->mu); - GPR_SWAP(grpc_call_list, todo, workqueue->call_list); + grpc_call_list_move(&workqueue->call_list, call_list); gpr_mu_unlock(&workqueue->mu); - grpc_call_list_run(todo); } -static void on_readable(void *arg, int success) { +static void on_readable(void *arg, int success, grpc_call_list *call_list) { grpc_workqueue *workqueue = arg; if (!success) { @@ -115,16 +117,15 @@ static void on_readable(void *arg, int success) { /* HACK: let wakeup_fd code know that we stole the fd */ workqueue->wakeup_fd.read_fd = 0; grpc_wakeup_fd_destroy(&workqueue->wakeup_fd); - grpc_fd_orphan(workqueue->wakeup_read_fd, NULL, "destroy"); + grpc_fd_orphan(workqueue->wakeup_read_fd, NULL, "destroy", call_list); gpr_free(workqueue); } else { - grpc_call_list todo = GRPC_CALL_LIST_INIT; gpr_mu_lock(&workqueue->mu); - GPR_SWAP(grpc_call_list, todo, workqueue->call_list); + grpc_call_list_move(&workqueue->call_list, call_list); grpc_wakeup_fd_consume_wakeup(&workqueue->wakeup_fd); gpr_mu_unlock(&workqueue->mu); - grpc_fd_notify_on_read(workqueue->wakeup_read_fd, &workqueue->read_closure); - grpc_call_list_run(todo); + grpc_fd_notify_on_read(workqueue->wakeup_read_fd, &workqueue->read_closure, + call_list); } } diff --git a/src/core/security/client_auth_filter.c b/src/core/security/client_auth_filter.c index f3ecfd0e60..0a2de4f7bb 100644 --- a/src/core/security/client_auth_filter.c +++ b/src/core/security/client_auth_filter.c @@ -76,17 +76,18 @@ typedef struct { } channel_data; static void bubble_up_error(grpc_call_element *elem, grpc_status_code status, - const char *error_msg) { + const char *error_msg, grpc_call_list *call_list) { call_data *calld = elem->call_data; gpr_log(GPR_ERROR, "Client side authentication failure: %s", error_msg); grpc_transport_stream_op_add_cancellation(&calld->op, status); - grpc_call_next_op(elem, &calld->op); + grpc_call_next_op(elem, &calld->op, call_list); } static void on_credentials_metadata(void *user_data, grpc_credentials_md *md_elems, size_t num_md, - grpc_credentials_status status) { + grpc_credentials_status status, + grpc_call_list *call_list) { grpc_call_element *elem = (grpc_call_element *)user_data; call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; @@ -95,7 +96,7 @@ static void on_credentials_metadata(void *user_data, size_t i; if (status != GRPC_CREDENTIALS_OK) { bubble_up_error(elem, GRPC_STATUS_UNAUTHENTICATED, - "Credentials failed to get metadata."); + "Credentials failed to get metadata.", call_list); return; } GPR_ASSERT(num_md <= MAX_CREDENTIALS_METADATA_COUNT); @@ -108,7 +109,7 @@ static void on_credentials_metadata(void *user_data, grpc_mdelem_from_slices(chand->md_ctx, gpr_slice_ref(md_elems[i].key), gpr_slice_ref(md_elems[i].value))); } - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); } static char *build_service_url(const char *url_scheme, call_data *calld) { @@ -132,7 +133,8 @@ static char *build_service_url(const char *url_scheme, call_data *calld) { } static void send_security_metadata(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; grpc_client_security_context *ctx = @@ -148,7 +150,7 @@ static void send_security_metadata(grpc_call_element *elem, if (!channel_creds_has_md && !call_creds_has_md) { /* Skip sending metadata altogether. */ - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); return; } @@ -157,7 +159,8 @@ static void send_security_metadata(grpc_call_element *elem, grpc_composite_credentials_create(channel_creds, ctx->creds, NULL); if (calld->creds == NULL) { bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT, - "Incompatible credentials set on channel and call."); + "Incompatible credentials set on channel and call.", + call_list); return; } } else { @@ -169,22 +172,24 @@ static void send_security_metadata(grpc_call_element *elem, build_service_url(chand->security_connector->base.url_scheme, calld); calld->op = *op; /* Copy op (originates from the caller's stack). */ GPR_ASSERT(calld->pollset); - grpc_credentials_get_request_metadata( - calld->creds, calld->pollset, service_url, on_credentials_metadata, elem); + grpc_credentials_get_request_metadata(calld->creds, calld->pollset, + service_url, on_credentials_metadata, + elem, call_list); gpr_free(service_url); } -static void on_host_checked(void *user_data, grpc_security_status status) { +static void on_host_checked(void *user_data, grpc_security_status status, + grpc_call_list *call_list) { grpc_call_element *elem = (grpc_call_element *)user_data; call_data *calld = elem->call_data; if (status == GRPC_SECURITY_OK) { - send_security_metadata(elem, &calld->op); + send_security_metadata(elem, &calld->op, call_list); } else { char *error_msg; gpr_asprintf(&error_msg, "Invalid host %s set in :authority metadata.", grpc_mdstr_as_c_string(calld->host)); - bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT, error_msg); + bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT, error_msg, call_list); gpr_free(error_msg); } } @@ -195,7 +200,8 @@ static void on_host_checked(void *user_data, grpc_security_status status) { op contains type and call direction information, in addition to the data that is being sent or received. */ static void auth_start_transport_op(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; @@ -247,26 +253,28 @@ static void auth_start_transport_op(grpc_call_element *elem, const char *call_host = grpc_mdstr_as_c_string(calld->host); calld->op = *op; /* Copy op (originates from the caller's stack). */ status = grpc_channel_security_connector_check_call_host( - chand->security_connector, call_host, on_host_checked, elem); + chand->security_connector, call_host, on_host_checked, elem, + call_list); if (status != GRPC_SECURITY_OK) { if (status == GRPC_SECURITY_ERROR) { char *error_msg; gpr_asprintf(&error_msg, "Invalid host %s set in :authority metadata.", call_host); - bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT, error_msg); + bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT, error_msg, + call_list); gpr_free(error_msg); } return; /* early exit */ } } - send_security_metadata(elem, op); + send_security_metadata(elem, op, call_list); return; /* early exit */ } } - /* pass control up or down the stack */ - grpc_call_next_op(elem, op); + /* pass control down the stack */ + grpc_call_next_op(elem, op, call_list); } /* Constructor for call_data */ @@ -285,7 +293,8 @@ static void init_call_elem(grpc_call_element *elem, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_call_element *elem) { +static void destroy_call_elem(grpc_call_element *elem, + grpc_call_list *call_list) { call_data *calld = elem->call_data; grpc_credentials_unref(calld->creds); if (calld->host != NULL) { @@ -300,7 +309,7 @@ static void destroy_call_elem(grpc_call_element *elem) { static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *metadata_context, int is_first, - int is_last) { + int is_last, grpc_call_list *call_list) { grpc_security_connector *sc = grpc_find_security_connector_in_args(args); /* grab pointers to our data from the channel element */ channel_data *chand = elem->channel_data; @@ -326,7 +335,8 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, } /* Destructor for channel data */ -static void destroy_channel_elem(grpc_channel_element *elem) { +static void destroy_channel_elem(grpc_channel_element *elem, + grpc_call_list *call_list) { /* grab pointers to our data from the channel element */ channel_data *chand = elem->channel_data; grpc_channel_security_connector *ctx = chand->security_connector; diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c index f151fba959..086c428112 100644 --- a/src/core/security/credentials.c +++ b/src/core/security/credentials.c @@ -108,16 +108,17 @@ void grpc_credentials_get_request_metadata(grpc_credentials *creds, grpc_pollset *pollset, const char *service_url, grpc_credentials_metadata_cb cb, - void *user_data) { + void *user_data, + grpc_call_list *call_list) { if (creds == NULL || !grpc_credentials_has_request_metadata(creds) || creds->vtable->get_request_metadata == NULL) { if (cb != NULL) { - cb(user_data, NULL, 0, GRPC_CREDENTIALS_OK); + cb(user_data, NULL, 0, GRPC_CREDENTIALS_OK, call_list); } return; } creds->vtable->get_request_metadata(creds, pollset, service_url, cb, - user_data); + user_data, call_list); } grpc_security_status grpc_credentials_create_security_connector( @@ -375,7 +376,8 @@ static void jwt_get_request_metadata(grpc_credentials *creds, grpc_pollset *pollset, const char *service_url, grpc_credentials_metadata_cb cb, - void *user_data) { + void *user_data, + grpc_call_list *call_list) { grpc_service_account_jwt_access_credentials *c = (grpc_service_account_jwt_access_credentials *)creds; gpr_timespec refresh_threshold = gpr_time_from_seconds( @@ -419,10 +421,11 @@ static void jwt_get_request_metadata(grpc_credentials *creds, } if (jwt_md != NULL) { - cb(user_data, jwt_md->entries, jwt_md->num_entries, GRPC_CREDENTIALS_OK); + cb(user_data, jwt_md->entries, jwt_md->num_entries, GRPC_CREDENTIALS_OK, + call_list); grpc_credentials_md_store_unref(jwt_md); } else { - cb(user_data, NULL, 0, GRPC_CREDENTIALS_ERROR); + cb(user_data, NULL, 0, GRPC_CREDENTIALS_ERROR, call_list); } } @@ -568,7 +571,8 @@ end: } static void on_oauth2_token_fetcher_http_response( - void *user_data, const grpc_httpcli_response *response) { + void *user_data, const grpc_httpcli_response *response, + grpc_call_list *call_list) { grpc_credentials_metadata_request *r = (grpc_credentials_metadata_request *)user_data; grpc_oauth2_token_fetcher_credentials *c = @@ -583,10 +587,10 @@ static void on_oauth2_token_fetcher_http_response( c->token_expiration = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), token_lifetime); r->cb(r->user_data, c->access_token_md->entries, - c->access_token_md->num_entries, status); + c->access_token_md->num_entries, status, call_list); } else { c->token_expiration = gpr_inf_past(GPR_CLOCK_REALTIME); - r->cb(r->user_data, NULL, 0, status); + r->cb(r->user_data, NULL, 0, status, call_list); } gpr_mu_unlock(&c->mu); grpc_credentials_metadata_request_destroy(r); @@ -594,7 +598,8 @@ static void on_oauth2_token_fetcher_http_response( static void oauth2_token_fetcher_get_request_metadata( grpc_credentials *creds, grpc_pollset *pollset, const char *service_url, - grpc_credentials_metadata_cb cb, void *user_data) { + grpc_credentials_metadata_cb cb, void *user_data, + grpc_call_list *call_list) { grpc_oauth2_token_fetcher_credentials *c = (grpc_oauth2_token_fetcher_credentials *)creds; gpr_timespec refresh_threshold = gpr_time_from_seconds( @@ -613,13 +618,14 @@ static void oauth2_token_fetcher_get_request_metadata( } if (cached_access_token_md != NULL) { cb(user_data, cached_access_token_md->entries, - cached_access_token_md->num_entries, GRPC_CREDENTIALS_OK); + cached_access_token_md->num_entries, GRPC_CREDENTIALS_OK, call_list); grpc_credentials_md_store_unref(cached_access_token_md); } else { c->fetch_func( grpc_credentials_metadata_request_create(creds, cb, user_data), &c->httpcli_context, pollset, on_oauth2_token_fetcher_http_response, - gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), refresh_threshold)); + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), refresh_threshold), + call_list); } } @@ -644,7 +650,8 @@ static grpc_credentials_vtable compute_engine_vtable = { static void compute_engine_fetch_oauth2( grpc_credentials_metadata_request *metadata_req, grpc_httpcli_context *httpcli_context, grpc_pollset *pollset, - grpc_httpcli_response_cb response_cb, gpr_timespec deadline) { + grpc_httpcli_response_cb response_cb, gpr_timespec deadline, + grpc_call_list *call_list) { grpc_httpcli_header header = {"Metadata-Flavor", "Google"}; grpc_httpcli_request request; memset(&request, 0, sizeof(grpc_httpcli_request)); @@ -653,7 +660,7 @@ static void compute_engine_fetch_oauth2( request.hdr_count = 1; request.hdrs = &header; grpc_httpcli_get(httpcli_context, pollset, &request, deadline, response_cb, - metadata_req); + metadata_req, call_list); } grpc_credentials *grpc_google_compute_engine_credentials_create( @@ -683,7 +690,8 @@ static grpc_credentials_vtable refresh_token_vtable = { static void refresh_token_fetch_oauth2( grpc_credentials_metadata_request *metadata_req, grpc_httpcli_context *httpcli_context, grpc_pollset *pollset, - grpc_httpcli_response_cb response_cb, gpr_timespec deadline) { + grpc_httpcli_response_cb response_cb, gpr_timespec deadline, + grpc_call_list *call_list) { grpc_google_refresh_token_credentials *c = (grpc_google_refresh_token_credentials *)metadata_req->creds; grpc_httpcli_header header = {"Content-Type", @@ -700,7 +708,7 @@ static void refresh_token_fetch_oauth2( request.hdrs = &header; request.handshaker = &grpc_httpcli_ssl; grpc_httpcli_post(httpcli_context, pollset, &request, body, strlen(body), - deadline, response_cb, metadata_req); + deadline, response_cb, metadata_req, call_list); gpr_free(body); } @@ -743,20 +751,23 @@ static int md_only_test_has_request_metadata_only( return 1; } -void on_simulated_token_fetch_done(void *user_data) { +static void on_simulated_token_fetch_done(void *user_data) { grpc_credentials_metadata_request *r = (grpc_credentials_metadata_request *)user_data; grpc_md_only_test_credentials *c = (grpc_md_only_test_credentials *)r->creds; + grpc_call_list call_list = GRPC_CALL_LIST_INIT; r->cb(r->user_data, c->md_store->entries, c->md_store->num_entries, - GRPC_CREDENTIALS_OK); + GRPC_CREDENTIALS_OK, &call_list); grpc_credentials_metadata_request_destroy(r); + grpc_call_list_run(&call_list); } static void md_only_test_get_request_metadata(grpc_credentials *creds, grpc_pollset *pollset, const char *service_url, grpc_credentials_metadata_cb cb, - void *user_data) { + void *user_data, + grpc_call_list *call_list) { grpc_md_only_test_credentials *c = (grpc_md_only_test_credentials *)creds; if (c->is_async) { @@ -765,7 +776,7 @@ static void md_only_test_get_request_metadata(grpc_credentials *creds, grpc_credentials_metadata_request_create(creds, cb, user_data); gpr_thd_new(&thd_id, on_simulated_token_fetch_done, cb_arg, NULL); } else { - cb(user_data, c->md_store->entries, 1, GRPC_CREDENTIALS_OK); + cb(user_data, c->md_store->entries, 1, GRPC_CREDENTIALS_OK, call_list); } } @@ -809,9 +820,10 @@ static void access_token_get_request_metadata(grpc_credentials *creds, grpc_pollset *pollset, const char *service_url, grpc_credentials_metadata_cb cb, - void *user_data) { + void *user_data, + grpc_call_list *call_list) { grpc_access_token_credentials *c = (grpc_access_token_credentials *)creds; - cb(user_data, c->access_token_md->entries, 1, GRPC_CREDENTIALS_OK); + cb(user_data, c->access_token_md->entries, 1, GRPC_CREDENTIALS_OK, call_list); } static grpc_credentials_vtable access_token_vtable = { @@ -958,11 +970,12 @@ static void composite_md_context_destroy( static void composite_metadata_cb(void *user_data, grpc_credentials_md *md_elems, size_t num_md, - grpc_credentials_status status) { + grpc_credentials_status status, + grpc_call_list *call_list) { grpc_composite_credentials_metadata_context *ctx = (grpc_composite_credentials_metadata_context *)user_data; if (status != GRPC_CREDENTIALS_OK) { - ctx->cb(ctx->user_data, NULL, 0, status); + ctx->cb(ctx->user_data, NULL, 0, status, call_list); return; } @@ -980,28 +993,30 @@ static void composite_metadata_cb(void *user_data, grpc_credentials *inner_creds = ctx->composite_creds->inner.creds_array[ctx->creds_index++]; if (grpc_credentials_has_request_metadata(inner_creds)) { - grpc_credentials_get_request_metadata(inner_creds, ctx->pollset, - ctx->service_url, - composite_metadata_cb, ctx); + grpc_credentials_get_request_metadata( + inner_creds, ctx->pollset, ctx->service_url, composite_metadata_cb, + ctx, call_list); return; } } /* We're done!. */ ctx->cb(ctx->user_data, ctx->md_elems->entries, ctx->md_elems->num_entries, - GRPC_CREDENTIALS_OK); + GRPC_CREDENTIALS_OK, call_list); composite_md_context_destroy(ctx); + grpc_call_list_run(call_list); } static void composite_get_request_metadata(grpc_credentials *creds, grpc_pollset *pollset, const char *service_url, grpc_credentials_metadata_cb cb, - void *user_data) { + void *user_data, + grpc_call_list *call_list) { grpc_composite_credentials *c = (grpc_composite_credentials *)creds; grpc_composite_credentials_metadata_context *ctx; if (!grpc_credentials_has_request_metadata(creds)) { - cb(user_data, NULL, 0, GRPC_CREDENTIALS_OK); + cb(user_data, NULL, 0, GRPC_CREDENTIALS_OK, call_list); return; } ctx = gpr_malloc(sizeof(grpc_composite_credentials_metadata_context)); @@ -1016,7 +1031,8 @@ static void composite_get_request_metadata(grpc_credentials *creds, grpc_credentials *inner_creds = c->inner.creds_array[ctx->creds_index++]; if (grpc_credentials_has_request_metadata(inner_creds)) { grpc_credentials_get_request_metadata(inner_creds, pollset, service_url, - composite_metadata_cb, ctx); + composite_metadata_cb, ctx, + call_list); return; } } @@ -1152,10 +1168,11 @@ static void iam_get_request_metadata(grpc_credentials *creds, grpc_pollset *pollset, const char *service_url, grpc_credentials_metadata_cb cb, - void *user_data) { + void *user_data, + grpc_call_list *call_list) { grpc_google_iam_credentials *c = (grpc_google_iam_credentials *)creds; - cb(user_data, c->iam_md->entries, c->iam_md->num_entries, - GRPC_CREDENTIALS_OK); + cb(user_data, c->iam_md->entries, c->iam_md->num_entries, GRPC_CREDENTIALS_OK, + call_list); } static grpc_credentials_vtable iam_vtable = { diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h index 8e4fed7615..a93d0ec653 100644 --- a/src/core/security/credentials.h +++ b/src/core/security/credentials.h @@ -126,7 +126,8 @@ char *grpc_get_well_known_google_credentials_file_path(void); typedef void (*grpc_credentials_metadata_cb)(void *user_data, grpc_credentials_md *md_elems, size_t num_md, - grpc_credentials_status status); + grpc_credentials_status status, + grpc_call_list *call_list); typedef struct { void (*destruct)(grpc_credentials *c); @@ -134,8 +135,8 @@ typedef struct { int (*has_request_metadata_only)(const grpc_credentials *c); void (*get_request_metadata)(grpc_credentials *c, grpc_pollset *pollset, const char *service_url, - grpc_credentials_metadata_cb cb, - void *user_data); + grpc_credentials_metadata_cb cb, void *user_data, + grpc_call_list *call_list); grpc_security_status (*create_security_connector)( grpc_credentials *c, const char *target, const grpc_channel_args *args, grpc_credentials *request_metadata_creds, @@ -156,7 +157,8 @@ void grpc_credentials_get_request_metadata(grpc_credentials *creds, grpc_pollset *pollset, const char *service_url, grpc_credentials_metadata_cb cb, - void *user_data); + void *user_data, + grpc_call_list *call_list); /* Creates a security connector for the channel. May also create new channel args for the channel to be used in place of the passed in const args if @@ -274,7 +276,8 @@ typedef void (*grpc_fetch_oauth2_func)(grpc_credentials_metadata_request *req, grpc_httpcli_context *http_context, grpc_pollset *pollset, grpc_httpcli_response_cb response_cb, - gpr_timespec deadline); + gpr_timespec deadline, + grpc_call_list *call_list); typedef struct { grpc_credentials base; diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c index 874dd59e84..5da29f5ad5 100644 --- a/src/core/security/google_default_credentials.c +++ b/src/core/security/google_default_credentials.c @@ -63,7 +63,8 @@ typedef struct { } compute_engine_detector; static void on_compute_engine_detection_http_response( - void *user_data, const grpc_httpcli_response *response) { + void *user_data, const grpc_httpcli_response *response, + grpc_call_list *call_list) { compute_engine_detector *detector = (compute_engine_detector *)user_data; if (response != NULL && response->status == 200 && response->hdr_count > 0) { /* Internet providers can return a generic response to all requests, so @@ -84,12 +85,16 @@ static void on_compute_engine_detection_http_response( gpr_mu_unlock(GRPC_POLLSET_MU(&detector->pollset)); } -static void destroy_pollset(void *p) { grpc_pollset_destroy(p); } +static void destroy_pollset(void *p, int s, grpc_call_list *call_list) { + grpc_pollset_destroy(p); +} static int is_stack_running_on_compute_engine(void) { compute_engine_detector detector; grpc_httpcli_request request; grpc_httpcli_context context; + grpc_call_list call_list = GRPC_CALL_LIST_INIT; + grpc_closure destroy_closure; /* The http call is local. If it takes more than one sec, it is for sure not on compute engine. */ @@ -108,7 +113,9 @@ static int is_stack_running_on_compute_engine(void) { grpc_httpcli_get( &context, &detector.pollset, &request, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), max_detection_delay), - on_compute_engine_detection_http_response, &detector); + on_compute_engine_detection_http_response, &detector, &call_list); + + grpc_call_list_run(&call_list); /* 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. */ @@ -121,7 +128,9 @@ static int is_stack_running_on_compute_engine(void) { gpr_mu_unlock(GRPC_POLLSET_MU(&detector.pollset)); grpc_httpcli_context_destroy(&context); - grpc_pollset_shutdown(&detector.pollset, destroy_pollset, &detector.pollset); + grpc_closure_init(&destroy_closure, destroy_pollset, &detector.pollset); + grpc_pollset_shutdown(&detector.pollset, &destroy_closure, &call_list); + grpc_call_list_run(&call_list); return detector.success; } diff --git a/src/core/security/jwt_verifier.c b/src/core/security/jwt_verifier.c index 790f2178db..a6736c922c 100644 --- a/src/core/security/jwt_verifier.c +++ b/src/core/security/jwt_verifier.c @@ -570,7 +570,8 @@ end: } static void on_keys_retrieved(void *user_data, - const grpc_httpcli_response *response) { + const grpc_httpcli_response *response, + grpc_call_list *call_list) { grpc_json *json = json_from_http(response); verifier_cb_ctx *ctx = (verifier_cb_ctx *)user_data; EVP_PKEY *verification_key = NULL; @@ -611,7 +612,8 @@ end: } static void on_openid_config_retrieved(void *user_data, - const grpc_httpcli_response *response) { + const grpc_httpcli_response *response, + grpc_call_list *call_list) { const grpc_json *cur; grpc_json *json = json_from_http(response); verifier_cb_ctx *ctx = (verifier_cb_ctx *)user_data; @@ -643,7 +645,7 @@ static void on_openid_config_retrieved(void *user_data, grpc_httpcli_get( &ctx->verifier->http_ctx, ctx->pollset, &req, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), grpc_jwt_verifier_max_delay), - on_keys_retrieved, ctx); + on_keys_retrieved, ctx, call_list); grpc_json_destroy(json); gpr_free(req.host); return; @@ -682,7 +684,8 @@ static void verifier_put_mapping(grpc_jwt_verifier *v, const char *email_domain, } /* Takes ownership of ctx. */ -static void retrieve_key_and_verify(verifier_cb_ctx *ctx) { +static void retrieve_key_and_verify(verifier_cb_ctx *ctx, + grpc_call_list *call_list) { const char *at_sign; grpc_httpcli_response_cb http_cb; char *path_prefix = NULL; @@ -745,7 +748,7 @@ static void retrieve_key_and_verify(verifier_cb_ctx *ctx) { grpc_httpcli_get( &ctx->verifier->http_ctx, ctx->pollset, &req, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), grpc_jwt_verifier_max_delay), - http_cb, ctx); + http_cb, ctx, call_list); gpr_free(req.host); gpr_free(req.path); return; @@ -758,8 +761,8 @@ error: void grpc_jwt_verifier_verify(grpc_jwt_verifier *verifier, grpc_pollset *pollset, const char *jwt, const char *audience, - grpc_jwt_verification_done_cb cb, - void *user_data) { + grpc_jwt_verification_done_cb cb, void *user_data, + grpc_call_list *call_list) { const char *dot = NULL; grpc_json *json; jose_header *header = NULL; @@ -792,7 +795,8 @@ void grpc_jwt_verifier_verify(grpc_jwt_verifier *verifier, if (GPR_SLICE_IS_EMPTY(signature)) goto error; retrieve_key_and_verify( verifier_cb_ctx_create(verifier, pollset, header, claims, audience, - signature, jwt, signed_jwt_len, user_data, cb)); + signature, jwt, signed_jwt_len, user_data, cb), + call_list); return; error: diff --git a/src/core/security/jwt_verifier.h b/src/core/security/jwt_verifier.h index 7a32debfcb..394935302b 100644 --- a/src/core/security/jwt_verifier.h +++ b/src/core/security/jwt_verifier.h @@ -123,8 +123,8 @@ typedef void (*grpc_jwt_verification_done_cb)(void *user_data, void grpc_jwt_verifier_verify(grpc_jwt_verifier *verifier, grpc_pollset *pollset, const char *jwt, const char *audience, - grpc_jwt_verification_done_cb cb, - void *user_data); + grpc_jwt_verification_done_cb cb, void *user_data, + grpc_call_list *call_list); /* --- TESTING ONLY exposed functions. --- */ diff --git a/src/core/security/secure_endpoint.c b/src/core/security/secure_endpoint.c index dcafc0fc0f..a501544341 100644 --- a/src/core/security/secure_endpoint.c +++ b/src/core/security/secure_endpoint.c @@ -67,9 +67,9 @@ typedef struct { int grpc_trace_secure_endpoint = 0; -static void destroy(secure_endpoint *secure_ep) { +static void destroy(secure_endpoint *secure_ep, grpc_call_list *call_list) { secure_endpoint *ep = secure_ep; - grpc_endpoint_destroy(ep->wrapped_ep); + grpc_endpoint_destroy(ep->wrapped_ep, call_list); tsi_frame_protector_destroy(ep->protector); gpr_slice_buffer_destroy(&ep->leftover_bytes); gpr_slice_unref(ep->read_staging_buffer); @@ -102,11 +102,12 @@ static void secure_endpoint_ref(secure_endpoint *ep, const char *reason, gpr_ref(&ep->ref); } #else -#define SECURE_ENDPOINT_UNREF(ep, reason) secure_endpoint_unref((ep)) +#define SECURE_ENDPOINT_UNREF(ep, reason, cl) secure_endpoint_unref((ep), (cl)) #define SECURE_ENDPOINT_REF(ep, reason) secure_endpoint_ref((ep)) -static void secure_endpoint_unref(secure_endpoint *ep) { +static void secure_endpoint_unref(secure_endpoint *ep, + grpc_call_list *call_list) { if (gpr_unref(&ep->ref)) { - destroy(ep); + destroy(ep, call_list); } } @@ -121,7 +122,8 @@ static void flush_read_staging_buffer(secure_endpoint *ep, gpr_uint8 **cur, *end = GPR_SLICE_END_PTR(ep->read_staging_buffer); } -static void call_read_cb(secure_endpoint *ep, int success) { +static void call_read_cb(secure_endpoint *ep, int success, + grpc_call_list *call_list) { if (grpc_trace_secure_endpoint) { size_t i; for (i = 0; i < ep->read_buffer->count; i++) { @@ -132,11 +134,11 @@ static void call_read_cb(secure_endpoint *ep, int success) { } } ep->read_buffer = NULL; - ep->read_cb->cb(ep->read_cb->cb_arg, success); - SECURE_ENDPOINT_UNREF(ep, "read"); + grpc_call_list_add(call_list, ep->read_cb, success); + SECURE_ENDPOINT_UNREF(ep, "read", call_list); } -static int on_read(void *user_data, int success) { +static void on_read(void *user_data, int success, grpc_call_list *call_list) { unsigned i; gpr_uint8 keep_looping = 0; tsi_result result = TSI_OK; @@ -146,7 +148,8 @@ static int on_read(void *user_data, int success) { if (!success) { gpr_slice_buffer_reset_and_unref(ep->read_buffer); - return 0; + call_read_cb(ep, 0, call_list); + return; } /* TODO(yangg) check error, maybe bail out early */ @@ -202,21 +205,16 @@ static int on_read(void *user_data, int success) { if (result != TSI_OK) { gpr_slice_buffer_reset_and_unref(ep->read_buffer); - return 0; + call_read_cb(ep, 0, call_list); + return; } - return 1; -} - -static void on_read_cb(void *user_data, int success) { - call_read_cb(user_data, on_read(user_data, success)); + call_read_cb(ep, 1, call_list); } -static grpc_endpoint_op_status endpoint_read(grpc_endpoint *secure_ep, - gpr_slice_buffer *slices, - grpc_closure *cb) { +static void endpoint_read(grpc_endpoint *secure_ep, gpr_slice_buffer *slices, + grpc_closure *cb, grpc_call_list *call_list) { secure_endpoint *ep = (secure_endpoint *)secure_ep; - int immediate_read_success = -1; ep->read_cb = cb; ep->read_buffer = slices; gpr_slice_buffer_reset_and_unref(ep->read_buffer); @@ -224,27 +222,13 @@ static grpc_endpoint_op_status endpoint_read(grpc_endpoint *secure_ep, if (ep->leftover_bytes.count) { gpr_slice_buffer_swap(&ep->leftover_bytes, &ep->source_buffer); GPR_ASSERT(ep->leftover_bytes.count == 0); - return on_read(ep, 1) ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR; + on_read(ep, 1, call_list); + return; } SECURE_ENDPOINT_REF(ep, "read"); - - switch ( - grpc_endpoint_read(ep->wrapped_ep, &ep->source_buffer, &ep->on_read)) { - case GRPC_ENDPOINT_DONE: - immediate_read_success = on_read(ep, 1); - break; - case GRPC_ENDPOINT_PENDING: - return GRPC_ENDPOINT_PENDING; - case GRPC_ENDPOINT_ERROR: - immediate_read_success = on_read(ep, 0); - break; - } - - GPR_ASSERT(immediate_read_success != -1); - SECURE_ENDPOINT_UNREF(ep, "read"); - - return immediate_read_success ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR; + grpc_endpoint_read(ep->wrapped_ep, &ep->source_buffer, &ep->on_read, + call_list); } static void flush_write_staging_buffer(secure_endpoint *ep, gpr_uint8 **cur, @@ -255,9 +239,8 @@ static void flush_write_staging_buffer(secure_endpoint *ep, gpr_uint8 **cur, *end = GPR_SLICE_END_PTR(ep->write_staging_buffer); } -static grpc_endpoint_op_status endpoint_write(grpc_endpoint *secure_ep, - gpr_slice_buffer *slices, - grpc_closure *cb) { +static void endpoint_write(grpc_endpoint *secure_ep, gpr_slice_buffer *slices, + grpc_closure *cb, grpc_call_list *call_list) { unsigned i; tsi_result result = TSI_OK; secure_endpoint *ep = (secure_endpoint *)secure_ep; @@ -329,32 +312,37 @@ static grpc_endpoint_op_status endpoint_write(grpc_endpoint *secure_ep, if (result != TSI_OK) { /* TODO(yangg) do different things according to the error type? */ gpr_slice_buffer_reset_and_unref(&ep->output_buffer); - return GRPC_ENDPOINT_ERROR; + grpc_call_list_add(call_list, cb, 0); + return; } - return grpc_endpoint_write(ep->wrapped_ep, &ep->output_buffer, cb); + grpc_endpoint_write(ep->wrapped_ep, &ep->output_buffer, cb, call_list); } -static void endpoint_shutdown(grpc_endpoint *secure_ep) { +static void endpoint_shutdown(grpc_endpoint *secure_ep, + grpc_call_list *call_list) { secure_endpoint *ep = (secure_endpoint *)secure_ep; - grpc_endpoint_shutdown(ep->wrapped_ep); + grpc_endpoint_shutdown(ep->wrapped_ep, call_list); } -static void endpoint_destroy(grpc_endpoint *secure_ep) { +static void endpoint_destroy(grpc_endpoint *secure_ep, + grpc_call_list *call_list) { secure_endpoint *ep = (secure_endpoint *)secure_ep; - SECURE_ENDPOINT_UNREF(ep, "destroy"); + SECURE_ENDPOINT_UNREF(ep, "destroy", call_list); } static void endpoint_add_to_pollset(grpc_endpoint *secure_ep, - grpc_pollset *pollset) { + grpc_pollset *pollset, + grpc_call_list *call_list) { secure_endpoint *ep = (secure_endpoint *)secure_ep; - grpc_endpoint_add_to_pollset(ep->wrapped_ep, pollset); + grpc_endpoint_add_to_pollset(ep->wrapped_ep, pollset, call_list); } static void endpoint_add_to_pollset_set(grpc_endpoint *secure_ep, - grpc_pollset_set *pollset_set) { + grpc_pollset_set *pollset_set, + grpc_call_list *call_list) { secure_endpoint *ep = (secure_endpoint *)secure_ep; - grpc_endpoint_add_to_pollset_set(ep->wrapped_ep, pollset_set); + grpc_endpoint_add_to_pollset_set(ep->wrapped_ep, pollset_set, call_list); } static char *endpoint_get_peer(grpc_endpoint *secure_ep) { @@ -386,7 +374,7 @@ grpc_endpoint *grpc_secure_endpoint_create( gpr_slice_buffer_init(&ep->output_buffer); gpr_slice_buffer_init(&ep->source_buffer); ep->read_buffer = NULL; - grpc_closure_init(&ep->on_read, on_read_cb, ep); + grpc_closure_init(&ep->on_read, on_read, ep); gpr_mu_init(&ep->protector_mu); gpr_ref_init(&ep->ref, 1); return &ep->base; diff --git a/src/core/security/secure_transport_setup.c b/src/core/security/secure_transport_setup.c index 181aa74ba9..1412ab21ad 100644 --- a/src/core/security/secure_transport_setup.c +++ b/src/core/security/secure_transport_setup.c @@ -58,23 +58,27 @@ typedef struct { grpc_closure on_handshake_data_received_from_peer; } grpc_secure_transport_setup; -static void on_handshake_data_received_from_peer(void *setup, int success); +static void on_handshake_data_received_from_peer(void *setup, int success, + grpc_call_list *call_list); -static void on_handshake_data_sent_to_peer(void *setup, int success); +static void on_handshake_data_sent_to_peer(void *setup, int success, + grpc_call_list *call_list); static void secure_transport_setup_done(grpc_secure_transport_setup *s, - int is_success) { + int is_success, + grpc_call_list *call_list) { if (is_success) { s->cb(s->user_data, GRPC_SECURITY_OK, s->wrapped_endpoint, - s->secure_endpoint); + s->secure_endpoint, call_list); } else { if (s->secure_endpoint != NULL) { - grpc_endpoint_shutdown(s->secure_endpoint); - grpc_endpoint_destroy(s->secure_endpoint); + grpc_endpoint_shutdown(s->secure_endpoint, call_list); + grpc_endpoint_destroy(s->secure_endpoint, call_list); } else { - grpc_endpoint_destroy(s->wrapped_endpoint); + grpc_endpoint_destroy(s->wrapped_endpoint, call_list); } - s->cb(s->user_data, GRPC_SECURITY_ERROR, s->wrapped_endpoint, NULL); + s->cb(s->user_data, GRPC_SECURITY_ERROR, s->wrapped_endpoint, NULL, + call_list); } if (s->handshaker != NULL) tsi_handshaker_destroy(s->handshaker); if (s->handshake_buffer != NULL) gpr_free(s->handshake_buffer); @@ -85,13 +89,14 @@ static void secure_transport_setup_done(grpc_secure_transport_setup *s, gpr_free(s); } -static void on_peer_checked(void *user_data, grpc_security_status status) { +static void on_peer_checked(void *user_data, grpc_security_status status, + grpc_call_list *call_list) { grpc_secure_transport_setup *s = user_data; tsi_frame_protector *protector; tsi_result result; if (status != GRPC_SECURITY_OK) { gpr_log(GPR_ERROR, "Error checking peer."); - secure_transport_setup_done(s, 0); + secure_transport_setup_done(s, 0, call_list); return; } result = @@ -99,7 +104,7 @@ static void on_peer_checked(void *user_data, grpc_security_status status) { if (result != TSI_OK) { gpr_log(GPR_ERROR, "Frame protector creation failed with error %s.", tsi_result_to_string(result)); - secure_transport_setup_done(s, 0); + secure_transport_setup_done(s, 0, call_list); return; } s->secure_endpoint = @@ -107,11 +112,12 @@ static void on_peer_checked(void *user_data, grpc_security_status status) { s->left_overs.slices, s->left_overs.count); s->left_overs.count = 0; s->left_overs.length = 0; - secure_transport_setup_done(s, 1); + secure_transport_setup_done(s, 1, call_list); return; } -static void check_peer(grpc_secure_transport_setup *s) { +static void check_peer(grpc_secure_transport_setup *s, + grpc_call_list *call_list) { grpc_security_status peer_status; tsi_peer peer; tsi_result result = tsi_handshaker_extract_peer(s->handshaker, &peer); @@ -119,21 +125,22 @@ static void check_peer(grpc_secure_transport_setup *s) { if (result != TSI_OK) { gpr_log(GPR_ERROR, "Peer extraction failed with error %s", tsi_result_to_string(result)); - secure_transport_setup_done(s, 0); + secure_transport_setup_done(s, 0, call_list); return; } peer_status = grpc_security_connector_check_peer(s->connector, peer, on_peer_checked, s); if (peer_status == GRPC_SECURITY_ERROR) { gpr_log(GPR_ERROR, "Peer check failed."); - secure_transport_setup_done(s, 0); + secure_transport_setup_done(s, 0, call_list); return; } else if (peer_status == GRPC_SECURITY_OK) { - on_peer_checked(s, peer_status); + on_peer_checked(s, peer_status, call_list); } } -static void send_handshake_bytes_to_peer(grpc_secure_transport_setup *s) { +static void send_handshake_bytes_to_peer(grpc_secure_transport_setup *s, + grpc_call_list *call_list) { size_t offset = 0; tsi_result result = TSI_OK; gpr_slice to_send; @@ -153,7 +160,7 @@ static void send_handshake_bytes_to_peer(grpc_secure_transport_setup *s) { if (result != TSI_OK) { gpr_log(GPR_ERROR, "Handshake failed with error %s", tsi_result_to_string(result)); - secure_transport_setup_done(s, 0); + secure_transport_setup_done(s, 0, call_list); return; } @@ -163,21 +170,12 @@ static void send_handshake_bytes_to_peer(grpc_secure_transport_setup *s) { gpr_slice_buffer_add(&s->outgoing, to_send); /* TODO(klempner,jboeuf): This should probably use the client setup deadline */ - switch (grpc_endpoint_write(s->wrapped_endpoint, &s->outgoing, - &s->on_handshake_data_sent_to_peer)) { - case GRPC_ENDPOINT_ERROR: - gpr_log(GPR_ERROR, "Could not send handshake data to peer."); - secure_transport_setup_done(s, 0); - break; - case GRPC_ENDPOINT_DONE: - on_handshake_data_sent_to_peer(s, 1); - break; - case GRPC_ENDPOINT_PENDING: - break; - } + grpc_endpoint_write(s->wrapped_endpoint, &s->outgoing, + &s->on_handshake_data_sent_to_peer, call_list); } -static void on_handshake_data_received_from_peer(void *setup, int success) { +static void on_handshake_data_received_from_peer(void *setup, int success, + grpc_call_list *call_list) { grpc_secure_transport_setup *s = setup; size_t consumed_slice_size = 0; tsi_result result = TSI_OK; @@ -187,7 +185,7 @@ static void on_handshake_data_received_from_peer(void *setup, int success) { if (!success) { gpr_log(GPR_ERROR, "Read failed."); - secure_transport_setup_done(s, 0); + secure_transport_setup_done(s, 0, call_list); return; } @@ -202,20 +200,11 @@ static void on_handshake_data_received_from_peer(void *setup, int success) { if (tsi_handshaker_is_in_progress(s->handshaker)) { /* We may need more data. */ if (result == TSI_INCOMPLETE_DATA) { - switch (grpc_endpoint_read(s->wrapped_endpoint, &s->incoming, - &s->on_handshake_data_received_from_peer)) { - case GRPC_ENDPOINT_DONE: - on_handshake_data_received_from_peer(s, 1); - break; - case GRPC_ENDPOINT_ERROR: - on_handshake_data_received_from_peer(s, 0); - break; - case GRPC_ENDPOINT_PENDING: - break; - } + grpc_endpoint_read(s->wrapped_endpoint, &s->incoming, + &s->on_handshake_data_received_from_peer, call_list); return; } else { - send_handshake_bytes_to_peer(s); + send_handshake_bytes_to_peer(s, call_list); return; } } @@ -223,7 +212,7 @@ static void on_handshake_data_received_from_peer(void *setup, int success) { if (result != TSI_OK) { gpr_log(GPR_ERROR, "Handshake failed with error %s", tsi_result_to_string(result)); - secure_transport_setup_done(s, 0); + secure_transport_setup_done(s, 0, call_list); return; } @@ -233,7 +222,7 @@ static void on_handshake_data_received_from_peer(void *setup, int success) { num_left_overs = (has_left_overs_in_current_slice ? 1 : 0) + s->incoming.count - i - 1; if (num_left_overs == 0) { - check_peer(s); + check_peer(s, call_list); return; } /* Put the leftovers in our buffer (ownership transfered). */ @@ -247,51 +236,41 @@ static void on_handshake_data_received_from_peer(void *setup, int success) { gpr_slice_buffer_addn( &s->left_overs, &s->incoming.slices[i + 1], num_left_overs - (size_t)has_left_overs_in_current_slice); - check_peer(s); + check_peer(s, call_list); } /* If setup is NULL, the setup is done. */ -static void on_handshake_data_sent_to_peer(void *setup, int success) { +static void on_handshake_data_sent_to_peer(void *setup, int success, + grpc_call_list *call_list) { grpc_secure_transport_setup *s = setup; /* Make sure that write is OK. */ if (!success) { gpr_log(GPR_ERROR, "Write failed."); - if (setup != NULL) secure_transport_setup_done(s, 0); + if (setup != NULL) secure_transport_setup_done(s, 0, call_list); return; } /* We may be done. */ if (tsi_handshaker_is_in_progress(s->handshaker)) { - /* TODO(klempner,jboeuf): This should probably use the client setup - deadline */ - switch (grpc_endpoint_read(s->wrapped_endpoint, &s->incoming, - &s->on_handshake_data_received_from_peer)) { - case GRPC_ENDPOINT_ERROR: - on_handshake_data_received_from_peer(s, 0); - break; - case GRPC_ENDPOINT_PENDING: - break; - case GRPC_ENDPOINT_DONE: - on_handshake_data_received_from_peer(s, 1); - break; - } + grpc_endpoint_read(s->wrapped_endpoint, &s->incoming, + &s->on_handshake_data_received_from_peer, call_list); } else { - check_peer(s); + check_peer(s, call_list); } } void grpc_setup_secure_transport(grpc_security_connector *connector, grpc_endpoint *nonsecure_endpoint, grpc_secure_transport_setup_done_cb cb, - void *user_data) { + void *user_data, grpc_call_list *call_list) { grpc_security_status result = GRPC_SECURITY_OK; grpc_secure_transport_setup *s = gpr_malloc(sizeof(grpc_secure_transport_setup)); memset(s, 0, sizeof(grpc_secure_transport_setup)); result = grpc_security_connector_create_handshaker(connector, &s->handshaker); if (result != GRPC_SECURITY_OK) { - secure_transport_setup_done(s, 0); + secure_transport_setup_done(s, 0, call_list); return; } s->connector = @@ -308,5 +287,5 @@ void grpc_setup_secure_transport(grpc_security_connector *connector, gpr_slice_buffer_init(&s->left_overs); gpr_slice_buffer_init(&s->outgoing); gpr_slice_buffer_init(&s->incoming); - send_handshake_bytes_to_peer(s); + send_handshake_bytes_to_peer(s, call_list); } diff --git a/src/core/security/secure_transport_setup.h b/src/core/security/secure_transport_setup.h index d9b802556d..867726ff4c 100644 --- a/src/core/security/secure_transport_setup.h +++ b/src/core/security/secure_transport_setup.h @@ -42,12 +42,13 @@ /* Ownership of the secure_endpoint is transfered. */ typedef void (*grpc_secure_transport_setup_done_cb)( void *user_data, grpc_security_status status, - grpc_endpoint *wrapped_endpoint, grpc_endpoint *secure_endpoint); + grpc_endpoint *wrapped_endpoint, grpc_endpoint *secure_endpoint, + grpc_call_list *call_list); /* Calls the callback upon completion. */ void grpc_setup_secure_transport(grpc_security_connector *connector, grpc_endpoint *nonsecure_endpoint, grpc_secure_transport_setup_done_cb cb, - void *user_data); + void *user_data, grpc_call_list *call_list); #endif /* GRPC_INTERNAL_CORE_SECURITY_SECURE_TRANSPORT_SETUP_H */ diff --git a/src/core/security/security_connector.c b/src/core/security/security_connector.c index ba9ac68c5f..dac6a5a5fc 100644 --- a/src/core/security/security_connector.c +++ b/src/core/security/security_connector.c @@ -119,9 +119,9 @@ grpc_security_status grpc_security_connector_check_peer( grpc_security_status grpc_channel_security_connector_check_call_host( grpc_channel_security_connector *sc, const char *host, - grpc_security_check_cb cb, void *user_data) { + grpc_security_check_cb cb, void *user_data, grpc_call_list *call_list) { if (sc == NULL || sc->check_call_host == NULL) return GRPC_SECURITY_ERROR; - return sc->check_call_host(sc, host, cb, user_data); + return sc->check_call_host(sc, host, cb, user_data, call_list); } #ifdef GRPC_SECURITY_CONNECTOR_REFCOUNT_DEBUG @@ -275,11 +275,11 @@ end: static grpc_security_status fake_channel_check_call_host( grpc_channel_security_connector *sc, const char *host, - grpc_security_check_cb cb, void *user_data) { + grpc_security_check_cb cb, void *user_data, grpc_call_list *call_list) { grpc_fake_channel_security_connector *c = (grpc_fake_channel_security_connector *)sc; if (c->call_host_check_is_async) { - cb(user_data, GRPC_SECURITY_OK); + cb(user_data, GRPC_SECURITY_OK, call_list); return GRPC_SECURITY_PENDING; } else { return GRPC_SECURITY_OK; @@ -495,7 +495,7 @@ static grpc_security_status ssl_server_check_peer(grpc_security_connector *sc, static grpc_security_status ssl_channel_check_call_host( grpc_channel_security_connector *sc, const char *host, - grpc_security_check_cb cb, void *user_data) { + grpc_security_check_cb cb, void *user_data, grpc_call_list *call_list) { grpc_ssl_channel_security_connector *c = (grpc_ssl_channel_security_connector *)sc; diff --git a/src/core/security/security_connector.h b/src/core/security/security_connector.h index 2c9aa1c5a4..d78d6ad69f 100644 --- a/src/core/security/security_connector.h +++ b/src/core/security/security_connector.h @@ -61,7 +61,8 @@ typedef struct grpc_security_connector grpc_security_connector; #define GRPC_SECURITY_CONNECTOR_ARG "grpc.security_connector" typedef void (*grpc_security_check_cb)(void *user_data, - grpc_security_status status); + grpc_security_status status, + grpc_call_list *call_list); typedef struct { void (*destroy)(grpc_security_connector *sc); @@ -138,7 +139,8 @@ struct grpc_channel_security_connector { grpc_security_status (*check_call_host)(grpc_channel_security_connector *sc, const char *host, grpc_security_check_cb cb, - void *user_data); + void *user_data, + grpc_call_list *call_list); }; /* Checks that the host that will be set for a call is acceptable. @@ -148,7 +150,7 @@ struct grpc_channel_security_connector { GRPC_SECURITY_PENDING unless an error is detected early on. */ grpc_security_status grpc_channel_security_connector_check_call_host( grpc_channel_security_connector *sc, const char *host, - grpc_security_check_cb cb, void *user_data); + grpc_security_check_cb cb, void *user_data, grpc_call_list *call_list); /* --- Creation security connectors. --- */ diff --git a/src/core/security/server_auth_filter.c b/src/core/security/server_auth_filter.c index 8468697472..2376cbbeb3 100644 --- a/src/core/security/server_auth_filter.c +++ b/src/core/security/server_auth_filter.c @@ -109,12 +109,14 @@ static grpc_mdelem *remove_consumed_md(void *user_data, grpc_mdelem *md) { return md; } +/* called from application code */ static void on_md_processing_done( void *user_data, const grpc_metadata *consumed_md, size_t num_consumed_md, const grpc_metadata *response_md, size_t num_response_md, grpc_status_code status, const char *error_details) { grpc_call_element *elem = user_data; call_data *calld = elem->call_data; + grpc_call_list call_list = GRPC_CALL_LIST_INIT; /* TODO(jboeuf): Implement support for response_md. */ if (response_md != NULL && num_response_md > 0) { @@ -129,7 +131,7 @@ static void on_md_processing_done( grpc_metadata_batch_filter(&calld->md_op->data.metadata, remove_consumed_md, elem); grpc_metadata_array_destroy(&calld->md); - calld->on_done_recv->cb(calld->on_done_recv->cb_arg, 1); + calld->on_done_recv->cb(calld->on_done_recv->cb_arg, 1, &call_list); } else { gpr_slice message; grpc_metadata_array_destroy(&calld->md); @@ -139,11 +141,14 @@ static void on_md_processing_done( message = gpr_slice_from_copied_string(error_details); grpc_sopb_reset(calld->recv_ops); grpc_transport_stream_op_add_close(&calld->transport_op, status, &message); - grpc_call_next_op(elem, &calld->transport_op); + grpc_call_next_op(elem, &calld->transport_op, &call_list); } + + grpc_call_list_run(&call_list); } -static void auth_on_recv(void *user_data, int success) { +static void auth_on_recv(void *user_data, int success, + grpc_call_list *call_list) { grpc_call_element *elem = user_data; call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; @@ -164,7 +169,7 @@ static void auth_on_recv(void *user_data, int success) { return; } } - calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); + calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success, call_list); } static void set_recv_ops_md_callbacks(grpc_call_element *elem, @@ -186,9 +191,10 @@ static void set_recv_ops_md_callbacks(grpc_call_element *elem, op contains type and call direction information, in addition to the data that is being sent or received. */ static void auth_start_transport_op(grpc_call_element *elem, - grpc_transport_stream_op *op) { + grpc_transport_stream_op *op, + grpc_call_list *call_list) { set_recv_ops_md_callbacks(elem, op); - grpc_call_next_op(elem, op); + grpc_call_next_op(elem, op, call_list); } /* Constructor for call_data */ @@ -227,12 +233,14 @@ static void init_call_elem(grpc_call_element *elem, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_call_element *elem) {} +static void destroy_call_elem(grpc_call_element *elem, + grpc_call_list *call_list) {} /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *mdctx, - int is_first, int is_last) { + int is_first, int is_last, + grpc_call_list *call_list) { grpc_security_connector *sc = grpc_find_security_connector_in_args(args); grpc_auth_metadata_processor *processor = grpc_find_auth_metadata_processor_in_args(args); @@ -256,7 +264,8 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, } /* Destructor for channel data */ -static void destroy_channel_elem(grpc_channel_element *elem) { +static void destroy_channel_elem(grpc_channel_element *elem, + grpc_call_list *call_list) { /* grab pointers to our data from the channel element */ channel_data *chand = elem->channel_data; GRPC_SECURITY_CONNECTOR_UNREF(chand->security_connector, diff --git a/src/core/security/server_secure_chttp2.c b/src/core/security/server_secure_chttp2.c index a6f50712f5..e6e2eee658 100644 --- a/src/core/security/server_secure_chttp2.c +++ b/src/core/security/server_secure_chttp2.c @@ -66,6 +66,7 @@ typedef struct grpc_server_secure_state { int is_shutdown; gpr_mu mu; gpr_refcount refcount; + grpc_closure destroy_closure; } grpc_server_secure_state; static void state_ref(grpc_server_secure_state *state) { @@ -127,7 +128,8 @@ static int remove_tcp_from_list_locked(grpc_server_secure_state *state, static void on_secure_transport_setup_done(void *statep, grpc_security_status status, grpc_endpoint *wrapped_endpoint, - grpc_endpoint *secure_endpoint) { + grpc_endpoint *secure_endpoint, + grpc_call_list *call_list) { grpc_server_secure_state *state = statep; grpc_transport *transport; grpc_mdctx *mdctx; @@ -137,16 +139,16 @@ static void on_secure_transport_setup_done(void *statep, remove_tcp_from_list_locked(state, wrapped_endpoint); if (!state->is_shutdown) { mdctx = grpc_mdctx_create(); - workqueue = grpc_workqueue_create(); + workqueue = grpc_workqueue_create(call_list); transport = grpc_create_chttp2_transport( grpc_server_get_channel_args(state->server), secure_endpoint, mdctx, - workqueue, 0); + 0); setup_transport(state, transport, mdctx, workqueue); grpc_chttp2_transport_start_reading(transport, NULL, 0); } else { /* We need to consume this here, because the server may already have gone * away. */ - grpc_endpoint_destroy(secure_endpoint); + grpc_endpoint_destroy(secure_endpoint, call_list); } gpr_mu_unlock(&state->mu); } else { @@ -158,7 +160,8 @@ static void on_secure_transport_setup_done(void *statep, state_unref(state); } -static void on_accept(void *statep, grpc_endpoint *tcp) { +static void on_accept(void *statep, grpc_endpoint *tcp, + grpc_call_list *call_list) { grpc_server_secure_state *state = statep; tcp_endpoint_list *node; state_ref(state); @@ -169,22 +172,24 @@ static void on_accept(void *statep, grpc_endpoint *tcp) { state->handshaking_tcp_endpoints = node; gpr_mu_unlock(&state->mu); grpc_setup_secure_transport(state->sc, tcp, on_secure_transport_setup_done, - state); + state, call_list); } /* Server callback: start listening on our ports */ static void start(grpc_server *server, void *statep, grpc_pollset **pollsets, - size_t pollset_count) { + size_t pollset_count, grpc_call_list *call_list) { grpc_server_secure_state *state = statep; - grpc_tcp_server_start(state->tcp, pollsets, pollset_count, on_accept, state); + grpc_tcp_server_start(state->tcp, pollsets, pollset_count, on_accept, state, + call_list); } -static void destroy_done(void *statep) { +static void destroy_done(void *statep, int success, grpc_call_list *call_list) { grpc_server_secure_state *state = statep; grpc_server_listener_destroy_done(state->server); gpr_mu_lock(&state->mu); while (state->handshaking_tcp_endpoints != NULL) { - grpc_endpoint_shutdown(state->handshaking_tcp_endpoints->tcp_endpoint); + grpc_endpoint_shutdown(state->handshaking_tcp_endpoints->tcp_endpoint, + call_list); remove_tcp_from_list_locked(state, state->handshaking_tcp_endpoints->tcp_endpoint); } @@ -194,14 +199,16 @@ static void destroy_done(void *statep) { /* Server callback: destroy the tcp listener (so we don't generate further callbacks) */ -static void destroy(grpc_server *server, void *statep) { +static void destroy(grpc_server *server, void *statep, + grpc_call_list *call_list) { grpc_server_secure_state *state = statep; grpc_tcp_server *tcp; gpr_mu_lock(&state->mu); state->is_shutdown = 1; tcp = state->tcp; gpr_mu_unlock(&state->mu); - grpc_tcp_server_destroy(tcp, destroy_done, state); + grpc_closure_init(&state->destroy_closure, destroy_done, state); + grpc_tcp_server_destroy(tcp, &state->destroy_closure, call_list); } int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, @@ -215,6 +222,7 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, int port_temp; grpc_security_status status = GRPC_SECURITY_ERROR; grpc_security_connector *sc = NULL; + grpc_call_list call_list = GRPC_CALL_LIST_INIT; /* create security context */ if (creds == NULL) goto error; @@ -277,6 +285,7 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, /* Register with the server only upon success */ grpc_server_add_listener(server, state, start, destroy); + grpc_call_list_run(&call_list); return port_num; /* Error path: cleanup and return */ @@ -288,10 +297,11 @@ error: grpc_resolved_addresses_destroy(resolved); } if (tcp) { - grpc_tcp_server_destroy(tcp, NULL, NULL); + grpc_tcp_server_destroy(tcp, NULL, &call_list); } if (state) { gpr_free(state); } + grpc_call_list_run(&call_list); return 0; } diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index d325b3f77d..24d8b3aa2c 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -64,6 +64,12 @@ typedef struct { gpr_mu mu; grpc_endpoint *connecting_endpoint; + grpc_endpoint *newly_connecting_endpoint; + + grpc_closure connected_closure; + + grpc_mdctx *mdctx; + grpc_workqueue *workqueue; } connector; static void connector_ref(grpc_connector *con) { @@ -71,9 +77,11 @@ static void connector_ref(grpc_connector *con) { gpr_ref(&c->refs); } -static void connector_unref(grpc_connector *con) { +static void connector_unref(grpc_connector *con, grpc_call_list *call_list) { connector *c = (connector *)con; if (gpr_unref(&c->refs)) { + grpc_mdctx_unref(c->mdctx); + GRPC_WORKQUEUE_UNREF(c->workqueue, "connector", call_list); gpr_free(c); } } @@ -81,7 +89,8 @@ static void connector_unref(grpc_connector *con) { static void on_secure_transport_setup_done(void *arg, grpc_security_status status, grpc_endpoint *wrapped_endpoint, - grpc_endpoint *secure_endpoint) { + grpc_endpoint *secure_endpoint, + grpc_call_list *call_list) { connector *c = arg; grpc_closure *notify; gpr_mu_lock(&c->mu); @@ -99,8 +108,7 @@ static void on_secure_transport_setup_done(void *arg, c->connecting_endpoint = NULL; gpr_mu_unlock(&c->mu); c->result->transport = grpc_create_chttp2_transport( - c->args.channel_args, secure_endpoint, c->args.metadata_context, - c->args.workqueue, 1); + c->args.channel_args, secure_endpoint, c->mdctx, 1); grpc_chttp2_transport_start_reading(c->result->transport, NULL, 0); c->result->filters = gpr_malloc(sizeof(grpc_channel_filter *) * 2); c->result->filters[0] = &grpc_http_client_filter; @@ -109,28 +117,29 @@ static void on_secure_transport_setup_done(void *arg, } notify = c->notify; c->notify = NULL; - notify->cb(notify->cb_arg, 1); + notify->cb(notify->cb_arg, 1, call_list); } -static void connected(void *arg, grpc_endpoint *tcp) { +static void connected(void *arg, int success, grpc_call_list *call_list) { connector *c = arg; grpc_closure *notify; + grpc_endpoint *tcp = c->newly_connecting_endpoint; if (tcp != NULL) { gpr_mu_lock(&c->mu); GPR_ASSERT(c->connecting_endpoint == NULL); c->connecting_endpoint = tcp; gpr_mu_unlock(&c->mu); grpc_setup_secure_transport(&c->security_connector->base, tcp, - on_secure_transport_setup_done, c); + on_secure_transport_setup_done, c, call_list); } else { memset(c->result, 0, sizeof(*c->result)); notify = c->notify; c->notify = NULL; - notify->cb(notify->cb_arg, 1); + notify->cb(notify->cb_arg, 1, call_list); } } -static void connector_shutdown(grpc_connector *con) { +static void connector_shutdown(grpc_connector *con, grpc_call_list *call_list) { connector *c = (connector *)con; grpc_endpoint *ep; gpr_mu_lock(&c->mu); @@ -138,14 +147,14 @@ static void connector_shutdown(grpc_connector *con) { c->connecting_endpoint = NULL; gpr_mu_unlock(&c->mu); if (ep) { - grpc_endpoint_shutdown(ep); + grpc_endpoint_shutdown(ep, call_list); } } static void connector_connect(grpc_connector *con, const grpc_connect_in_args *args, grpc_connect_out_args *result, - grpc_closure *notify) { + grpc_closure *notify, grpc_call_list *call_list) { connector *c = (connector *)con; GPR_ASSERT(c->notify == NULL); GPR_ASSERT(notify->cb); @@ -155,9 +164,10 @@ static void connector_connect(grpc_connector *con, gpr_mu_lock(&c->mu); GPR_ASSERT(c->connecting_endpoint == NULL); gpr_mu_unlock(&c->mu); - grpc_tcp_client_connect(connected, c, args->interested_parties, - args->workqueue, args->addr, args->addr_len, - args->deadline); + grpc_closure_init(&c->connected_closure, connected, c); + grpc_tcp_client_connect(&c->connected_closure, &c->newly_connecting_endpoint, + args->interested_parties, args->addr, args->addr_len, + args->deadline, call_list); } static const grpc_connector_vtable connector_vtable = { @@ -190,7 +200,8 @@ static void subchannel_factory_unref(grpc_subchannel_factory *scf) { } static grpc_subchannel *subchannel_factory_create_subchannel( - grpc_subchannel_factory *scf, grpc_subchannel_args *args) { + grpc_subchannel_factory *scf, grpc_subchannel_args *args, + grpc_call_list *call_list) { subchannel_factory *f = (subchannel_factory *)scf; connector *c = gpr_malloc(sizeof(*c)); grpc_channel_args *final_args = @@ -199,12 +210,15 @@ static grpc_subchannel *subchannel_factory_create_subchannel( memset(c, 0, sizeof(*c)); c->base.vtable = &connector_vtable; c->security_connector = f->security_connector; + c->mdctx = f->mdctx; + grpc_mdctx_ref(c->mdctx); + c->workqueue = grpc_channel_get_workqueue(f->master); + GRPC_WORKQUEUE_REF(c->workqueue, "connector"); gpr_ref_init(&c->refs, 1); - args->mdctx = f->mdctx; args->args = final_args; args->master = f->master; s = grpc_subchannel_create(&c->base, args); - grpc_connector_unref(&c->base); + grpc_connector_unref(&c->base, call_list); grpc_channel_args_destroy(final_args); return s; } @@ -232,6 +246,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, subchannel_factory *f; #define MAX_FILTERS 3 const grpc_channel_filter *filters[MAX_FILTERS]; + grpc_call_list call_list = GRPC_CALL_LIST_INIT; size_t n = 0; GPR_ASSERT(reserved == NULL); @@ -250,7 +265,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, "Failed to create security connector."); } mdctx = grpc_mdctx_create(); - workqueue = grpc_workqueue_create(); + workqueue = grpc_workqueue_create(&call_list); connector_arg = grpc_security_connector_to_arg(&connector->base); args_copy = grpc_channel_args_copy_and_add( @@ -282,8 +297,8 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, } grpc_client_channel_set_resolver(grpc_channel_get_channel_stack(channel), - resolver); - GRPC_RESOLVER_UNREF(resolver, "create"); + resolver, &call_list); + GRPC_RESOLVER_UNREF(resolver, "create", &call_list); grpc_subchannel_factory_unref(&f->base); GRPC_SECURITY_CONNECTOR_UNREF(&connector->base, "channel_create"); @@ -292,5 +307,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, grpc_channel_args_destroy(new_args_from_connector); } + grpc_call_list_run(&call_list); + return channel; } diff --git a/src/core/surface/server.h b/src/core/surface/server.h index 1d82d07ced..f87296797c 100644 --- a/src/core/surface/server.h +++ b/src/core/surface/server.h @@ -45,11 +45,11 @@ grpc_server *grpc_server_create_from_filters( /* Add a listener to the server: when the server starts, it will call start, and when it shuts down, it will call destroy */ -void grpc_server_add_listener(grpc_server *server, void *listener, - void (*start)(grpc_server *server, void *arg, - grpc_pollset **pollsets, - size_t npollsets), - void (*destroy)(grpc_server *server, void *arg)); +void grpc_server_add_listener( + grpc_server *server, void *listener, + void (*start)(grpc_server *server, void *arg, grpc_pollset **pollsets, + size_t npollsets, grpc_call_list *call_list), + void (*destroy)(grpc_server *server, void *arg, grpc_call_list *call_list)); void grpc_server_listener_destroy_done(void *server); diff --git a/src/core/transport/chttp2_transport.h b/src/core/transport/chttp2_transport.h index 8bd8af6236..fa0d6e4151 100644 --- a/src/core/transport/chttp2_transport.h +++ b/src/core/transport/chttp2_transport.h @@ -35,7 +35,6 @@ #define GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_TRANSPORT_H #include "src/core/iomgr/endpoint.h" -#include "src/core/iomgr/workqueue.h" #include "src/core/transport/transport.h" extern int grpc_http_trace; @@ -43,7 +42,7 @@ extern int grpc_flowctl_trace; grpc_transport *grpc_create_chttp2_transport( const grpc_channel_args *channel_args, grpc_endpoint *ep, - grpc_mdctx *metadata_context, grpc_workqueue *workqueue, int is_client); + grpc_mdctx *metadata_context, int is_client); void grpc_chttp2_transport_start_reading(grpc_transport *transport, gpr_slice *slices, size_t nslices); |