diff options
Diffstat (limited to 'src/core/surface/call.c')
| -rw-r--r-- | src/core/surface/call.c | 115 |
1 files changed, 109 insertions, 6 deletions
diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 327a096ffb..5839d3ac2e 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -143,6 +143,8 @@ typedef enum { struct grpc_call { grpc_completion_queue *cq; grpc_channel *channel; + grpc_call *parent; + grpc_call *first_child; grpc_mdctx *metadata_context; /* TODO(ctiller): share with cq if possible? */ gpr_mu mu; @@ -176,6 +178,8 @@ struct grpc_call { gpr_uint8 cancel_alarm; /** bitmask of allocated completion events in completions */ gpr_uint8 allocated_completions; + /** flag indicating that cancellation is inherited */ + gpr_uint8 cancellation_is_inherited; /* flags with bits corresponding to write states allowing us to determine what was sent */ @@ -267,6 +271,11 @@ struct grpc_call { /** completion events - for completion queue use */ grpc_cq_completion completions[MAX_CONCURRENT_COMPLETIONS]; + + /** siblings: children of the same parent form a list, and this list is protected under + parent->mu */ + grpc_call *sibling_next; + grpc_call *sibling_prev; }; #define CALL_STACK_FROM_CALL(call) ((grpc_call_stack *)((call) + 1)) @@ -290,7 +299,9 @@ static void finished_loose_op(void *call, int success); static void lock(grpc_call *call); static void unlock(grpc_call *call); -grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, +grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call, + gpr_uint32 propagation_mask, + grpc_completion_queue *cq, const void *server_transport_data, grpc_mdelem **add_initial_metadata, size_t add_initial_metadata_count, @@ -306,9 +317,10 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, gpr_mu_init(&call->completion_mu); call->channel = channel; call->cq = cq; - if (cq) { + if (cq != NULL) { GRPC_CQ_INTERNAL_REF(cq, "bind"); } + call->parent = parent_call; call->is_client = server_transport_data == NULL; for (i = 0; i < GRPC_IOREQ_OP_COUNT; i++) { call->request_set[i] = REQSET_EMPTY; @@ -347,6 +359,46 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, } grpc_call_stack_init(channel_stack, server_transport_data, initial_op_ptr, CALL_STACK_FROM_CALL(call)); + if (parent_call != NULL) { + GRPC_CALL_INTERNAL_REF(parent_call, "child"); + GPR_ASSERT(call->is_client); + GPR_ASSERT(!parent_call->is_client); + + gpr_mu_lock(&parent_call->mu); + + if (propagation_mask & GRPC_PROPAGATE_DEADLINE) { + send_deadline = gpr_time_min( + gpr_convert_clock_type(send_deadline, + parent_call->send_deadline.clock_type), + parent_call->send_deadline); + } + /* for now GRPC_PROPAGATE_TRACING_CONTEXT *MUST* be passed with + * GRPC_PROPAGATE_STATS_CONTEXT */ + /* TODO(ctiller): This should change to use the appropriate census start_op + * call. */ + if (propagation_mask & GRPC_PROPAGATE_CENSUS_TRACING_CONTEXT) { + GPR_ASSERT(propagation_mask & GRPC_PROPAGATE_CENSUS_STATS_CONTEXT); + grpc_call_context_set(call, GRPC_CONTEXT_TRACING, + parent_call->context[GRPC_CONTEXT_TRACING].value, + NULL); + } else { + GPR_ASSERT(propagation_mask & GRPC_PROPAGATE_CENSUS_STATS_CONTEXT); + } + if (propagation_mask & GRPC_PROPAGATE_CANCELLATION) { + call->cancellation_is_inherited = 1; + } + + if (parent_call->first_child == NULL) { + parent_call->first_child = call; + call->sibling_next = call->sibling_prev = call; + } else { + call->sibling_next = parent_call->first_child; + call->sibling_prev = parent_call->first_child->sibling_prev; + call->sibling_next->sibling_prev = call->sibling_prev->sibling_next = call; + } + + gpr_mu_unlock(&parent_call->mu); + } if (gpr_time_cmp(send_deadline, gpr_inf_future(send_deadline.clock_type)) != 0) { set_deadline_alarm(call, send_deadline); @@ -870,6 +922,8 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) { static void call_on_done_recv(void *pc, int success) { grpc_call *call = pc; + grpc_call *child_call; + grpc_call *next_child_call; size_t i; GRPC_TIMER_BEGIN(GRPC_PTAG_CALL_ON_DONE_RECV, 0); lock(call); @@ -903,6 +957,19 @@ static void call_on_done_recv(void *pc, int success) { GPR_ASSERT(call->read_state <= READ_STATE_STREAM_CLOSED); call->read_state = READ_STATE_STREAM_CLOSED; call->cancel_alarm |= call->have_alarm; + /* propagate cancellation to any interested children */ + child_call = call->first_child; + if (child_call != NULL) { + do { + next_child_call = child_call->sibling_next; + if (child_call->cancellation_is_inherited) { + GRPC_CALL_INTERNAL_REF(child_call, "propagate_cancel"); + grpc_call_cancel(child_call); + GRPC_CALL_INTERNAL_UNREF(child_call, "propagate_cancel", 0); + } + child_call = next_child_call; + } while (child_call != call->first_child); + } GRPC_CALL_INTERNAL_UNREF(call, "closed", 0); } finish_read_ops(call); @@ -1176,6 +1243,22 @@ grpc_call_error grpc_call_start_ioreq_and_call_back( void grpc_call_destroy(grpc_call *c) { int cancel; + grpc_call *parent = c->parent; + + if (parent) { + gpr_mu_lock(&parent->mu); + if (c == parent->first_child) { + parent->first_child = c->sibling_next; + if (c == parent->first_child) { + parent->first_child = NULL; + } + c->sibling_prev->sibling_next = c->sibling_next; + c->sibling_next->sibling_prev = c->sibling_prev; + } + gpr_mu_unlock(&parent->mu); + GRPC_CALL_INTERNAL_UNREF(parent, "child", 1); + } + lock(c); GPR_ASSERT(!c->destroy_called); c->destroy_called = 1; @@ -1283,9 +1366,9 @@ static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline) { } GRPC_CALL_INTERNAL_REF(call, "alarm"); call->have_alarm = 1; - grpc_alarm_init(&call->alarm, - gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), - call_alarm, call, gpr_now(GPR_CLOCK_MONOTONIC)); + call->send_deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); + grpc_alarm_init(&call->alarm, call->send_deadline, call_alarm, call, + gpr_now(GPR_CLOCK_MONOTONIC)); } /* we offset status by a small amount when storing it into transport metadata @@ -1377,7 +1460,8 @@ static void recv_metadata(grpc_call *call, grpc_metadata_batch *md) { } } if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) != - 0) { + 0 && + !call->is_client) { set_deadline_alarm(call, md->deadline); } if (!is_trailing) { @@ -1455,6 +1539,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, /* Flag validation: currently allow no flags */ if (op->flags != 0) return GRPC_CALL_ERROR_INVALID_FLAGS; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_SEND_INITIAL_METADATA; req->data.send_metadata.count = op->data.send_initial_metadata.count; req->data.send_metadata.metadata = @@ -1465,7 +1550,11 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, if (!are_write_flags_valid(op->flags)) { return GRPC_CALL_ERROR_INVALID_FLAGS; } + if (op->data.send_message == NULL) { + return GRPC_CALL_ERROR_INVALID_MESSAGE; + } req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_SEND_MESSAGE; req->data.send_message = op->data.send_message; req->flags = op->flags; @@ -1477,6 +1566,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, return GRPC_CALL_ERROR_NOT_ON_SERVER; } req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_SEND_CLOSE; req->flags = op->flags; break; @@ -1487,6 +1577,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, return GRPC_CALL_ERROR_NOT_ON_CLIENT; } req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_SEND_TRAILING_METADATA; req->flags = op->flags; req->data.send_metadata.count = @@ -1494,6 +1585,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, req->data.send_metadata.metadata = op->data.send_status_from_server.trailing_metadata; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_SEND_STATUS; req->data.send_status.code = op->data.send_status_from_server.status; req->data.send_status.details = @@ -1503,6 +1595,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, op->data.send_status_from_server.status_details, 0) : NULL; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_SEND_CLOSE; break; case GRPC_OP_RECV_INITIAL_METADATA: @@ -1512,14 +1605,17 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, return GRPC_CALL_ERROR_NOT_ON_SERVER; } req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_RECV_INITIAL_METADATA; req->data.recv_metadata = op->data.recv_initial_metadata; + req->data.recv_metadata->count = 0; req->flags = op->flags; break; case GRPC_OP_RECV_MESSAGE: /* Flag validation: currently allow no flags */ if (op->flags != 0) return GRPC_CALL_ERROR_INVALID_FLAGS; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_RECV_MESSAGE; req->data.recv_message = op->data.recv_message; req->flags = op->flags; @@ -1531,21 +1627,26 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, return GRPC_CALL_ERROR_NOT_ON_SERVER; } req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_RECV_STATUS; req->flags = op->flags; req->data.recv_status.set_value = set_status_value_directly; req->data.recv_status.user_data = op->data.recv_status_on_client.status; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_RECV_STATUS_DETAILS; req->data.recv_status_details.details = op->data.recv_status_on_client.status_details; req->data.recv_status_details.details_capacity = op->data.recv_status_on_client.status_details_capacity; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_RECV_TRAILING_METADATA; req->data.recv_metadata = op->data.recv_status_on_client.trailing_metadata; + req->data.recv_metadata->count = 0; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_RECV_CLOSE; finish_func = finish_batch_with_close; break; @@ -1553,12 +1654,14 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, /* Flag validation: currently allow no flags */ if (op->flags != 0) return GRPC_CALL_ERROR_INVALID_FLAGS; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_RECV_STATUS; req->flags = op->flags; req->data.recv_status.set_value = set_cancelled_value; req->data.recv_status.user_data = op->data.recv_close_on_server.cancelled; req = &reqs[out++]; + if (out > GRPC_IOREQ_OP_COUNT) return GRPC_CALL_ERROR_BATCH_TOO_BIG; req->op = GRPC_IOREQ_RECV_CLOSE; finish_func = finish_batch_with_close; break; |