|
|
|
@ -72,46 +72,6 @@ |
|
|
|
|
// Used to create arena for the first call.
|
|
|
|
|
#define ESTIMATED_MDELEM_COUNT 16 |
|
|
|
|
|
|
|
|
|
/* Status data for a request can come from several sources; this
|
|
|
|
|
enumerates them all, and acts as a priority sorting for which |
|
|
|
|
status to return to the application - earlier entries override |
|
|
|
|
later ones */ |
|
|
|
|
typedef enum { |
|
|
|
|
/* Status came from the application layer overriding whatever
|
|
|
|
|
the wire says */ |
|
|
|
|
STATUS_FROM_API_OVERRIDE = 0, |
|
|
|
|
/* Status came from 'the wire' - or somewhere below the surface
|
|
|
|
|
layer */ |
|
|
|
|
STATUS_FROM_WIRE, |
|
|
|
|
/* Status was created by some internal channel stack operation: must come via
|
|
|
|
|
add_batch_error */ |
|
|
|
|
STATUS_FROM_CORE, |
|
|
|
|
/* Status was created by some surface error */ |
|
|
|
|
STATUS_FROM_SURFACE, |
|
|
|
|
/* Status came from the server sending status */ |
|
|
|
|
STATUS_FROM_SERVER_STATUS, |
|
|
|
|
STATUS_SOURCE_COUNT |
|
|
|
|
} status_source; |
|
|
|
|
|
|
|
|
|
typedef struct { |
|
|
|
|
bool is_set; |
|
|
|
|
grpc_error* error; |
|
|
|
|
} received_status; |
|
|
|
|
|
|
|
|
|
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) { |
|
|
|
|
if ((atm & 1) == 0) { |
|
|
|
|
return {false, GRPC_ERROR_NONE}; |
|
|
|
|
} else { |
|
|
|
|
return {true, (grpc_error*)(atm & ~static_cast<gpr_atm>(1))}; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#define MAX_ERRORS_PER_BATCH 4 |
|
|
|
|
|
|
|
|
|
typedef struct batch_control { |
|
|
|
|
grpc_call* call; |
|
|
|
|
/* Share memory for cq_completion and notify_tag as they are never needed
|
|
|
|
@ -136,10 +96,7 @@ typedef struct batch_control { |
|
|
|
|
grpc_closure start_batch; |
|
|
|
|
grpc_closure finish_batch; |
|
|
|
|
gpr_refcount steps_to_complete; |
|
|
|
|
|
|
|
|
|
grpc_error* errors[MAX_ERRORS_PER_BATCH]; |
|
|
|
|
gpr_atm num_errors; |
|
|
|
|
|
|
|
|
|
grpc_error* batch_error; |
|
|
|
|
grpc_transport_stream_op_batch op; |
|
|
|
|
} batch_control; |
|
|
|
|
|
|
|
|
@ -167,8 +124,6 @@ struct grpc_call { |
|
|
|
|
grpc_completion_queue* cq; |
|
|
|
|
grpc_polling_entity pollent; |
|
|
|
|
grpc_channel* channel; |
|
|
|
|
// backpointer to owning server if this is a server side call.
|
|
|
|
|
grpc_server* server; |
|
|
|
|
gpr_timespec start_time; |
|
|
|
|
/* parent_call* */ gpr_atm parent_call_atm; |
|
|
|
|
child_call* child; |
|
|
|
@ -204,9 +159,6 @@ struct grpc_call { |
|
|
|
|
// A char* indicating the peer name.
|
|
|
|
|
gpr_atm peer_string; |
|
|
|
|
|
|
|
|
|
/* 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 */ |
|
|
|
|
grpc_call_final_info final_info; |
|
|
|
@ -239,6 +191,7 @@ struct grpc_call { |
|
|
|
|
grpc_closure receiving_initial_metadata_ready; |
|
|
|
|
grpc_closure receiving_trailing_metadata_ready; |
|
|
|
|
uint32_t test_only_last_message_flags; |
|
|
|
|
gpr_atm cancelled; |
|
|
|
|
|
|
|
|
|
grpc_closure release_call; |
|
|
|
|
|
|
|
|
@ -250,8 +203,11 @@ struct grpc_call { |
|
|
|
|
} client; |
|
|
|
|
struct { |
|
|
|
|
int* cancelled; |
|
|
|
|
// backpointer to owning server if this is a server side call.
|
|
|
|
|
grpc_server* server; |
|
|
|
|
} server; |
|
|
|
|
} final_op; |
|
|
|
|
grpc_error* status_error; |
|
|
|
|
|
|
|
|
|
/* recv_state can contain one of the following values:
|
|
|
|
|
RECV_NONE : : no initial metadata and messages received |
|
|
|
@ -289,23 +245,15 @@ grpc_core::TraceFlag grpc_compression_trace(false, "compression"); |
|
|
|
|
|
|
|
|
|
static void execute_batch(grpc_call* call, grpc_transport_stream_op_batch* op, |
|
|
|
|
grpc_closure* start_batch_closure); |
|
|
|
|
static void cancel_with_status(grpc_call* c, status_source source, |
|
|
|
|
grpc_status_code status, |
|
|
|
|
|
|
|
|
|
static void cancel_with_status(grpc_call* c, grpc_status_code status, |
|
|
|
|
const char* description); |
|
|
|
|
static void cancel_with_error(grpc_call* c, status_source source, |
|
|
|
|
grpc_error* error); |
|
|
|
|
static void cancel_with_error(grpc_call* c, grpc_error* error); |
|
|
|
|
static void destroy_call(void* call_stack, grpc_error* error); |
|
|
|
|
static void receiving_slice_ready(void* bctlp, grpc_error* error); |
|
|
|
|
static void get_final_status( |
|
|
|
|
grpc_call* call, void (*set_value)(grpc_status_code code, void* user_data), |
|
|
|
|
void* set_value_user_data, grpc_slice* details, const char** error_string); |
|
|
|
|
static void set_status_value_directly(grpc_status_code status, void* dest); |
|
|
|
|
static void set_status_from_error(grpc_call* call, status_source source, |
|
|
|
|
grpc_error* error); |
|
|
|
|
static void set_final_status(grpc_call* call, grpc_error* error); |
|
|
|
|
static void process_data_after_md(batch_control* bctl); |
|
|
|
|
static void post_batch_completion(batch_control* bctl); |
|
|
|
|
static void add_batch_error(batch_control* bctl, grpc_error* error, |
|
|
|
|
bool has_cancelled); |
|
|
|
|
|
|
|
|
|
static void add_init_error(grpc_error** composite, grpc_error* new_err) { |
|
|
|
|
if (new_err == GRPC_ERROR_NONE) return; |
|
|
|
@ -356,6 +304,7 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, |
|
|
|
|
gpr_arena_alloc(arena, GPR_ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call)) + |
|
|
|
|
channel_stack->call_stack_size)); |
|
|
|
|
gpr_ref_init(&call->ext_ref, 1); |
|
|
|
|
gpr_atm_no_barrier_store(&call->cancelled, 0); |
|
|
|
|
call->arena = arena; |
|
|
|
|
grpc_call_combiner_init(&call->call_combiner); |
|
|
|
|
*out_call = call; |
|
|
|
@ -369,7 +318,6 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, |
|
|
|
|
grpc_slice path = grpc_empty_slice(); |
|
|
|
|
if (call->is_client) { |
|
|
|
|
GRPC_STATS_INC_CLIENT_CALLS_CREATED(); |
|
|
|
|
call->server = nullptr; |
|
|
|
|
GPR_ASSERT(args->add_initial_metadata_count < |
|
|
|
|
MAX_SEND_EXTRA_METADATA_COUNT); |
|
|
|
|
for (i = 0; i < args->add_initial_metadata_count; i++) { |
|
|
|
@ -384,7 +332,7 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, |
|
|
|
|
static_cast<int>(args->add_initial_metadata_count); |
|
|
|
|
} else { |
|
|
|
|
GRPC_STATS_INC_SERVER_CALLS_CREATED(); |
|
|
|
|
call->server = args->server; |
|
|
|
|
call->final_op.server.server = args->server; |
|
|
|
|
GPR_ASSERT(args->add_initial_metadata_count == 0); |
|
|
|
|
call->send_extra_metadata_count = 0; |
|
|
|
|
} |
|
|
|
@ -466,10 +414,10 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, |
|
|
|
|
gpr_mu_unlock(&pc->child_list_mu); |
|
|
|
|
} |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
cancel_with_error(call, STATUS_FROM_SURFACE, GRPC_ERROR_REF(error)); |
|
|
|
|
cancel_with_error(call, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
if (immediately_cancel) { |
|
|
|
|
cancel_with_error(call, STATUS_FROM_API_OVERRIDE, GRPC_ERROR_CANCELLED); |
|
|
|
|
cancel_with_error(call, GRPC_ERROR_CANCELLED); |
|
|
|
|
} |
|
|
|
|
if (args->cq != nullptr) { |
|
|
|
|
GPR_ASSERT(args->pollset_set_alternative == nullptr && |
|
|
|
@ -496,7 +444,7 @@ grpc_error* grpc_call_create(const grpc_call_create_args* args, |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
grpc_core::channelz::ServerNode* channelz_server = |
|
|
|
|
grpc_server_get_channelz_node(call->server); |
|
|
|
|
grpc_server_get_channelz_node(call->final_op.server.server); |
|
|
|
|
if (channelz_server != nullptr) { |
|
|
|
|
channelz_server->RecordCallStarted(); |
|
|
|
|
} |
|
|
|
@ -571,16 +519,13 @@ static void destroy_call(void* call, grpc_error* error) { |
|
|
|
|
GRPC_CQ_INTERNAL_UNREF(c->cq, "bind"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
get_final_status(c, set_status_value_directly, &c->final_info.final_status, |
|
|
|
|
nullptr, &(c->final_info.error_string)); |
|
|
|
|
grpc_error_get_status(c->status_error, c->send_deadline, |
|
|
|
|
&c->final_info.final_status, nullptr, nullptr, |
|
|
|
|
&(c->final_info.error_string)); |
|
|
|
|
GRPC_ERROR_UNREF(c->status_error); |
|
|
|
|
c->final_info.stats.latency = |
|
|
|
|
gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), c->start_time); |
|
|
|
|
|
|
|
|
|
for (i = 0; i < STATUS_SOURCE_COUNT; i++) { |
|
|
|
|
GRPC_ERROR_UNREF( |
|
|
|
|
unpack_received_status(gpr_atm_acq_load(&c->status[i])).error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_call_stack_destroy(CALL_STACK_FROM_CALL(c), &c->final_info, |
|
|
|
|
GRPC_CLOSURE_INIT(&c->release_call, release_call, c, |
|
|
|
|
grpc_schedule_on_exec_ctx)); |
|
|
|
@ -618,7 +563,7 @@ void grpc_call_unref(grpc_call* c) { |
|
|
|
|
bool cancel = gpr_atm_acq_load(&c->any_ops_sent_atm) != 0 && |
|
|
|
|
gpr_atm_acq_load(&c->received_final_op_atm) == 0; |
|
|
|
|
if (cancel) { |
|
|
|
|
cancel_with_error(c, STATUS_FROM_API_OVERRIDE, GRPC_ERROR_CANCELLED); |
|
|
|
|
cancel_with_error(c, GRPC_ERROR_CANCELLED); |
|
|
|
|
} else { |
|
|
|
|
// Unset the call combiner cancellation closure. This has the
|
|
|
|
|
// effect of scheduling the previously set cancellation closure, if
|
|
|
|
@ -636,8 +581,7 @@ 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); |
|
|
|
|
grpc_core::ExecCtx exec_ctx; |
|
|
|
|
cancel_with_error(call, STATUS_FROM_API_OVERRIDE, GRPC_ERROR_CANCELLED); |
|
|
|
|
|
|
|
|
|
cancel_with_error(call, GRPC_ERROR_CANCELLED); |
|
|
|
|
return GRPC_CALL_OK; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -691,8 +635,7 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call* c, |
|
|
|
|
"c=%p, status=%d, description=%s, reserved=%p)", |
|
|
|
|
4, (c, (int)status, description, reserved)); |
|
|
|
|
GPR_ASSERT(reserved == nullptr); |
|
|
|
|
cancel_with_status(c, STATUS_FROM_API_OVERRIDE, status, description); |
|
|
|
|
|
|
|
|
|
cancel_with_status(c, status, description); |
|
|
|
|
return GRPC_CALL_OK; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -712,15 +655,17 @@ static void done_termination(void* arg, grpc_error* error) { |
|
|
|
|
gpr_free(state); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void cancel_with_error(grpc_call* c, status_source source, |
|
|
|
|
grpc_error* error) { |
|
|
|
|
static void cancel_with_error(grpc_call* c, grpc_error* error) { |
|
|
|
|
if (!gpr_atm_rel_cas(&c->cancelled, 0, 1)) { |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
GRPC_CALL_INTERNAL_REF(c, "termination"); |
|
|
|
|
// Inform the call combiner of the cancellation, so that it can cancel
|
|
|
|
|
// any in-flight asynchronous actions that may be holding the call
|
|
|
|
|
// combiner. This ensures that the cancel_stream batch can be sent
|
|
|
|
|
// down the filter stack in a timely manner.
|
|
|
|
|
grpc_call_combiner_cancel(&c->call_combiner, GRPC_ERROR_REF(error)); |
|
|
|
|
set_status_from_error(c, source, GRPC_ERROR_REF(error)); |
|
|
|
|
cancel_state* state = static_cast<cancel_state*>(gpr_malloc(sizeof(*state))); |
|
|
|
|
state->call = c; |
|
|
|
|
GRPC_CLOSURE_INIT(&state->finish_batch, done_termination, state, |
|
|
|
@ -743,90 +688,45 @@ static grpc_error* error_from_status(grpc_status_code status, |
|
|
|
|
GRPC_ERROR_INT_GRPC_STATUS, status); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void cancel_with_status(grpc_call* c, status_source source, |
|
|
|
|
grpc_status_code status, |
|
|
|
|
static void cancel_with_status(grpc_call* c, grpc_status_code status, |
|
|
|
|
const char* description) { |
|
|
|
|
cancel_with_error(c, source, error_from_status(status, description)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/*******************************************************************************
|
|
|
|
|
* FINAL STATUS CODE MANIPULATION |
|
|
|
|
*/ |
|
|
|
|
|
|
|
|
|
static bool get_final_status_from( |
|
|
|
|
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, const char** error_string) { |
|
|
|
|
grpc_status_code code; |
|
|
|
|
grpc_slice slice = grpc_empty_slice(); |
|
|
|
|
grpc_error_get_status(error, call->send_deadline, &code, &slice, nullptr, |
|
|
|
|
error_string); |
|
|
|
|
if (code == GRPC_STATUS_OK && !allow_ok_status) { |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
set_value(code, set_value_user_data); |
|
|
|
|
if (details != nullptr) { |
|
|
|
|
*details = grpc_slice_ref_internal(slice); |
|
|
|
|
} |
|
|
|
|
return true; |
|
|
|
|
cancel_with_error(c, error_from_status(status, description)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void get_final_status( |
|
|
|
|
grpc_call* call, void (*set_value)(grpc_status_code code, void* user_data), |
|
|
|
|
void* set_value_user_data, grpc_slice* details, const char** error_string) { |
|
|
|
|
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])); |
|
|
|
|
} |
|
|
|
|
static void set_final_status(grpc_call* call, grpc_error* error) { |
|
|
|
|
if (grpc_call_error_trace.enabled()) { |
|
|
|
|
gpr_log(GPR_INFO, "get_final_status %s", call->is_client ? "CLI" : "SVR"); |
|
|
|
|
for (i = 0; i < STATUS_SOURCE_COUNT; i++) { |
|
|
|
|
if (status[i].is_set) { |
|
|
|
|
gpr_log(GPR_INFO, " %d: %s", i, grpc_error_string(status[i].error)); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
gpr_log(GPR_DEBUG, "set_final_status %s", call->is_client ? "CLI" : "SVR"); |
|
|
|
|
gpr_log(GPR_DEBUG, "%s", grpc_error_string(error)); |
|
|
|
|
} |
|
|
|
|
/* first search through ignoring "OK" statuses: if something went wrong,
|
|
|
|
|
* ensure we report it */ |
|
|
|
|
for (int allow_ok_status = 0; allow_ok_status < 2; allow_ok_status++) { |
|
|
|
|
/* 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 (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, |
|
|
|
|
error_string)) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
if (call->is_client) { |
|
|
|
|
grpc_error_get_status(error, call->send_deadline, |
|
|
|
|
call->final_op.client.status, |
|
|
|
|
call->final_op.client.status_details, nullptr, |
|
|
|
|
call->final_op.client.error_string); |
|
|
|
|
// explicitly take a ref
|
|
|
|
|
grpc_slice_ref_internal(*call->final_op.client.status_details); |
|
|
|
|
call->status_error = error; |
|
|
|
|
grpc_core::channelz::ChannelNode* channelz_channel = |
|
|
|
|
grpc_channel_get_channelz_node(call->channel); |
|
|
|
|
if (channelz_channel != nullptr) { |
|
|
|
|
if (*call->final_op.client.status != GRPC_STATUS_OK) { |
|
|
|
|
channelz_channel->RecordCallFailed(); |
|
|
|
|
} else { |
|
|
|
|
channelz_channel->RecordCallSucceeded(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
/* If no clearly defined status exists, search for 'anything' */ |
|
|
|
|
for (i = 0; i < STATUS_SOURCE_COUNT; i++) { |
|
|
|
|
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, |
|
|
|
|
error_string)) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
*call->final_op.server.cancelled = |
|
|
|
|
error != GRPC_ERROR_NONE || call->status_error != GRPC_ERROR_NONE; |
|
|
|
|
grpc_core::channelz::ServerNode* channelz_server = |
|
|
|
|
grpc_server_get_channelz_node(call->final_op.server.server); |
|
|
|
|
if (channelz_server != nullptr) { |
|
|
|
|
if (*call->final_op.server.cancelled) { |
|
|
|
|
channelz_server->RecordCallFailed(); |
|
|
|
|
} else { |
|
|
|
|
channelz_server->RecordCallSucceeded(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
/* If nothing exists, set some default */ |
|
|
|
|
if (call->is_client) { |
|
|
|
|
set_value(GRPC_STATUS_UNKNOWN, set_value_user_data); |
|
|
|
|
} else { |
|
|
|
|
set_value(GRPC_STATUS_OK, set_value_user_data); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void set_status_from_error(grpc_call* call, status_source source, |
|
|
|
|
grpc_error* error) { |
|
|
|
|
if (!gpr_atm_rel_cas(&call->status[source], |
|
|
|
|
pack_received_status({false, GRPC_ERROR_NONE}), |
|
|
|
|
pack_received_status({true, error}))) { |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
@ -1045,6 +945,7 @@ static grpc_stream_compression_algorithm decode_stream_compression( |
|
|
|
|
static void publish_app_metadata(grpc_call* call, grpc_metadata_batch* b, |
|
|
|
|
int is_trailing) { |
|
|
|
|
if (b->list.count == 0) return; |
|
|
|
|
if (!call->is_client && is_trailing) return; |
|
|
|
|
if (is_trailing && call->buffered_metadata[1] == nullptr) return; |
|
|
|
|
GPR_TIMER_SCOPE("publish_app_metadata", 0); |
|
|
|
|
grpc_metadata_array* dest; |
|
|
|
@ -1098,9 +999,12 @@ static void recv_initial_filter(grpc_call* call, grpc_metadata_batch* b) { |
|
|
|
|
publish_app_metadata(call, b, false); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void recv_trailing_filter(void* args, grpc_metadata_batch* b) { |
|
|
|
|
static void recv_trailing_filter(void* args, grpc_metadata_batch* b, |
|
|
|
|
grpc_error* batch_error) { |
|
|
|
|
grpc_call* call = static_cast<grpc_call*>(args); |
|
|
|
|
if (b->idx.named.grpc_status != nullptr) { |
|
|
|
|
if (batch_error != GRPC_ERROR_NONE) { |
|
|
|
|
set_final_status(call, batch_error); |
|
|
|
|
} else if (b->idx.named.grpc_status != nullptr) { |
|
|
|
|
grpc_status_code status_code = |
|
|
|
|
grpc_get_status_code_from_metadata(b->idx.named.grpc_status->md); |
|
|
|
|
grpc_error* error = GRPC_ERROR_NONE; |
|
|
|
@ -1118,8 +1022,18 @@ static void recv_trailing_filter(void* args, grpc_metadata_batch* b) { |
|
|
|
|
error = grpc_error_set_str(error, GRPC_ERROR_STR_GRPC_MESSAGE, |
|
|
|
|
grpc_empty_slice()); |
|
|
|
|
} |
|
|
|
|
set_status_from_error(call, STATUS_FROM_WIRE, error); |
|
|
|
|
set_final_status(call, GRPC_ERROR_REF(error)); |
|
|
|
|
grpc_metadata_batch_remove(b, b->idx.named.grpc_status); |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
} else if (!call->is_client) { |
|
|
|
|
set_final_status(call, GRPC_ERROR_NONE); |
|
|
|
|
} else { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"Received trailing metadata with no error and no status"); |
|
|
|
|
set_final_status( |
|
|
|
|
call, grpc_error_set_int( |
|
|
|
|
GRPC_ERROR_CREATE_FROM_STATIC_STRING("No status received"), |
|
|
|
|
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNKNOWN)); |
|
|
|
|
} |
|
|
|
|
publish_app_metadata(call, b, true); |
|
|
|
|
} |
|
|
|
@ -1134,14 +1048,6 @@ grpc_call_stack* grpc_call_get_call_stack(grpc_call* call) { |
|
|
|
|
* BATCH API IMPLEMENTATION |
|
|
|
|
*/ |
|
|
|
|
|
|
|
|
|
static void set_status_value_directly(grpc_status_code status, void* dest) { |
|
|
|
|
*static_cast<grpc_status_code*>(dest) = status; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void set_cancelled_value(grpc_status_code status, void* dest) { |
|
|
|
|
*static_cast<int*>(dest) = (status != GRPC_STATUS_OK); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static bool are_write_flags_valid(uint32_t flags) { |
|
|
|
|
/* check that only bits in GRPC_WRITE_(INTERNAL?)_USED_MASK are set */ |
|
|
|
|
const uint32_t allowed_write_positions = |
|
|
|
@ -1209,31 +1115,15 @@ static void finish_batch_completion(void* user_data, |
|
|
|
|
GRPC_CALL_INTERNAL_UNREF(call, "completion"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static grpc_error* consolidate_batch_errors(batch_control* bctl) { |
|
|
|
|
size_t n = static_cast<size_t>(gpr_atm_acq_load(&bctl->num_errors)); |
|
|
|
|
if (n == 0) { |
|
|
|
|
return GRPC_ERROR_NONE; |
|
|
|
|
} else if (n == 1) { |
|
|
|
|
/* Skip creating a composite error in the case that only one error was
|
|
|
|
|
logged */ |
|
|
|
|
grpc_error* e = bctl->errors[0]; |
|
|
|
|
bctl->errors[0] = nullptr; |
|
|
|
|
return e; |
|
|
|
|
} else { |
|
|
|
|
grpc_error* error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( |
|
|
|
|
"Call batch failed", bctl->errors, n); |
|
|
|
|
for (size_t i = 0; i < n; i++) { |
|
|
|
|
GRPC_ERROR_UNREF(bctl->errors[i]); |
|
|
|
|
bctl->errors[i] = nullptr; |
|
|
|
|
} |
|
|
|
|
return error; |
|
|
|
|
} |
|
|
|
|
static void reset_batch_errors(batch_control* bctl) { |
|
|
|
|
GRPC_ERROR_UNREF(bctl->batch_error); |
|
|
|
|
bctl->batch_error = GRPC_ERROR_NONE; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void post_batch_completion(batch_control* bctl) { |
|
|
|
|
grpc_call* next_child_call; |
|
|
|
|
grpc_call* call = bctl->call; |
|
|
|
|
grpc_error* error = consolidate_batch_errors(bctl); |
|
|
|
|
grpc_error* error = GRPC_ERROR_REF(bctl->batch_error); |
|
|
|
|
|
|
|
|
|
if (bctl->op.send_initial_metadata) { |
|
|
|
|
grpc_metadata_batch_destroy( |
|
|
|
@ -1259,8 +1149,7 @@ static void post_batch_completion(batch_control* bctl) { |
|
|
|
|
next_child_call = child->child->sibling_next; |
|
|
|
|
if (child->cancellation_is_inherited) { |
|
|
|
|
GRPC_CALL_INTERNAL_REF(child, "propagate_cancel"); |
|
|
|
|
cancel_with_error(child, STATUS_FROM_API_OVERRIDE, |
|
|
|
|
GRPC_ERROR_CANCELLED); |
|
|
|
|
cancel_with_error(child, GRPC_ERROR_CANCELLED); |
|
|
|
|
GRPC_CALL_INTERNAL_UNREF(child, "propagate_cancel"); |
|
|
|
|
} |
|
|
|
|
child = next_child_call; |
|
|
|
@ -1268,33 +1157,6 @@ static void post_batch_completion(batch_control* bctl) { |
|
|
|
|
} |
|
|
|
|
gpr_mu_unlock(&pc->child_list_mu); |
|
|
|
|
} |
|
|
|
|
if (call->is_client) { |
|
|
|
|
get_final_status(call, set_status_value_directly, |
|
|
|
|
call->final_op.client.status, |
|
|
|
|
call->final_op.client.status_details, |
|
|
|
|
call->final_op.client.error_string); |
|
|
|
|
grpc_core::channelz::ChannelNode* channelz_channel = |
|
|
|
|
grpc_channel_get_channelz_node(call->channel); |
|
|
|
|
if (channelz_channel != nullptr) { |
|
|
|
|
if (*call->final_op.client.status != GRPC_STATUS_OK) { |
|
|
|
|
channelz_channel->RecordCallFailed(); |
|
|
|
|
} else { |
|
|
|
|
channelz_channel->RecordCallSucceeded(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
get_final_status(call, set_cancelled_value, |
|
|
|
|
call->final_op.server.cancelled, nullptr, nullptr); |
|
|
|
|
grpc_core::channelz::ServerNode* channelz_server = |
|
|
|
|
grpc_server_get_channelz_node(call->server); |
|
|
|
|
if (channelz_server != nullptr) { |
|
|
|
|
if (*call->final_op.server.cancelled) { |
|
|
|
|
channelz_server->RecordCallFailed(); |
|
|
|
|
} else { |
|
|
|
|
channelz_server->RecordCallSucceeded(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
error = GRPC_ERROR_NONE; |
|
|
|
|
} |
|
|
|
@ -1303,9 +1165,10 @@ static void post_batch_completion(batch_control* bctl) { |
|
|
|
|
grpc_byte_buffer_destroy(*call->receiving_buffer); |
|
|
|
|
*call->receiving_buffer = nullptr; |
|
|
|
|
} |
|
|
|
|
reset_batch_errors(bctl); |
|
|
|
|
|
|
|
|
|
if (bctl->completion_data.notify_tag.is_closure) { |
|
|
|
|
/* unrefs bctl->error */ |
|
|
|
|
/* unrefs error */ |
|
|
|
|
bctl->call = nullptr; |
|
|
|
|
/* This closure may be meant to be run within some combiner. Since we aren't
|
|
|
|
|
* running in any combiner here, we need to use GRPC_CLOSURE_SCHED instead |
|
|
|
@ -1315,7 +1178,7 @@ static void post_batch_completion(batch_control* bctl) { |
|
|
|
|
error); |
|
|
|
|
GRPC_CALL_INTERNAL_UNREF(call, "completion"); |
|
|
|
|
} else { |
|
|
|
|
/* unrefs bctl->error */ |
|
|
|
|
/* unrefs error */ |
|
|
|
|
grpc_cq_end_op(bctl->call->cq, bctl->completion_data.notify_tag.tag, error, |
|
|
|
|
finish_batch_completion, bctl, |
|
|
|
|
&bctl->completion_data.cq_completion); |
|
|
|
@ -1424,8 +1287,10 @@ static void receiving_stream_ready(void* bctlp, grpc_error* error) { |
|
|
|
|
grpc_call* call = bctl->call; |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
call->receiving_stream.reset(); |
|
|
|
|
add_batch_error(bctl, GRPC_ERROR_REF(error), true); |
|
|
|
|
cancel_with_error(call, STATUS_FROM_SURFACE, GRPC_ERROR_REF(error)); |
|
|
|
|
if (bctl->batch_error == GRPC_ERROR_NONE) { |
|
|
|
|
bctl->batch_error = GRPC_ERROR_REF(error); |
|
|
|
|
} |
|
|
|
|
cancel_with_error(call, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
/* If recv_state is RECV_NONE, we will save the batch_control
|
|
|
|
|
* object with rel_cas, and will not use it after the cas. Its corresponding |
|
|
|
@ -1443,7 +1308,7 @@ static void receiving_stream_ready_in_call_combiner(void* bctlp, |
|
|
|
|
grpc_error* error) { |
|
|
|
|
batch_control* bctl = static_cast<batch_control*>(bctlp); |
|
|
|
|
grpc_call* call = bctl->call; |
|
|
|
|
GRPC_CALL_COMBINER_STOP(&call->call_combiner, "call_recv_message_ready"); |
|
|
|
|
GRPC_CALL_COMBINER_STOP(&call->call_combiner, "recv_message_ready"); |
|
|
|
|
receiving_stream_ready(bctlp, error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -1461,8 +1326,7 @@ static void validate_filtered_metadata(batch_control* bctl) { |
|
|
|
|
call->incoming_stream_compression_algorithm, |
|
|
|
|
call->incoming_message_compression_algorithm); |
|
|
|
|
gpr_log(GPR_ERROR, "%s", error_msg); |
|
|
|
|
cancel_with_status(call, STATUS_FROM_SURFACE, GRPC_STATUS_INTERNAL, |
|
|
|
|
error_msg); |
|
|
|
|
cancel_with_status(call, GRPC_STATUS_INTERNAL, error_msg); |
|
|
|
|
gpr_free(error_msg); |
|
|
|
|
} else if ( |
|
|
|
|
grpc_compression_algorithm_from_message_stream_compression_algorithm( |
|
|
|
@ -1474,8 +1338,7 @@ static void validate_filtered_metadata(batch_control* bctl) { |
|
|
|
|
"compression (%d).", |
|
|
|
|
call->incoming_stream_compression_algorithm, |
|
|
|
|
call->incoming_message_compression_algorithm); |
|
|
|
|
cancel_with_status(call, STATUS_FROM_SURFACE, GRPC_STATUS_INTERNAL, |
|
|
|
|
error_msg); |
|
|
|
|
cancel_with_status(call, GRPC_STATUS_INTERNAL, error_msg); |
|
|
|
|
gpr_free(error_msg); |
|
|
|
|
} else { |
|
|
|
|
char* error_msg = nullptr; |
|
|
|
@ -1485,8 +1348,7 @@ static void validate_filtered_metadata(batch_control* bctl) { |
|
|
|
|
gpr_asprintf(&error_msg, "Invalid compression algorithm value '%d'.", |
|
|
|
|
compression_algorithm); |
|
|
|
|
gpr_log(GPR_ERROR, "%s", error_msg); |
|
|
|
|
cancel_with_status(call, STATUS_FROM_SURFACE, GRPC_STATUS_UNIMPLEMENTED, |
|
|
|
|
error_msg); |
|
|
|
|
cancel_with_status(call, GRPC_STATUS_UNIMPLEMENTED, error_msg); |
|
|
|
|
} else if (grpc_compression_options_is_algorithm_enabled( |
|
|
|
|
&compression_options, compression_algorithm) == 0) { |
|
|
|
|
/* check if algorithm is supported by current channel config */ |
|
|
|
@ -1495,8 +1357,7 @@ static void validate_filtered_metadata(batch_control* bctl) { |
|
|
|
|
gpr_asprintf(&error_msg, "Compression algorithm '%s' is disabled.", |
|
|
|
|
algo_name); |
|
|
|
|
gpr_log(GPR_ERROR, "%s", error_msg); |
|
|
|
|
cancel_with_status(call, STATUS_FROM_SURFACE, GRPC_STATUS_UNIMPLEMENTED, |
|
|
|
|
error_msg); |
|
|
|
|
cancel_with_status(call, GRPC_STATUS_UNIMPLEMENTED, error_msg); |
|
|
|
|
} |
|
|
|
|
gpr_free(error_msg); |
|
|
|
|
|
|
|
|
@ -1514,24 +1375,12 @@ static void validate_filtered_metadata(batch_control* bctl) { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void add_batch_error(batch_control* bctl, grpc_error* error, |
|
|
|
|
bool has_cancelled) { |
|
|
|
|
if (error == GRPC_ERROR_NONE) return; |
|
|
|
|
int idx = static_cast<int>(gpr_atm_full_fetch_add(&bctl->num_errors, 1)); |
|
|
|
|
if (idx == 0 && !has_cancelled) { |
|
|
|
|
cancel_with_error(bctl->call, STATUS_FROM_CORE, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
bctl->errors[idx] = error; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void receiving_initial_metadata_ready(void* bctlp, grpc_error* error) { |
|
|
|
|
batch_control* bctl = static_cast<batch_control*>(bctlp); |
|
|
|
|
grpc_call* call = bctl->call; |
|
|
|
|
|
|
|
|
|
GRPC_CALL_COMBINER_STOP(&call->call_combiner, |
|
|
|
|
"call_recv_initial_metadata_ready"); |
|
|
|
|
GRPC_CALL_COMBINER_STOP(&call->call_combiner, "recv_initial_metadata_ready"); |
|
|
|
|
|
|
|
|
|
add_batch_error(bctl, GRPC_ERROR_REF(error), false); |
|
|
|
|
if (error == GRPC_ERROR_NONE) { |
|
|
|
|
grpc_metadata_batch* md = |
|
|
|
|
&call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */]; |
|
|
|
@ -1544,6 +1393,11 @@ static void receiving_initial_metadata_ready(void* bctlp, grpc_error* error) { |
|
|
|
|
if (md->deadline != GRPC_MILLIS_INF_FUTURE && !call->is_client) { |
|
|
|
|
call->send_deadline = md->deadline; |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
if (bctl->batch_error == GRPC_ERROR_NONE) { |
|
|
|
|
bctl->batch_error = GRPC_ERROR_REF(error); |
|
|
|
|
} |
|
|
|
|
cancel_with_error(call, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_closure* saved_rsr_closure = nullptr; |
|
|
|
@ -1580,20 +1434,23 @@ static void receiving_initial_metadata_ready(void* bctlp, grpc_error* error) { |
|
|
|
|
static void receiving_trailing_metadata_ready(void* bctlp, grpc_error* error) { |
|
|
|
|
batch_control* bctl = static_cast<batch_control*>(bctlp); |
|
|
|
|
grpc_call* call = bctl->call; |
|
|
|
|
GRPC_CALL_COMBINER_STOP(&call->call_combiner, |
|
|
|
|
"call_recv_trailing_metadata_ready"); |
|
|
|
|
add_batch_error(bctl, GRPC_ERROR_REF(error), false); |
|
|
|
|
GRPC_CALL_COMBINER_STOP(&call->call_combiner, "recv_trailing_metadata_ready"); |
|
|
|
|
grpc_metadata_batch* md = |
|
|
|
|
&call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; |
|
|
|
|
recv_trailing_filter(call, md); |
|
|
|
|
recv_trailing_filter(call, md, GRPC_ERROR_REF(error)); |
|
|
|
|
finish_batch_step(bctl); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void finish_batch(void* bctlp, grpc_error* error) { |
|
|
|
|
batch_control* bctl = static_cast<batch_control*>(bctlp); |
|
|
|
|
grpc_call* call = bctl->call; |
|
|
|
|
GRPC_CALL_COMBINER_STOP(&call->call_combiner, "call_on_complete"); |
|
|
|
|
add_batch_error(bctl, GRPC_ERROR_REF(error), false); |
|
|
|
|
GRPC_CALL_COMBINER_STOP(&call->call_combiner, "on_complete"); |
|
|
|
|
if (bctl->batch_error == GRPC_ERROR_NONE) { |
|
|
|
|
bctl->batch_error = GRPC_ERROR_REF(error); |
|
|
|
|
} |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
cancel_with_error(call, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
finish_batch_step(bctl); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -1795,28 +1652,32 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, |
|
|
|
|
call->send_extra_metadata_count = 1; |
|
|
|
|
call->send_extra_metadata[0].md = grpc_channel_get_reffed_status_elem( |
|
|
|
|
call->channel, op->data.send_status_from_server.status); |
|
|
|
|
{ |
|
|
|
|
grpc_error* override_error = GRPC_ERROR_NONE; |
|
|
|
|
if (op->data.send_status_from_server.status != GRPC_STATUS_OK) { |
|
|
|
|
override_error = |
|
|
|
|
error_from_status(op->data.send_status_from_server.status, |
|
|
|
|
"Returned non-ok status"); |
|
|
|
|
} |
|
|
|
|
if (op->data.send_status_from_server.status_details != nullptr) { |
|
|
|
|
call->send_extra_metadata[1].md = grpc_mdelem_from_slices( |
|
|
|
|
GRPC_MDSTR_GRPC_MESSAGE, |
|
|
|
|
grpc_slice_ref_internal( |
|
|
|
|
*op->data.send_status_from_server.status_details)); |
|
|
|
|
call->send_extra_metadata_count++; |
|
|
|
|
grpc_error* status_error = |
|
|
|
|
op->data.send_status_from_server.status == GRPC_STATUS_OK |
|
|
|
|
? GRPC_ERROR_NONE |
|
|
|
|
: grpc_error_set_int( |
|
|
|
|
GRPC_ERROR_CREATE_FROM_STATIC_STRING( |
|
|
|
|
"Server returned error"), |
|
|
|
|
GRPC_ERROR_INT_GRPC_STATUS, |
|
|
|
|
static_cast<intptr_t>( |
|
|
|
|
op->data.send_status_from_server.status)); |
|
|
|
|
if (op->data.send_status_from_server.status_details != nullptr) { |
|
|
|
|
call->send_extra_metadata[1].md = grpc_mdelem_from_slices( |
|
|
|
|
GRPC_MDSTR_GRPC_MESSAGE, |
|
|
|
|
grpc_slice_ref_internal( |
|
|
|
|
*op->data.send_status_from_server.status_details)); |
|
|
|
|
call->send_extra_metadata_count++; |
|
|
|
|
if (status_error != GRPC_ERROR_NONE) { |
|
|
|
|
char* msg = grpc_slice_to_c_string( |
|
|
|
|
GRPC_MDVALUE(call->send_extra_metadata[1].md)); |
|
|
|
|
override_error = |
|
|
|
|
grpc_error_set_str(override_error, GRPC_ERROR_STR_GRPC_MESSAGE, |
|
|
|
|
status_error = |
|
|
|
|
grpc_error_set_str(status_error, GRPC_ERROR_STR_GRPC_MESSAGE, |
|
|
|
|
grpc_slice_from_copied_string(msg)); |
|
|
|
|
gpr_free(msg); |
|
|
|
|
} |
|
|
|
|
set_status_from_error(call, STATUS_FROM_API_OVERRIDE, override_error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
call->status_error = status_error; |
|
|
|
|
if (!prepare_application_metadata( |
|
|
|
|
call, |
|
|
|
|
static_cast<int>( |
|
|
|
|