Chiebot-Mirror
/
grpc
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix race detected by TSAN

Browse Source
pull/9529/head
Craig Tiller 8 years ago
parent 7b005f7fdd
commit 2dc32eabcb
1 changed files with 37 additions and 44 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 81
      src/core/lib/surface/call.c

81
src/core/lib/surface/call.c
Unescape View File

@ -86,8 +86,11 @@ typedef enum {
/* Status came from 'the wire' - or somewhere below the surface /* Status came from 'the wire' - or somewhere below the surface
layer */ layer */
STATUS_FROM_WIRE, STATUS_FROM_WIRE,
/* Status was created by some internal channel stack operation */ /* Status was created by some internal channel stack operation: must come via
add_batch_error */
STATUS_FROM_CORE, STATUS_FROM_CORE,
/* Status was created by some surface error */
STATUS_FROM_SURFACE,
/* Status came from the server sending status */ /* Status came from the server sending status */
STATUS_FROM_SERVER_STATUS, STATUS_FROM_SERVER_STATUS,
STATUS_SOURCE_COUNT STATUS_SOURCE_COUNT
@ -221,11 +224,11 @@ struct grpc_call {
static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call, static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
grpc_transport_stream_op *op); grpc_transport_stream_op *op);
static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c, static void cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_status_code status, status_source source, grpc_status_code status,
const char *description); const char *description);
static void cancel_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c, static void cancel_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_error *error); status_source source, grpc_error *error);
static void destroy_call(grpc_exec_ctx *exec_ctx, void *call_stack, static void destroy_call(grpc_exec_ctx *exec_ctx, void *call_stack,
grpc_error *error); grpc_error *error);
static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp, static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
@ -337,7 +340,7 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
args->server_transport_data, path, call->start_time, send_deadline, args->server_transport_data, path, call->start_time, send_deadline,
CALL_STACK_FROM_CALL(call)); CALL_STACK_FROM_CALL(call));
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
cancel_with_error(exec_ctx, call, GRPC_ERROR_REF(error)); cancel_with_error(exec_ctx, call, STATUS_FROM_SURFACE, GRPC_ERROR_REF(error));
} }
if (args->cq != NULL) { if (args->cq != NULL) {
GPR_ASSERT( GPR_ASSERT(
@ -512,7 +515,6 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
grpc_status_code status, grpc_status_code status,
const char *description, const char *description,
void *reserved) { void *reserved) {
grpc_call_error r;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
GRPC_API_TRACE( GRPC_API_TRACE(
"grpc_call_cancel_with_status(" "grpc_call_cancel_with_status("
@ -520,16 +522,16 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
4, (c, (int)status, description, reserved)); 4, (c, (int)status, description, reserved));
GPR_ASSERT(reserved == NULL); GPR_ASSERT(reserved == NULL);
gpr_mu_lock(&c->mu); gpr_mu_lock(&c->mu);
r = cancel_with_status(&exec_ctx, c, status, description); cancel_with_status(&exec_ctx, c, STATUS_FROM_API_OVERRIDE, status,
description);
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
grpc_exec_ctx_finish(&exec_ctx); grpc_exec_ctx_finish(&exec_ctx);
return r; return GRPC_CALL_OK;
} }
typedef struct termination_closure { typedef struct termination_closure {
grpc_closure closure; grpc_closure closure;
grpc_call *call; grpc_call *call;
grpc_error *error;
grpc_transport_stream_op op; grpc_transport_stream_op op;
} termination_closure; } termination_closure;
@ -544,36 +546,27 @@ static void send_termination(grpc_exec_ctx *exec_ctx, void *tcp,
grpc_error *error) { grpc_error *error) {
termination_closure *tc = tcp; termination_closure *tc = tcp;
memset(&tc->op, 0, sizeof(tc->op)); memset(&tc->op, 0, sizeof(tc->op));
tc->op.cancel_error = tc->error; tc->op.cancel_error = GRPC_ERROR_REF(error);
/* reuse closure to catch completion */ /* reuse closure to catch completion */
grpc_closure_init(&tc->closure, done_termination, tc, tc->op.on_complete = grpc_closure_init(&tc->closure, done_termination, tc,
grpc_schedule_on_exec_ctx); grpc_schedule_on_exec_ctx);
tc->op.on_complete = &tc->closure;
execute_op(exec_ctx, tc->call, &tc->op); execute_op(exec_ctx, tc->call, &tc->op);
} }
static grpc_call_error terminate_with_status(grpc_exec_ctx *exec_ctx, static void terminate_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c,
termination_closure *tc) { grpc_error *error) {
set_status_from_error(exec_ctx, tc->call, STATUS_FROM_API_OVERRIDE,
GRPC_ERROR_REF(tc->error));
grpc_closure_init(&tc->closure, send_termination, tc,
grpc_schedule_on_exec_ctx);
GRPC_CALL_INTERNAL_REF(tc->call, "termination");
grpc_closure_sched(exec_ctx, &tc->closure, GRPC_ERROR_NONE);
return GRPC_CALL_OK;
}
static grpc_call_error terminate_with_error(grpc_exec_ctx *exec_ctx,
grpc_call *c, grpc_error *error) {
termination_closure *tc = gpr_malloc(sizeof(*tc)); termination_closure *tc = gpr_malloc(sizeof(*tc));
memset(tc, 0, sizeof(*tc)); memset(tc, 0, sizeof(*tc));
tc->call = c; tc->call = c;
tc->error = error; GRPC_CALL_INTERNAL_REF(tc->call, "termination");
return terminate_with_status(exec_ctx, tc); grpc_closure_sched(exec_ctx, grpc_closure_init(&tc->closure, send_termination,
tc, grpc_schedule_on_exec_ctx),
error);
} }
static void cancel_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c, static void cancel_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_error *error) { status_source source, grpc_error *error) {
set_status_from_error(exec_ctx, c, source, GRPC_ERROR_REF(error));
terminate_with_error(exec_ctx, c, error); terminate_with_error(exec_ctx, c, error);
} }
@ -585,11 +578,11 @@ static grpc_error *error_from_status(grpc_status_code status,
GRPC_ERROR_INT_GRPC_STATUS, status); GRPC_ERROR_INT_GRPC_STATUS, status);
} }
static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c, static void cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_status_code status, status_source source, grpc_status_code status,
const char *description) { const char *description) {
return terminate_with_error(exec_ctx, c, cancel_with_error(exec_ctx, c, source,
error_from_status(status, description)); error_from_status(status, description));
} }
/******************************************************************************* /*******************************************************************************
@ -1014,11 +1007,6 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
gpr_mu_lock(&call->mu); gpr_mu_lock(&call->mu);
if (error != GRPC_ERROR_NONE) {
set_status_from_error(exec_ctx, call, STATUS_FROM_CORE,
GRPC_ERROR_REF(error));
}
if (bctl->send_initial_metadata) { if (bctl->send_initial_metadata) {
grpc_metadata_batch_destroy( grpc_metadata_batch_destroy(
exec_ctx, exec_ctx,
@ -1161,7 +1149,7 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
grpc_call *call = bctl->call; grpc_call *call = bctl->call;
gpr_mu_lock(&bctl->call->mu); gpr_mu_lock(&bctl->call->mu);
if (error != GRPC_ERROR_NONE) { if (error != GRPC_ERROR_NONE) {
cancel_with_error(exec_ctx, call, GRPC_ERROR_REF(error)); cancel_with_error(exec_ctx, call, STATUS_FROM_SURFACE, GRPC_ERROR_REF(error));
} }
if (call->has_initial_md_been_received || error != GRPC_ERROR_NONE || if (call->has_initial_md_been_received || error != GRPC_ERROR_NONE ||
call->receiving_stream == NULL) { call->receiving_stream == NULL) {
@ -1188,7 +1176,8 @@ static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx,
gpr_asprintf(&error_msg, "Invalid compression algorithm value '%d'.", gpr_asprintf(&error_msg, "Invalid compression algorithm value '%d'.",
algo); algo);
gpr_log(GPR_ERROR, "%s", error_msg); gpr_log(GPR_ERROR, "%s", error_msg);
cancel_with_status(exec_ctx, call, GRPC_STATUS_UNIMPLEMENTED, error_msg); cancel_with_status(exec_ctx, call, STATUS_FROM_SURFACE,
GRPC_STATUS_UNIMPLEMENTED, error_msg);
} else if (grpc_compression_options_is_algorithm_enabled( } else if (grpc_compression_options_is_algorithm_enabled(
&compression_options, algo) == 0) { &compression_options, algo) == 0) {
/* check if algorithm is supported by current channel config */ /* check if algorithm is supported by current channel config */
@ -1197,7 +1186,8 @@ static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx,
gpr_asprintf(&error_msg, "Compression algorithm '%s' is disabled.", gpr_asprintf(&error_msg, "Compression algorithm '%s' is disabled.",
algo_name); algo_name);
gpr_log(GPR_ERROR, "%s", error_msg); gpr_log(GPR_ERROR, "%s", error_msg);
cancel_with_status(exec_ctx, call, GRPC_STATUS_UNIMPLEMENTED, error_msg); cancel_with_status(exec_ctx, call, STATUS_FROM_SURFACE,
GRPC_STATUS_UNIMPLEMENTED, error_msg);
} else { } else {
call->incoming_compression_algorithm = algo; call->incoming_compression_algorithm = algo;
} }
@ -1227,7 +1217,10 @@ static void add_batch_error(grpc_exec_ctx *exec_ctx, batch_control *bctl,
grpc_error *error) { grpc_error *error) {
if (error == GRPC_ERROR_NONE) return; if (error == GRPC_ERROR_NONE) return;
int idx = (int)gpr_atm_no_barrier_fetch_add(&bctl->num_errors, 1); int idx = (int)gpr_atm_no_barrier_fetch_add(&bctl->num_errors, 1);
if (idx == 0) cancel_with_error(exec_ctx, bctl->call, GRPC_ERROR_REF(error)); if (idx == 0) {
cancel_with_error(exec_ctx, bctl->call, STATUS_FROM_CORE,
GRPC_ERROR_REF(error));
}
bctl->errors[idx] = error; bctl->errors[idx] = error;
} }

Write Preview
Loading…
Cancel
Save