diff options
Diffstat (limited to 'src/core/lib/surface')
-rw-r--r-- | src/core/lib/surface/call.c | 174 | ||||
-rw-r--r-- | src/core/lib/surface/call.h | 1 | ||||
-rw-r--r-- | src/core/lib/surface/channel.c | 6 | ||||
-rw-r--r-- | src/core/lib/surface/completion_queue.c | 46 | ||||
-rw-r--r-- | src/core/lib/surface/completion_queue.h | 3 | ||||
-rw-r--r-- | src/core/lib/surface/init.c | 2 | ||||
-rw-r--r-- | src/core/lib/surface/lame_client.c | 2 | ||||
-rw-r--r-- | src/core/lib/surface/server.c | 9 |
8 files changed, 133 insertions, 110 deletions
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index 3352e427cd..af32c5b378 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -101,7 +101,18 @@ typedef struct { grpc_error *error; } received_status; -#define MAX_ERRORS_PER_BATCH 3 +static gpr_atm pack_received_status(received_status r) { + return r.is_set ? (1 | (gpr_atm)r.error) : 0; +} + +static received_status unpack_received_status(gpr_atm atm) { + return (atm & 1) == 0 + ? (received_status){.is_set = false, .error = GRPC_ERROR_NONE} + : (received_status){.is_set = true, + .error = (grpc_error *)(atm & ~(gpr_atm)1)}; +} + +#define MAX_ERRORS_PER_BATCH 4 typedef struct batch_control { grpc_call *call; @@ -142,8 +153,6 @@ struct grpc_call { bool destroy_called; /** flag indicating that cancellation is inherited */ bool cancellation_is_inherited; - /** bitmask of live batches */ - uint8_t used_batches; /** which ops are in-flight */ bool sent_initial_metadata; bool sending_message; @@ -165,8 +174,8 @@ struct grpc_call { Element 0 is initial metadata, element 1 is trailing metadata. */ grpc_metadata_array *buffered_metadata[2]; - /* Received call statuses from various sources */ - received_status status[STATUS_SOURCE_COUNT]; + /* Packed received call statuses from various sources */ + gpr_atm status[STATUS_SOURCE_COUNT]; /* Call data useful used for reporting. Only valid after the call has * completed */ @@ -245,7 +254,7 @@ static void set_status_from_error(grpc_exec_ctx *exec_ctx, grpc_call *call, static void process_data_after_md(grpc_exec_ctx *exec_ctx, batch_control *bctl); static void post_batch_completion(grpc_exec_ctx *exec_ctx, batch_control *bctl); static void add_batch_error(grpc_exec_ctx *exec_ctx, batch_control *bctl, - grpc_error *error); + grpc_error *error, bool has_cancelled); static void add_init_error(grpc_error **composite, grpc_error *new) { if (new == GRPC_ERROR_NONE) return; @@ -446,7 +455,8 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), c->start_time); for (i = 0; i < STATUS_SOURCE_COUNT; i++) { - GRPC_ERROR_UNREF(c->status[i].error); + GRPC_ERROR_UNREF( + unpack_received_status(gpr_atm_no_barrier_load(&c->status[i])).error); } grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c), &c->final_info, c); @@ -481,7 +491,10 @@ void grpc_call_destroy(grpc_call *c) { c->destroy_called = 1; cancel = !c->received_final_op; gpr_mu_unlock(&c->mu); - if (cancel) grpc_call_cancel(c, NULL); + if (cancel) { + cancel_with_error(&exec_ctx, c, STATUS_FROM_API_OVERRIDE, + GRPC_ERROR_CANCELLED); + } GRPC_CALL_INTERNAL_UNREF(&exec_ctx, c, "destroy"); grpc_exec_ctx_finish(&exec_ctx); GPR_TIMER_END("grpc_call_destroy", 0); @@ -490,8 +503,11 @@ void grpc_call_destroy(grpc_call *c) { grpc_call_error grpc_call_cancel(grpc_call *call, void *reserved) { GRPC_API_TRACE("grpc_call_cancel(call=%p, reserved=%p)", 2, (call, reserved)); GPR_ASSERT(!reserved); - return grpc_call_cancel_with_status(call, GRPC_STATUS_CANCELLED, "Cancelled", - NULL); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + cancel_with_error(&exec_ctx, call, STATUS_FROM_API_OVERRIDE, + GRPC_ERROR_CANCELLED); + grpc_exec_ctx_finish(&exec_ctx); + return GRPC_CALL_OK; } static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call, @@ -608,13 +624,12 @@ static void cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c, */ static bool get_final_status_from( - grpc_call *call, status_source from_source, bool allow_ok_status, + grpc_call *call, grpc_error *error, bool allow_ok_status, void (*set_value)(grpc_status_code code, void *user_data), void *set_value_user_data, grpc_slice *details) { grpc_status_code code; const char *msg = NULL; - grpc_error_get_status(call->status[from_source].error, call->send_deadline, - &code, &msg, NULL); + grpc_error_get_status(error, call->send_deadline, &code, &msg, NULL); if (code == GRPC_STATUS_OK && !allow_ok_status) { return false; } @@ -632,12 +647,15 @@ static void get_final_status(grpc_call *call, void *user_data), void *set_value_user_data, grpc_slice *details) { int i; + received_status status[STATUS_SOURCE_COUNT]; + for (i = 0; i < STATUS_SOURCE_COUNT; i++) { + status[i] = unpack_received_status(gpr_atm_acq_load(&call->status[i])); + } if (grpc_call_error_trace) { gpr_log(GPR_DEBUG, "get_final_status %s", call->is_client ? "CLI" : "SVR"); for (i = 0; i < STATUS_SOURCE_COUNT; i++) { - if (call->status[i].is_set) { - gpr_log(GPR_DEBUG, " %d: %s", i, - grpc_error_string(call->status[i].error)); + if (status[i].is_set) { + gpr_log(GPR_DEBUG, " %d: %s", i, grpc_error_string(status[i].error)); } } } @@ -647,9 +665,9 @@ static void get_final_status(grpc_call *call, /* search for the best status we can present: ideally the error we use has a clearly defined grpc-status, and we'll prefer that. */ for (i = 0; i < STATUS_SOURCE_COUNT; i++) { - if (call->status[i].is_set && - grpc_error_has_clear_grpc_status(call->status[i].error)) { - if (get_final_status_from(call, (status_source)i, allow_ok_status != 0, + if (status[i].is_set && + grpc_error_has_clear_grpc_status(status[i].error)) { + if (get_final_status_from(call, status[i].error, allow_ok_status != 0, set_value, set_value_user_data, details)) { return; } @@ -657,8 +675,8 @@ static void get_final_status(grpc_call *call, } /* If no clearly defined status exists, search for 'anything' */ for (i = 0; i < STATUS_SOURCE_COUNT; i++) { - if (call->status[i].is_set) { - if (get_final_status_from(call, (status_source)i, allow_ok_status != 0, + if (status[i].is_set) { + if (get_final_status_from(call, status[i].error, allow_ok_status != 0, set_value, set_value_user_data, details)) { return; } @@ -675,12 +693,13 @@ static void get_final_status(grpc_call *call, static void set_status_from_error(grpc_exec_ctx *exec_ctx, grpc_call *call, status_source source, grpc_error *error) { - if (call->status[source].is_set) { + if (!gpr_atm_rel_cas(&call->status[source], + pack_received_status((received_status){ + .is_set = false, .error = GRPC_ERROR_NONE}), + pack_received_status((received_status){ + .is_set = true, .error = error}))) { GRPC_ERROR_UNREF(error); - return; } - call->status[source].is_set = true; - call->status[source].error = error; } /******************************************************************************* @@ -991,25 +1010,48 @@ static bool are_initial_metadata_flags_valid(uint32_t flags, bool is_client) { return !(flags & invalid_positions); } -static batch_control *allocate_batch_control(grpc_call *call) { - size_t i; - for (i = 0; i < MAX_CONCURRENT_BATCHES; i++) { - if ((call->used_batches & (1 << i)) == 0) { - call->used_batches = (uint8_t)(call->used_batches | (uint8_t)(1 << i)); - return &call->active_batches[i]; - } +static int batch_slot_for_op(grpc_op_type type) { + switch (type) { + case GRPC_OP_SEND_INITIAL_METADATA: + return 0; + case GRPC_OP_SEND_MESSAGE: + return 1; + case GRPC_OP_SEND_CLOSE_FROM_CLIENT: + case GRPC_OP_SEND_STATUS_FROM_SERVER: + return 2; + case GRPC_OP_RECV_INITIAL_METADATA: + return 3; + case GRPC_OP_RECV_MESSAGE: + return 4; + case GRPC_OP_RECV_CLOSE_ON_SERVER: + case GRPC_OP_RECV_STATUS_ON_CLIENT: + return 5; + } + GPR_UNREACHABLE_CODE(return 123456789); +} + +static batch_control *allocate_batch_control(grpc_call *call, + const grpc_op *ops, + size_t num_ops) { + int slot = batch_slot_for_op(ops[0].op); + for (size_t i = 1; i < num_ops; i++) { + int op_slot = batch_slot_for_op(ops[i].op); + slot = GPR_MIN(slot, op_slot); } - return NULL; + batch_control *bctl = &call->active_batches[slot]; + if (bctl->call != NULL) { + return NULL; + } + memset(bctl, 0, sizeof(*bctl)); + bctl->call = call; + return bctl; } static void finish_batch_completion(grpc_exec_ctx *exec_ctx, void *user_data, grpc_cq_completion *storage) { batch_control *bctl = user_data; grpc_call *call = bctl->call; - gpr_mu_lock(&call->mu); - call->used_batches = (uint8_t)( - call->used_batches & ~(uint8_t)(1 << (bctl - call->active_batches))); - gpr_mu_unlock(&call->mu); + bctl->call = NULL; GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "completion"); } @@ -1092,12 +1134,8 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx, if (bctl->is_notify_tag_closure) { /* unrefs bctl->error */ + bctl->call = NULL; grpc_closure_run(exec_ctx, bctl->notify_tag, error); - gpr_mu_lock(&call->mu); - bctl->call->used_batches = - (uint8_t)(bctl->call->used_batches & - ~(uint8_t)(1 << (bctl - bctl->call->active_batches))); - gpr_mu_unlock(&call->mu); GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "completion"); } else { /* unrefs bctl->error */ @@ -1185,6 +1223,11 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp, grpc_call *call = bctl->call; gpr_mu_lock(&bctl->call->mu); if (error != GRPC_ERROR_NONE) { + if (call->receiving_stream != NULL) { + grpc_byte_stream_destroy(exec_ctx, call->receiving_stream); + call->receiving_stream = NULL; + } + add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error), true); cancel_with_error(exec_ctx, call, STATUS_FROM_SURFACE, GRPC_ERROR_REF(error)); } @@ -1251,10 +1294,10 @@ static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx, } static void add_batch_error(grpc_exec_ctx *exec_ctx, batch_control *bctl, - grpc_error *error) { + grpc_error *error, bool has_cancelled) { if (error == GRPC_ERROR_NONE) return; int idx = (int)gpr_atm_no_barrier_fetch_add(&bctl->num_errors, 1); - if (idx == 0) { + if (idx == 0 && !has_cancelled) { cancel_with_error(exec_ctx, bctl->call, STATUS_FROM_CORE, GRPC_ERROR_REF(error)); } @@ -1268,7 +1311,7 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx, gpr_mu_lock(&call->mu); - add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error)); + add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error), false); if (error == GRPC_ERROR_NONE) { grpc_metadata_batch *md = &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */]; @@ -1305,10 +1348,15 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, grpc_error *error) { batch_control *bctl = bctlp; - add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error)); + add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error), false); finish_batch_step(exec_ctx, bctl); } +static void free_no_op_completion(grpc_exec_ctx *exec_ctx, void *p, + grpc_cq_completion *completion) { + gpr_free(completion); +} + static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, grpc_call *call, const grpc_op *ops, size_t nops, void *notify_tag, @@ -1323,32 +1371,34 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, grpc_metadata compression_md; GPR_TIMER_BEGIN("grpc_call_start_batch", 0); - GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, notify_tag); - /* TODO(ctiller): this feels like it could be made lock-free */ - gpr_mu_lock(&call->mu); - bctl = allocate_batch_control(call); - memset(bctl, 0, sizeof(*bctl)); - bctl->call = call; - bctl->notify_tag = notify_tag; - bctl->is_notify_tag_closure = (uint8_t)(is_notify_tag_closure != 0); - - grpc_transport_stream_op *stream_op = &bctl->op; - memset(stream_op, 0, sizeof(*stream_op)); - stream_op->covered_by_poller = true; - if (nops == 0) { - GRPC_CALL_INTERNAL_REF(call, "completion"); if (!is_notify_tag_closure) { grpc_cq_begin_op(call->cq, notify_tag); + grpc_cq_end_op(exec_ctx, call->cq, notify_tag, GRPC_ERROR_NONE, + free_no_op_completion, NULL, + gpr_malloc(sizeof(grpc_cq_completion))); + } else { + grpc_closure_sched(exec_ctx, notify_tag, GRPC_ERROR_NONE); } - gpr_mu_unlock(&call->mu); - post_batch_completion(exec_ctx, bctl); error = GRPC_CALL_OK; goto done; } + /* TODO(ctiller): this feels like it could be made lock-free */ + bctl = allocate_batch_control(call, ops, nops); + if (bctl == NULL) { + return GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + } + bctl->notify_tag = notify_tag; + bctl->is_notify_tag_closure = (uint8_t)(is_notify_tag_closure != 0); + + gpr_mu_lock(&call->mu); + grpc_transport_stream_op *stream_op = &bctl->op; + memset(stream_op, 0, sizeof(*stream_op)); + stream_op->covered_by_poller = true; + /* rewrite batch ops into a transport op */ for (i = 0; i < nops; i++) { op = &ops[i]; diff --git a/src/core/lib/surface/call.h b/src/core/lib/surface/call.h index b70343ddf1..7d4d0db28d 100644 --- a/src/core/lib/surface/call.h +++ b/src/core/lib/surface/call.h @@ -110,6 +110,7 @@ void grpc_call_log_batch(char *file, int line, gpr_log_severity severity, /* Set a context pointer. No thread safety guarantees are made wrt this value. */ +/* TODO(#9731): add exec_ctx to destroy */ void grpc_call_context_set(grpc_call *call, grpc_context_index elem, void *value, void (*destroy)(void *value)); /* Get a context pointer. */ diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c index 429dbad7c7..d6acd392c1 100644 --- a/src/core/lib/surface/channel.c +++ b/src/core/lib/surface/channel.c @@ -128,7 +128,8 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, } channel->default_authority = grpc_mdelem_from_slices( exec_ctx, GRPC_MDSTR_AUTHORITY, - grpc_slice_from_copied_string(args->args[i].value.string)); + grpc_slice_intern( + grpc_slice_from_static_string(args->args[i].value.string))); } } else if (0 == strcmp(args->args[i].key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)) { @@ -144,7 +145,8 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, } else { channel->default_authority = grpc_mdelem_from_slices( exec_ctx, GRPC_MDSTR_AUTHORITY, - grpc_slice_from_copied_string(args->args[i].value.string)); + grpc_slice_intern( + grpc_slice_from_static_string(args->args[i].value.string))); } } } else if (0 == strcmp(args->args[i].key, diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c index 1830842d00..b80656aa8d 100644 --- a/src/core/lib/surface/completion_queue.c +++ b/src/core/lib/surface/completion_queue.c @@ -96,9 +96,6 @@ struct grpc_completion_queue { #define POLLSET_FROM_CQ(cq) ((grpc_pollset *)(cq + 1)) #define CQ_FROM_POLLSET(ps) (((grpc_completion_queue *)ps) - 1) -static gpr_mu g_freelist_mu; -static grpc_completion_queue *g_freelist; - int grpc_cq_pluck_trace; int grpc_cq_event_timeout_trace; @@ -113,21 +110,6 @@ int grpc_cq_event_timeout_trace; static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc, grpc_error *error); -void grpc_cq_global_init(void) { gpr_mu_init(&g_freelist_mu); } - -void grpc_cq_global_shutdown(void) { - gpr_mu_destroy(&g_freelist_mu); - while (g_freelist) { - grpc_completion_queue *next = g_freelist->next_free; - grpc_pollset_destroy(POLLSET_FROM_CQ(g_freelist)); -#ifndef NDEBUG - gpr_free(g_freelist->outstanding_tags); -#endif - gpr_free(g_freelist); - g_freelist = next; - } -} - grpc_completion_queue *grpc_completion_queue_create(void *reserved) { grpc_completion_queue *cc; GPR_ASSERT(!reserved); @@ -136,22 +118,12 @@ grpc_completion_queue *grpc_completion_queue_create(void *reserved) { GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved)); - gpr_mu_lock(&g_freelist_mu); - if (g_freelist == NULL) { - gpr_mu_unlock(&g_freelist_mu); - - cc = gpr_malloc(sizeof(grpc_completion_queue) + grpc_pollset_size()); - grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu); + cc = gpr_malloc(sizeof(grpc_completion_queue) + grpc_pollset_size()); + grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu); #ifndef NDEBUG - cc->outstanding_tags = NULL; - cc->outstanding_tag_capacity = 0; + cc->outstanding_tags = NULL; + cc->outstanding_tag_capacity = 0; #endif - } else { - cc = g_freelist; - g_freelist = g_freelist->next_free; - gpr_mu_unlock(&g_freelist_mu); - /* pollset already initialized */ - } /* Initial ref is dropped by grpc_completion_queue_shutdown */ gpr_ref_init(&cc->pending_events, 1); @@ -203,11 +175,11 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) { #endif if (gpr_unref(&cc->owning_refs)) { GPR_ASSERT(cc->completed_head.next == (uintptr_t)&cc->completed_head); - grpc_pollset_reset(POLLSET_FROM_CQ(cc)); - gpr_mu_lock(&g_freelist_mu); - cc->next_free = g_freelist; - g_freelist = cc; - gpr_mu_unlock(&g_freelist_mu); + grpc_pollset_destroy(POLLSET_FROM_CQ(cc)); +#ifndef NDEBUG + gpr_free(cc->outstanding_tags); +#endif + gpr_free(cc); } } diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index c1cafba5f2..5d73dd7216 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -99,7 +99,4 @@ bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc); void grpc_cq_mark_server_cq(grpc_completion_queue *cc); int grpc_cq_is_server_cq(grpc_completion_queue *cc); -void grpc_cq_global_init(void); -void grpc_cq_global_shutdown(void); - #endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */ diff --git a/src/core/lib/surface/init.c b/src/core/lib/surface/init.c index b338ac4c48..91bd014a0e 100644 --- a/src/core/lib/surface/init.c +++ b/src/core/lib/surface/init.c @@ -209,7 +209,6 @@ void grpc_init(void) { grpc_iomgr_init(); grpc_executor_init(); gpr_timers_global_init(); - grpc_cq_global_init(); grpc_handshaker_factory_registry_init(); grpc_security_init(); for (i = 0; i < g_number_of_plugins; i++) { @@ -236,7 +235,6 @@ void grpc_shutdown(void) { gpr_mu_lock(&g_init_mu); if (--g_initializations == 0) { grpc_executor_shutdown(&exec_ctx); - grpc_cq_global_shutdown(); grpc_iomgr_shutdown(&exec_ctx); gpr_timers_global_destroy(); grpc_tracer_shutdown(); diff --git a/src/core/lib/surface/lame_client.c b/src/core/lib/surface/lame_client.c index 48de0e1d5b..49bc4c114b 100644 --- a/src/core/lib/surface/lame_client.c +++ b/src/core/lib/surface/lame_client.c @@ -122,7 +122,7 @@ static void lame_start_transport_op(grpc_exec_ctx *exec_ctx, static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { + const grpc_call_element_args *args) { call_data *calld = elem->call_data; gpr_atm_no_barrier_store(&calld->filled_metadata, 0); return GRPC_ERROR_NONE; diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index 6ab1c0d94d..fdb230b3e2 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -540,7 +540,8 @@ static void publish_new_rpc(grpc_exec_ctx *exec_ctx, void *arg, &calld->kill_zombie_closure, kill_zombie, grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0), grpc_schedule_on_exec_ctx); - grpc_closure_sched(exec_ctx, &calld->kill_zombie_closure, error); + grpc_closure_sched(exec_ctx, &calld->kill_zombie_closure, + GRPC_ERROR_REF(error)); return; } @@ -879,7 +880,7 @@ static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd, static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { + const grpc_call_element_args *args) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; memset(calld, 0, sizeof(call_data)); @@ -1198,7 +1199,9 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, crm->server_registered_method = rm; crm->flags = rm->flags; crm->has_host = has_host; - crm->host = host; + if (has_host) { + crm->host = host; + } crm->method = method; } GPR_ASSERT(slots <= UINT32_MAX); |