Merge github.com:grpc/grpc into new_transport_op
diff --git a/src/core/ext/census/grpc_filter.c b/src/core/ext/census/grpc_filter.c
index b80d831..69d488a 100644
--- a/src/core/ext/census/grpc_filter.c
+++ b/src/core/ext/census/grpc_filter.c
@@ -78,7 +78,8 @@
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
if (op->send_initial_metadata) {
- extract_and_annotate_method_tag(op->send_initial_metadata, calld, chand);
+ extract_and_annotate_method_tag(
+ op->payload->send_initial_metadata.send_initial_metadata, calld, chand);
}
}
@@ -107,9 +108,12 @@
call_data *calld = elem->call_data;
if (op->recv_initial_metadata) {
/* substitute our callback for the op callback */
- calld->recv_initial_metadata = op->recv_initial_metadata;
- calld->on_done_recv = op->recv_initial_metadata_ready;
- op->recv_initial_metadata_ready = &calld->finish_recv;
+ calld->recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata;
+ calld->on_done_recv =
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready;
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready =
+ &calld->finish_recv;
}
}
diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c
index bf64f84..d308096 100644
--- a/src/core/ext/client_channel/client_channel.c
+++ b/src/core/ext/client_channel/client_channel.c
@@ -449,7 +449,7 @@
static void start_transport_op_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error_ignored) {
grpc_transport_op *op = arg;
- grpc_channel_element *elem = op->transport_private.args[0];
+ grpc_channel_element *elem = op->handler_private.extra_arg;
channel_data *chand = elem->channel_data;
if (op->on_connectivity_state_change != NULL) {
@@ -510,12 +510,12 @@
op->bind_pollset);
}
- op->transport_private.args[0] = elem;
+ op->handler_private.extra_arg = elem;
GRPC_CHANNEL_STACK_REF(chand->owning_stack, "start_transport_op");
grpc_closure_sched(
- exec_ctx, grpc_closure_init(
- &op->transport_private.closure, start_transport_op_locked,
- op, grpc_combiner_scheduler(chand->combiner, false)),
+ exec_ctx,
+ grpc_closure_init(&op->handler_private.closure, start_transport_op_locked,
+ op, grpc_combiner_scheduler(chand->combiner, false)),
GRPC_ERROR_NONE);
}
@@ -926,7 +926,7 @@
return;
}
/* if this is a cancellation, then we can raise our cancelled flag */
- if (op->cancel_error != GRPC_ERROR_NONE) {
+ if (op->cancel_stream) {
if (!gpr_atm_rel_cas(&calld->subchannel_call, 0,
(gpr_atm)(uintptr_t)CANCELLED_CALL)) {
/* recurse to retry */
@@ -939,27 +939,29 @@
cancelled before any ops are passed down (e.g., if the deadline
is in the past when the call starts), we can return the right
error to the caller when the first op does get passed down. */
- calld->cancel_error = GRPC_ERROR_REF(op->cancel_error);
+ calld->cancel_error =
+ GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error);
switch (calld->creation_phase) {
case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING:
- fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error));
+ fail_locked(exec_ctx, calld,
+ GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
break;
case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL:
- pick_subchannel_locked(exec_ctx, elem, NULL, 0,
- &calld->connected_subchannel, NULL,
- GRPC_ERROR_REF(op->cancel_error));
+ pick_subchannel_locked(
+ exec_ctx, elem, NULL, 0, &calld->connected_subchannel, NULL,
+ GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
break;
}
grpc_transport_stream_op_finish_with_failure(
- exec_ctx, op, GRPC_ERROR_REF(op->cancel_error));
+ exec_ctx, op,
+ GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
/* early out */
return;
}
}
/* if we don't have a subchannel, try to get one */
if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING &&
- calld->connected_subchannel == NULL &&
- op->send_initial_metadata != NULL) {
+ calld->connected_subchannel == NULL && op->send_initial_metadata) {
calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL;
grpc_closure_init(&calld->next_step, subchannel_ready_locked, elem,
grpc_combiner_scheduler(chand->combiner, true));
@@ -967,10 +969,11 @@
/* If a subchannel is not available immediately, the polling entity from
call_data should be provided to channel_data's interested_parties, so
that IO of the lb_policy and resolver could be done under it. */
- if (pick_subchannel_locked(exec_ctx, elem, op->send_initial_metadata,
- op->send_initial_metadata_flags,
- &calld->connected_subchannel, &calld->next_step,
- GRPC_ERROR_NONE)) {
+ if (pick_subchannel_locked(
+ exec_ctx, elem,
+ op->payload->send_initial_metadata.send_initial_metadata,
+ op->payload->send_initial_metadata.send_initial_metadata_flags,
+ &calld->connected_subchannel, &calld->next_step, GRPC_ERROR_NONE)) {
calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
} else {
@@ -1007,7 +1010,7 @@
GPR_TIMER_BEGIN("start_transport_stream_op_locked", 0);
grpc_transport_stream_op *op = arg;
- grpc_call_element *elem = op->handler_private.args[0];
+ grpc_call_element *elem = op->handler_private.extra_arg;
call_data *calld = elem->call_data;
start_transport_stream_op_locked_inner(exec_ctx, op, elem);
@@ -1050,7 +1053,7 @@
}
/* we failed; lock and figure out what to do */
GRPC_CALL_STACK_REF(calld->owning_call, "start_transport_stream_op");
- op->handler_private.args[0] = elem;
+ op->handler_private.extra_arg = elem;
grpc_closure_sched(
exec_ctx,
grpc_closure_init(&op->handler_private.closure,
diff --git a/src/core/ext/load_reporting/load_reporting_filter.c b/src/core/ext/load_reporting/load_reporting_filter.c
index c275063..7e720b8 100644
--- a/src/core/ext/load_reporting/load_reporting_filter.c
+++ b/src/core/ext/load_reporting/load_reporting_filter.c
@@ -190,10 +190,13 @@
call_data *calld = elem->call_data;
if (op->recv_initial_metadata) {
- calld->recv_initial_metadata = op->recv_initial_metadata;
+ calld->recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata;
/* substitute our callback for the higher callback */
- calld->ops_recv_initial_metadata_ready = op->recv_initial_metadata_ready;
- op->recv_initial_metadata_ready = &calld->on_initial_md_ready;
+ calld->ops_recv_initial_metadata_ready =
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready;
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready =
+ &calld->on_initial_md_ready;
}
grpc_call_next_op(exec_ctx, elem, op);
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index a368453..8e40308 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -1125,8 +1125,9 @@
GPR_TIMER_BEGIN("perform_stream_op_locked", 0);
grpc_transport_stream_op *op = stream_op;
- grpc_chttp2_transport *t = op->handler_private.args[0];
- grpc_chttp2_stream *s = op->handler_private.args[1];
+ grpc_chttp2_stream *s = op->handler_private.extra_arg;
+ grpc_transport_stream_op_payload *op_payload = op->payload;
+ grpc_chttp2_transport *t = s->t;
if (grpc_http_trace) {
char *str = grpc_transport_stream_op_string(op);
@@ -1134,10 +1135,12 @@
op->on_complete);
gpr_free(str);
if (op->send_initial_metadata) {
- log_metadata(op->send_initial_metadata, s->id, t->is_client, true);
+ log_metadata(op_payload->send_initial_metadata.send_initial_metadata,
+ s->id, t->is_client, true);
}
if (op->send_trailing_metadata) {
- log_metadata(op->send_trailing_metadata, s->id, t->is_client, false);
+ log_metadata(op_payload->send_trailing_metadata.send_trailing_metadata,
+ s->id, t->is_client, false);
}
}
@@ -1152,23 +1155,25 @@
on_complete->next_data.scratch = CLOSURE_BARRIER_FIRST_REF_BIT;
on_complete->error_data.error = GRPC_ERROR_NONE;
- if (op->collect_stats != NULL) {
+ if (op->collect_stats) {
GPR_ASSERT(s->collecting_stats == NULL);
- s->collecting_stats = op->collect_stats;
+ s->collecting_stats = op_payload->collect_stats.collect_stats;
on_complete->next_data.scratch |= CLOSURE_BARRIER_STATS_BIT;
}
- if (op->cancel_error != GRPC_ERROR_NONE) {
- grpc_chttp2_cancel_stream(exec_ctx, t, s, op->cancel_error);
+ if (op->cancel_stream) {
+ grpc_chttp2_cancel_stream(exec_ctx, t, s,
+ op_payload->cancel_stream.cancel_error);
}
- if (op->send_initial_metadata != NULL) {
+ if (op->send_initial_metadata) {
GPR_ASSERT(s->send_initial_metadata_finished == NULL);
on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
s->send_initial_metadata_finished = add_closure_barrier(on_complete);
- s->send_initial_metadata = op->send_initial_metadata;
+ s->send_initial_metadata =
+ op_payload->send_initial_metadata.send_initial_metadata;
const size_t metadata_size =
- grpc_metadata_batch_size(op->send_initial_metadata);
+ grpc_metadata_batch_size(s->send_initial_metadata);
const size_t metadata_peer_limit =
t->settings[GRPC_PEER_SETTINGS]
[GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE];
@@ -1188,7 +1193,7 @@
GRPC_ERROR_INT_LIMIT, (intptr_t)metadata_peer_limit),
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED));
} else {
- if (contains_non_ok_status(op->send_initial_metadata)) {
+ if (contains_non_ok_status(s->send_initial_metadata)) {
s->seen_error = true;
}
if (!s->write_closed) {
@@ -1222,7 +1227,7 @@
}
}
- if (op->send_message != NULL) {
+ if (op->send_message) {
on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
s->fetching_send_message_finished = add_closure_barrier(op->on_complete);
if (s->write_closed) {
@@ -1236,14 +1241,14 @@
GPR_ASSERT(s->fetching_send_message == NULL);
uint8_t *frame_hdr =
grpc_slice_buffer_tiny_add(&s->flow_controlled_buffer, 5);
- uint32_t flags = op->send_message->flags;
+ uint32_t flags = op_payload->send_message.send_message->flags;
frame_hdr[0] = (flags & GRPC_WRITE_INTERNAL_COMPRESS) != 0;
- size_t len = op->send_message->length;
+ size_t len = op_payload->send_message.send_message->length;
frame_hdr[1] = (uint8_t)(len >> 24);
frame_hdr[2] = (uint8_t)(len >> 16);
frame_hdr[3] = (uint8_t)(len >> 8);
frame_hdr[4] = (uint8_t)(len);
- s->fetching_send_message = op->send_message;
+ s->fetching_send_message = op_payload->send_message.send_message;
s->fetched_send_message_length = 0;
s->next_message_end_offset = s->flow_controlled_bytes_written +
(int64_t)s->flow_controlled_buffer.length +
@@ -1260,14 +1265,15 @@
}
}
- if (op->send_trailing_metadata != NULL) {
+ if (op->send_trailing_metadata) {
GPR_ASSERT(s->send_trailing_metadata_finished == NULL);
on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
s->send_trailing_metadata_finished = add_closure_barrier(on_complete);
- s->send_trailing_metadata = op->send_trailing_metadata;
+ s->send_trailing_metadata =
+ op_payload->send_trailing_metadata.send_trailing_metadata;
s->write_buffering = false;
const size_t metadata_size =
- grpc_metadata_batch_size(op->send_trailing_metadata);
+ grpc_metadata_batch_size(s->send_trailing_metadata);
const size_t metadata_peer_limit =
t->settings[GRPC_PEER_SETTINGS]
[GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE];
@@ -1283,14 +1289,14 @@
GRPC_ERROR_INT_LIMIT, (intptr_t)metadata_peer_limit),
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED));
} else {
- if (contains_non_ok_status(op->send_trailing_metadata)) {
+ if (contains_non_ok_status(s->send_trailing_metadata)) {
s->seen_error = true;
}
if (s->write_closed) {
s->send_trailing_metadata = NULL;
grpc_chttp2_complete_closure_step(
exec_ctx, t, s, &s->send_trailing_metadata_finished,
- grpc_metadata_batch_is_empty(op->send_trailing_metadata)
+ grpc_metadata_batch_is_empty(s->send_trailing_metadata)
? GRPC_ERROR_NONE
: GRPC_ERROR_CREATE("Attempt to send trailing metadata after "
"stream was closed"),
@@ -1305,17 +1311,19 @@
}
}
- if (op->recv_initial_metadata != NULL) {
+ if (op->recv_initial_metadata) {
GPR_ASSERT(s->recv_initial_metadata_ready == NULL);
- s->recv_initial_metadata_ready = op->recv_initial_metadata_ready;
- s->recv_initial_metadata = op->recv_initial_metadata;
+ s->recv_initial_metadata_ready =
+ op_payload->recv_initial_metadata.recv_initial_metadata_ready;
+ s->recv_initial_metadata =
+ op_payload->recv_initial_metadata.recv_initial_metadata;
grpc_chttp2_maybe_complete_recv_initial_metadata(exec_ctx, t, s);
}
- if (op->recv_message != NULL) {
+ if (op->recv_message) {
GPR_ASSERT(s->recv_message_ready == NULL);
- s->recv_message_ready = op->recv_message_ready;
- s->recv_message = op->recv_message;
+ s->recv_message_ready = op_payload->recv_message.recv_message_ready;
+ s->recv_message = op_payload->recv_message.recv_message;
if (s->id != 0 &&
(s->incoming_frames.head == NULL || s->incoming_frames.head->is_tail)) {
incoming_byte_stream_update_flow_control(exec_ctx, t, s, 5, 0);
@@ -1323,10 +1331,11 @@
grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s);
}
- if (op->recv_trailing_metadata != NULL) {
+ if (op->recv_trailing_metadata) {
GPR_ASSERT(s->recv_trailing_metadata_finished == NULL);
s->recv_trailing_metadata_finished = add_closure_barrier(on_complete);
- s->recv_trailing_metadata = op->recv_trailing_metadata;
+ s->recv_trailing_metadata =
+ op_payload->recv_trailing_metadata.recv_trailing_metadata;
s->final_metadata_requested = true;
grpc_chttp2_maybe_complete_recv_trailing_metadata(exec_ctx, t, s);
}
@@ -1350,8 +1359,7 @@
gpr_free(str);
}
- op->handler_private.args[0] = gt;
- op->handler_private.args[1] = gs;
+ op->handler_private.extra_arg = gs;
GRPC_CHTTP2_STREAM_REF(s, "perform_stream_op");
grpc_closure_sched(
exec_ctx,
@@ -1421,7 +1429,7 @@
void *stream_op,
grpc_error *error_ignored) {
grpc_transport_op *op = stream_op;
- grpc_chttp2_transport *t = op->transport_private.args[0];
+ grpc_chttp2_transport *t = op->handler_private.extra_arg;
grpc_error *close_transport = op->disconnect_with_error;
if (op->on_connectivity_state_change != NULL) {
@@ -1467,10 +1475,10 @@
grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
char *msg = grpc_transport_op_string(op);
gpr_free(msg);
- op->transport_private.args[0] = gt;
+ op->handler_private.extra_arg = gt;
GRPC_CHTTP2_REF_TRANSPORT(t, "transport_op");
grpc_closure_sched(
- exec_ctx, grpc_closure_init(&op->transport_private.closure,
+ exec_ctx, grpc_closure_init(&op->handler_private.closure,
perform_transport_op_locked, op,
grpc_combiner_scheduler(t->combiner, false)),
GRPC_ERROR_NONE);
diff --git a/src/core/lib/channel/channel_stack.c b/src/core/lib/channel/channel_stack.c
index 3fb2a60..e92e46a 100644
--- a/src/core/lib/channel/channel_stack.c
+++ b/src/core/lib/channel/channel_stack.c
@@ -293,6 +293,7 @@
grpc_call_element *elem,
grpc_error *error) {
grpc_transport_stream_op *op = grpc_make_transport_stream_op(NULL);
- op->cancel_error = error;
+ op->cancel_stream = true;
+ op->payload->cancel_stream.cancel_error = error;
elem->filter->start_transport_stream_op(exec_ctx, elem, op);
}
diff --git a/src/core/lib/channel/compress_filter.c b/src/core/lib/channel/compress_filter.c
index aa41014..5321760 100644
--- a/src/core/lib/channel/compress_filter.c
+++ b/src/core/lib/channel/compress_filter.c
@@ -210,7 +210,8 @@
grpc_slice_buffer_stream_init(&calld->replacement_stream, &calld->slices,
calld->send_flags);
- calld->send_op->send_message = &calld->replacement_stream.base;
+ calld->send_op->payload->send_message.send_message =
+ &calld->replacement_stream.base;
calld->post_send = calld->send_op->on_complete;
calld->send_op->on_complete = &calld->send_done;
@@ -231,9 +232,9 @@
static void continue_send_message(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem) {
call_data *calld = elem->call_data;
- while (grpc_byte_stream_next(exec_ctx, calld->send_op->send_message,
- &calld->incoming_slice, ~(size_t)0,
- &calld->got_slice)) {
+ while (grpc_byte_stream_next(
+ exec_ctx, calld->send_op->payload->send_message.send_message,
+ &calld->incoming_slice, ~(size_t)0, &calld->got_slice)) {
grpc_slice_buffer_add(&calld->slices, calld->incoming_slice);
if (calld->send_length == calld->slices.length) {
finish_send_message(exec_ctx, elem);
@@ -251,17 +252,18 @@
if (op->send_initial_metadata) {
grpc_error *error = process_send_initial_metadata(
- exec_ctx, elem, op->send_initial_metadata);
+ exec_ctx, elem,
+ op->payload->send_initial_metadata.send_initial_metadata);
if (error != GRPC_ERROR_NONE) {
grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error);
return;
}
}
- if (op->send_message != NULL &&
- !skip_compression(elem, op->send_message->flags)) {
+ if (op->send_message &&
+ !skip_compression(elem, op->payload->send_message.send_message->flags)) {
calld->send_op = op;
- calld->send_length = op->send_message->length;
- calld->send_flags = op->send_message->flags;
+ calld->send_length = op->payload->send_message.send_message->length;
+ calld->send_flags = op->payload->send_message.send_message->flags;
continue_send_message(exec_ctx, elem);
} else {
/* pass control down the stack */
diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c
index 5a12d62..7efced2 100644
--- a/src/core/lib/channel/deadline_filter.c
+++ b/src/core/lib/channel/deadline_filter.c
@@ -200,12 +200,12 @@
grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
grpc_transport_stream_op* op) {
grpc_deadline_state* deadline_state = elem->call_data;
- if (op->cancel_error != GRPC_ERROR_NONE) {
+ if (op->cancel_stream) {
cancel_timer_if_needed(exec_ctx, deadline_state);
} else {
// Make sure we know when the call is complete, so that we can cancel
// the timer.
- if (op->recv_trailing_metadata != NULL) {
+ if (op->recv_trailing_metadata) {
inject_on_complete_cb(deadline_state, op);
}
}
@@ -286,26 +286,29 @@
grpc_call_element* elem,
grpc_transport_stream_op* op) {
server_call_data* calld = elem->call_data;
- if (op->cancel_error != GRPC_ERROR_NONE) {
+ if (op->cancel_stream) {
cancel_timer_if_needed(exec_ctx, &calld->base.deadline_state);
} else {
// If we're receiving initial metadata, we need to get the deadline
// from the recv_initial_metadata_ready callback. So we inject our
// own callback into that hook.
- if (op->recv_initial_metadata_ready != NULL) {
- calld->next_recv_initial_metadata_ready = op->recv_initial_metadata_ready;
- calld->recv_initial_metadata = op->recv_initial_metadata;
+ if (op->recv_initial_metadata) {
+ calld->next_recv_initial_metadata_ready =
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready;
+ calld->recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata;
grpc_closure_init(&calld->recv_initial_metadata_ready,
recv_initial_metadata_ready, elem,
grpc_schedule_on_exec_ctx);
- op->recv_initial_metadata_ready = &calld->recv_initial_metadata_ready;
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready =
+ &calld->recv_initial_metadata_ready;
}
// Make sure we know when the call is complete, so that we can cancel
// the timer.
// Note that we trigger this on recv_trailing_metadata, even though
// the client never sends trailing metadata, because this is the
// hook that tells us when the call is complete on the server side.
- if (op->recv_trailing_metadata != NULL) {
+ if (op->recv_trailing_metadata) {
inject_on_complete_cb(&calld->base.deadline_state, op);
}
}
diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c
index c031533..aaefa51 100644
--- a/src/core/lib/channel/http_client_filter.c
+++ b/src/core/lib/channel/http_client_filter.c
@@ -63,7 +63,7 @@
uint8_t *payload_bytes;
/* Vars to read data off of send_message */
- grpc_transport_stream_op send_op;
+ grpc_transport_stream_op *send_op;
uint32_t send_length;
uint32_t send_flags;
grpc_slice incoming_slice;
@@ -219,9 +219,9 @@
grpc_call_element *elem) {
call_data *calld = elem->call_data;
uint8_t *wrptr = calld->payload_bytes;
- while (grpc_byte_stream_next(exec_ctx, calld->send_op.send_message,
- &calld->incoming_slice, ~(size_t)0,
- &calld->got_slice)) {
+ while (grpc_byte_stream_next(
+ exec_ctx, calld->send_op->payload->send_message.send_message,
+ &calld->incoming_slice, ~(size_t)0, &calld->got_slice)) {
memcpy(wrptr, GRPC_SLICE_START_PTR(calld->incoming_slice),
GRPC_SLICE_LENGTH(calld->incoming_slice));
wrptr += GRPC_SLICE_LENGTH(calld->incoming_slice);
@@ -242,40 +242,47 @@
/* Pass down the original send_message op that was blocked.*/
grpc_slice_buffer_stream_init(&calld->replacement_stream, &calld->slices,
calld->send_flags);
- calld->send_op.send_message = &calld->replacement_stream.base;
- calld->post_send = calld->send_op.on_complete;
- calld->send_op.on_complete = &calld->send_done;
- grpc_call_next_op(exec_ctx, elem, &calld->send_op);
+ calld->send_op->payload->send_message.send_message =
+ &calld->replacement_stream.base;
+ calld->post_send = calld->send_op->on_complete;
+ calld->send_op->on_complete = &calld->send_done;
+ grpc_call_next_op(exec_ctx, elem, calld->send_op);
} else {
continue_send_message(exec_ctx, elem);
}
}
-static grpc_error *hc_mutate_op(grpc_exec_ctx *exec_ctx,
- grpc_call_element *elem,
- grpc_transport_stream_op *op) {
+typedef struct hc_mutate_op_result {
+ grpc_error *error;
+ bool op_stalled;
+} hc_mutate_op_result;
+
+static hc_mutate_op_result hc_mutate_op(grpc_exec_ctx *exec_ctx,
+ grpc_call_element *elem,
+ grpc_transport_stream_op *op) {
/* grab pointers to our data from the call element */
call_data *calld = elem->call_data;
channel_data *channeld = elem->channel_data;
- grpc_error *error;
+ hc_mutate_op_result result = {.error = GRPC_ERROR_NONE, .op_stalled = false};
- if (op->send_initial_metadata != NULL) {
+ if (op->send_initial_metadata) {
/* Decide which HTTP VERB to use. We use GET if the request is marked
cacheable, and the operation contains both initial metadata and send
message, and the payload is below the size threshold, and all the data
for this request is immediately available. */
grpc_mdelem method = GRPC_MDELEM_METHOD_POST;
- if ((op->send_initial_metadata_flags &
+ if (op->send_message &&
+ (op->payload->send_initial_metadata.send_initial_metadata_flags &
GRPC_INITIAL_METADATA_CACHEABLE_REQUEST) &&
- op->send_message != NULL &&
- op->send_message->length < channeld->max_payload_size_for_get) {
+ op->payload->send_message.send_message->length <
+ channeld->max_payload_size_for_get) {
method = GRPC_MDELEM_METHOD_GET;
/* The following write to calld->send_message_blocked isn't racy with
reads in hc_start_transport_op (which deals with SEND_MESSAGE ops) because
being here means ops->send_message is not NULL, which is primarily
guarding the read there. */
calld->send_message_blocked = true;
- } else if (op->send_initial_metadata_flags &
+ } else if (op->payload->send_initial_metadata.send_initial_metadata_flags &
GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) {
method = GRPC_MDELEM_METHOD_PUT;
}
@@ -283,25 +290,28 @@
/* Attempt to read the data from send_message and create a header field. */
if (grpc_mdelem_eq(method, GRPC_MDELEM_METHOD_GET)) {
/* allocate memory to hold the entire payload */
- calld->payload_bytes = gpr_malloc(op->send_message->length);
+ calld->payload_bytes =
+ gpr_malloc(op->payload->send_message.send_message->length);
/* read slices of send_message and copy into payload_bytes */
- calld->send_op = *op;
- calld->send_length = op->send_message->length;
- calld->send_flags = op->send_message->flags;
+ calld->send_op = op;
+ calld->send_length = op->payload->send_message.send_message->length;
+ calld->send_flags = op->payload->send_message.send_message->flags;
continue_send_message(exec_ctx, elem);
+ result.op_stalled = true;
if (calld->send_message_blocked == false) {
/* when all the send_message data is available, then create a MDELEM and
append to headers */
grpc_mdelem payload_bin = grpc_mdelem_from_slices(
exec_ctx, GRPC_MDSTR_GRPC_PAYLOAD_BIN,
- grpc_slice_from_copied_buffer((const char *)calld->payload_bytes,
- op->send_message->length));
- error =
- grpc_metadata_batch_add_tail(exec_ctx, op->send_initial_metadata,
- &calld->payload_bin, payload_bin);
- if (error != GRPC_ERROR_NONE) return error;
+ grpc_slice_from_copied_buffer(
+ (const char *)calld->payload_bytes,
+ op->payload->send_message.send_message->length));
+ result.error = grpc_metadata_batch_add_tail(
+ exec_ctx, op->payload->send_initial_metadata.send_initial_metadata,
+ &calld->payload_bin, payload_bin);
+ if (result.error != GRPC_ERROR_NONE) return result;
calld->on_complete = op->on_complete;
op->on_complete = &calld->hc_on_complete;
op->send_message = NULL;
@@ -314,52 +324,65 @@
}
}
- remove_if_present(exec_ctx, op->send_initial_metadata, GRPC_BATCH_METHOD);
- remove_if_present(exec_ctx, op->send_initial_metadata, GRPC_BATCH_SCHEME);
- remove_if_present(exec_ctx, op->send_initial_metadata, GRPC_BATCH_TE);
- remove_if_present(exec_ctx, op->send_initial_metadata,
+ remove_if_present(exec_ctx,
+ op->payload->send_initial_metadata.send_initial_metadata,
+ GRPC_BATCH_METHOD);
+ remove_if_present(exec_ctx,
+ op->payload->send_initial_metadata.send_initial_metadata,
+ GRPC_BATCH_SCHEME);
+ remove_if_present(exec_ctx,
+ op->payload->send_initial_metadata.send_initial_metadata,
+ GRPC_BATCH_TE);
+ remove_if_present(exec_ctx,
+ op->payload->send_initial_metadata.send_initial_metadata,
GRPC_BATCH_CONTENT_TYPE);
- remove_if_present(exec_ctx, op->send_initial_metadata,
+ remove_if_present(exec_ctx,
+ op->payload->send_initial_metadata.send_initial_metadata,
GRPC_BATCH_USER_AGENT);
/* Send : prefixed headers, which have to be before any application
layer headers. */
- error = grpc_metadata_batch_add_head(exec_ctx, op->send_initial_metadata,
- &calld->method, method);
- if (error != GRPC_ERROR_NONE) return error;
- error =
- grpc_metadata_batch_add_head(exec_ctx, op->send_initial_metadata,
- &calld->scheme, channeld->static_scheme);
- if (error != GRPC_ERROR_NONE) return error;
- error = grpc_metadata_batch_add_tail(exec_ctx, op->send_initial_metadata,
- &calld->te_trailers,
- GRPC_MDELEM_TE_TRAILERS);
- if (error != GRPC_ERROR_NONE) return error;
- error = grpc_metadata_batch_add_tail(
- exec_ctx, op->send_initial_metadata, &calld->content_type,
- GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC);
- if (error != GRPC_ERROR_NONE) return error;
- error = grpc_metadata_batch_add_tail(exec_ctx, op->send_initial_metadata,
- &calld->user_agent,
- GRPC_MDELEM_REF(channeld->user_agent));
- if (error != GRPC_ERROR_NONE) return error;
+ result.error = grpc_metadata_batch_add_head(
+ exec_ctx, op->payload->send_initial_metadata.send_initial_metadata,
+ &calld->method, method);
+ if (result.error != GRPC_ERROR_NONE) return result;
+ result.error = grpc_metadata_batch_add_head(
+ exec_ctx, op->payload->send_initial_metadata.send_initial_metadata,
+ &calld->scheme, channeld->static_scheme);
+ if (result.error != GRPC_ERROR_NONE) return result;
+ result.error = grpc_metadata_batch_add_tail(
+ exec_ctx, op->payload->send_initial_metadata.send_initial_metadata,
+ &calld->te_trailers, GRPC_MDELEM_TE_TRAILERS);
+ if (result.error != GRPC_ERROR_NONE) return result;
+ result.error = grpc_metadata_batch_add_tail(
+ exec_ctx, op->payload->send_initial_metadata.send_initial_metadata,
+ &calld->content_type, GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC);
+ if (result.error != GRPC_ERROR_NONE) return result;
+ result.error = grpc_metadata_batch_add_tail(
+ exec_ctx, op->payload->send_initial_metadata.send_initial_metadata,
+ &calld->user_agent, GRPC_MDELEM_REF(channeld->user_agent));
+ if (result.error != GRPC_ERROR_NONE) return result;
}
- if (op->recv_initial_metadata != NULL) {
+ if (op->recv_initial_metadata) {
/* substitute our callback for the higher callback */
- calld->recv_initial_metadata = op->recv_initial_metadata;
- calld->on_done_recv_initial_metadata = op->recv_initial_metadata_ready;
- op->recv_initial_metadata_ready = &calld->hc_on_recv_initial_metadata;
+ calld->recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata;
+ calld->on_done_recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready;
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready =
+ &calld->hc_on_recv_initial_metadata;
}
- if (op->recv_trailing_metadata != NULL) {
+ if (op->recv_trailing_metadata) {
/* substitute our callback for the higher callback */
- calld->recv_trailing_metadata = op->recv_trailing_metadata;
+ calld->recv_trailing_metadata =
+ op->payload->recv_trailing_metadata.recv_trailing_metadata;
calld->on_done_recv_trailing_metadata = op->on_complete;
op->on_complete = &calld->hc_on_recv_trailing_metadata;
}
- return GRPC_ERROR_NONE;
+ return result;
}
static void hc_start_transport_op(grpc_exec_ctx *exec_ctx,
@@ -367,18 +390,14 @@
grpc_transport_stream_op *op) {
GPR_TIMER_BEGIN("hc_start_transport_op", 0);
GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
- grpc_error *error = hc_mutate_op(exec_ctx, elem, op);
- if (error != GRPC_ERROR_NONE) {
- grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error);
+ hc_mutate_op_result result = hc_mutate_op(exec_ctx, elem, op);
+ if (result.error != GRPC_ERROR_NONE) {
+ grpc_transport_stream_op_finish_with_failure(exec_ctx, op, result.error);
+ } else if (result.op_stalled) {
+ /* Don't forward the op. send_message contains slices that aren't ready yet.
+ The call will be forwarded by the op_complete of slice read call. */
} else {
- call_data *calld = elem->call_data;
- if (op->send_message != NULL && calld->send_message_blocked) {
- /* Don't forward the op. send_message contains slices that aren't ready
- yet. The call will be forwarded by the op_complete of slice read call.
- */
- } else {
- grpc_call_next_op(exec_ctx, elem, op);
- }
+ grpc_call_next_op(exec_ctx, elem, op);
}
GPR_TIMER_END("hc_start_transport_op", 0);
}
diff --git a/src/core/lib/channel/http_server_filter.c b/src/core/lib/channel/http_server_filter.c
index fb70de8..920549b 100644
--- a/src/core/lib/channel/http_server_filter.c
+++ b/src/core/lib/channel/http_server_filter.c
@@ -57,8 +57,7 @@
bool payload_bin_delivered;
grpc_metadata_batch *recv_initial_metadata;
- bool *recv_idempotent_request;
- bool *recv_cacheable_request;
+ uint32_t *recv_initial_metadata_flags;
/** Closure to call when finished with the hs_on_recv hook */
grpc_closure *on_done_recv;
/** Closure to call when we retrieve read message from the payload-bin header
@@ -115,14 +114,21 @@
if (b->idx.named.method != NULL) {
if (grpc_mdelem_eq(b->idx.named.method->md, GRPC_MDELEM_METHOD_POST)) {
- *calld->recv_idempotent_request = false;
- *calld->recv_cacheable_request = false;
+ *calld->recv_initial_metadata_flags &=
+ ~(GRPC_INITIAL_METADATA_CACHEABLE_REQUEST |
+ GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST);
} else if (grpc_mdelem_eq(b->idx.named.method->md,
GRPC_MDELEM_METHOD_PUT)) {
- *calld->recv_idempotent_request = true;
+ *calld->recv_initial_metadata_flags &=
+ ~GRPC_INITIAL_METADATA_CACHEABLE_REQUEST;
+ *calld->recv_initial_metadata_flags |=
+ GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST;
} else if (grpc_mdelem_eq(b->idx.named.method->md,
GRPC_MDELEM_METHOD_GET)) {
- *calld->recv_cacheable_request = true;
+ *calld->recv_initial_metadata_flags |=
+ GRPC_INITIAL_METADATA_CACHEABLE_REQUEST;
+ *calld->recv_initial_metadata_flags |=
+ ~GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST;
} else {
add_error(error_name, &error,
grpc_attach_md_to_error(GRPC_ERROR_CREATE("Bad header"),
@@ -276,19 +282,24 @@
/* grab pointers to our data from the call element */
call_data *calld = elem->call_data;
- if (op->send_initial_metadata != NULL) {
+ if (op->send_initial_metadata) {
grpc_error *error = GRPC_ERROR_NONE;
static const char *error_name = "Failed sending initial metadata";
- add_error(error_name, &error, grpc_metadata_batch_add_head(
- exec_ctx, op->send_initial_metadata,
- &calld->status, GRPC_MDELEM_STATUS_200));
+ add_error(
+ error_name, &error,
+ grpc_metadata_batch_add_head(
+ exec_ctx, op->payload->send_initial_metadata.send_initial_metadata,
+ &calld->status, GRPC_MDELEM_STATUS_200));
+ add_error(
+ error_name, &error,
+ grpc_metadata_batch_add_tail(
+ exec_ctx, op->payload->send_initial_metadata.send_initial_metadata,
+ &calld->content_type,
+ GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC));
add_error(error_name, &error,
- grpc_metadata_batch_add_tail(
- exec_ctx, op->send_initial_metadata, &calld->content_type,
- GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC));
- add_error(error_name, &error,
- server_filter_outgoing_metadata(exec_ctx, elem,
- op->send_initial_metadata));
+ server_filter_outgoing_metadata(
+ exec_ctx, elem,
+ op->payload->send_initial_metadata.send_initial_metadata));
if (error != GRPC_ERROR_NONE) {
grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error);
return;
@@ -297,20 +308,23 @@
if (op->recv_initial_metadata) {
/* substitute our callback for the higher callback */
- GPR_ASSERT(op->recv_idempotent_request != NULL);
- GPR_ASSERT(op->recv_cacheable_request != NULL);
- calld->recv_initial_metadata = op->recv_initial_metadata;
- calld->recv_idempotent_request = op->recv_idempotent_request;
- calld->recv_cacheable_request = op->recv_cacheable_request;
- calld->on_done_recv = op->recv_initial_metadata_ready;
- op->recv_initial_metadata_ready = &calld->hs_on_recv;
+ GPR_ASSERT(op->payload->recv_initial_metadata.recv_flags != NULL);
+ calld->recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata;
+ calld->recv_initial_metadata_flags =
+ op->payload->recv_initial_metadata.recv_flags;
+ calld->on_done_recv =
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready;
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready =
+ &calld->hs_on_recv;
}
if (op->recv_message) {
- calld->recv_message_ready = op->recv_message_ready;
- calld->pp_recv_message = op->recv_message;
- if (op->recv_message_ready) {
- op->recv_message_ready = &calld->hs_recv_message_ready;
+ calld->recv_message_ready = op->payload->recv_message.recv_message_ready;
+ calld->pp_recv_message = op->payload->recv_message.recv_message;
+ if (op->payload->recv_message.recv_message_ready) {
+ op->payload->recv_message.recv_message_ready =
+ &calld->hs_recv_message_ready;
}
if (op->on_complete) {
calld->on_complete = op->on_complete;
@@ -320,7 +334,8 @@
if (op->send_trailing_metadata) {
grpc_error *error = server_filter_outgoing_metadata(
- exec_ctx, elem, op->send_trailing_metadata);
+ exec_ctx, elem,
+ op->payload->send_trailing_metadata.send_trailing_metadata);
if (error != GRPC_ERROR_NONE) {
grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error);
return;
diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c
index b424c0d..e99b767 100644
--- a/src/core/lib/channel/message_size_filter.c
+++ b/src/core/lib/channel/message_size_filter.c
@@ -141,11 +141,13 @@
grpc_transport_stream_op* op) {
call_data* calld = elem->call_data;
// Check max send message size.
- if (op->send_message != NULL && calld->max_send_size >= 0 &&
- op->send_message->length > (size_t)calld->max_send_size) {
+ if (op->send_message && calld->max_send_size >= 0 &&
+ op->payload->send_message.send_message->length >
+ (size_t)calld->max_send_size) {
char* message_string;
gpr_asprintf(&message_string, "Sent message larger than max (%u vs. %d)",
- op->send_message->length, calld->max_send_size);
+ op->payload->send_message.send_message->length,
+ calld->max_send_size);
grpc_transport_stream_op_finish_with_failure(
exec_ctx, op, grpc_error_set_int(GRPC_ERROR_CREATE(message_string),
GRPC_ERROR_INT_GRPC_STATUS,
@@ -154,10 +156,11 @@
return;
}
// Inject callback for receiving a message.
- if (op->recv_message_ready != NULL) {
- calld->next_recv_message_ready = op->recv_message_ready;
- calld->recv_message = op->recv_message;
- op->recv_message_ready = &calld->recv_message_ready;
+ if (op->payload->recv_message.recv_message_ready != NULL) {
+ calld->next_recv_message_ready =
+ op->payload->recv_message.recv_message_ready;
+ calld->recv_message = op->payload->recv_message.recv_message;
+ op->payload->recv_message.recv_message_ready = &calld->recv_message_ready;
}
// Chain to the next filter.
grpc_call_next_op(exec_ctx, elem, op);
diff --git a/src/core/lib/security/transport/client_auth_filter.c b/src/core/lib/security/transport/client_auth_filter.c
index a23082a..98ef365 100644
--- a/src/core/lib/security/transport/client_auth_filter.c
+++ b/src/core/lib/security/transport/client_auth_filter.c
@@ -120,8 +120,8 @@
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAUTHENTICATED);
} else {
GPR_ASSERT(num_md <= MAX_CREDENTIALS_METADATA_COUNT);
- GPR_ASSERT(op->send_initial_metadata != NULL);
- mdb = op->send_initial_metadata;
+ GPR_ASSERT(op->send_initial_metadata);
+ mdb = op->payload->send_initial_metadata.send_initial_metadata;
for (i = 0; i < num_md; i++) {
add_error(&error,
grpc_metadata_batch_add_tail(
@@ -174,7 +174,9 @@
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
grpc_client_security_context *ctx =
- (grpc_client_security_context *)op->context[GRPC_CONTEXT_SECURITY].value;
+ (grpc_client_security_context *)op->payload
+ ->context[GRPC_CONTEXT_SECURITY]
+ .value;
grpc_call_credentials *channel_call_creds =
chand->security_connector->request_metadata_creds;
int call_creds_has_md = (ctx != NULL) && (ctx->creds != NULL);
@@ -248,23 +250,25 @@
grpc_linked_mdelem *l;
grpc_client_security_context *sec_ctx = NULL;
- if (calld->security_context_set == 0 && op->cancel_error == GRPC_ERROR_NONE) {
+ if (calld->security_context_set == 0 && !op->cancel_stream) {
calld->security_context_set = 1;
- GPR_ASSERT(op->context);
- if (op->context[GRPC_CONTEXT_SECURITY].value == NULL) {
- op->context[GRPC_CONTEXT_SECURITY].value =
+ GPR_ASSERT(op->payload->context != NULL);
+ if (op->payload->context[GRPC_CONTEXT_SECURITY].value == NULL) {
+ op->payload->context[GRPC_CONTEXT_SECURITY].value =
grpc_client_security_context_create();
- op->context[GRPC_CONTEXT_SECURITY].destroy =
+ op->payload->context[GRPC_CONTEXT_SECURITY].destroy =
grpc_client_security_context_destroy;
}
- sec_ctx = op->context[GRPC_CONTEXT_SECURITY].value;
+ sec_ctx = op->payload->context[GRPC_CONTEXT_SECURITY].value;
GRPC_AUTH_CONTEXT_UNREF(sec_ctx->auth_context, "client auth filter");
sec_ctx->auth_context =
GRPC_AUTH_CONTEXT_REF(chand->auth_context, "client_auth_filter");
}
- if (op->send_initial_metadata != NULL) {
- for (l = op->send_initial_metadata->list.head; l != NULL; l = l->next) {
+ if (op->send_initial_metadata) {
+ for (l = op->payload->send_initial_metadata.send_initial_metadata->list
+ .head;
+ l != NULL; l = l->next) {
grpc_mdelem md = l->md;
/* Pointer comparison is OK for md_elems created from the same context.
*/
diff --git a/src/core/lib/security/transport/server_auth_filter.c b/src/core/lib/security/transport/server_auth_filter.c
index 14619d9..afd07d6 100644
--- a/src/core/lib/security/transport/server_auth_filter.c
+++ b/src/core/lib/security/transport/server_auth_filter.c
@@ -139,9 +139,10 @@
? error_details
: "Authentication metadata processing failed.";
calld->transport_op->send_initial_metadata = NULL;
- if (calld->transport_op->send_message != NULL) {
- grpc_byte_stream_destroy(&exec_ctx, calld->transport_op->send_message);
- calld->transport_op->send_message = NULL;
+ if (calld->transport_op->send_message) {
+ grpc_byte_stream_destroy(
+ &exec_ctx, calld->transport_op->payload->send_message.send_message);
+ calld->transport_op->send_message = false;
}
calld->transport_op->send_trailing_metadata = NULL;
grpc_closure_sched(&exec_ctx, calld->on_done_recv,
@@ -173,11 +174,14 @@
grpc_transport_stream_op *op) {
call_data *calld = elem->call_data;
- if (op->recv_initial_metadata != NULL) {
+ if (op->recv_initial_metadata) {
/* substitute our callback for the higher callback */
- calld->recv_initial_metadata = op->recv_initial_metadata;
- calld->on_done_recv = op->recv_initial_metadata_ready;
- op->recv_initial_metadata_ready = &calld->auth_on_recv;
+ calld->recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata;
+ calld->on_done_recv =
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready;
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready =
+ &calld->auth_on_recv;
calld->transport_op = op;
}
}
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index c2547c5..0c803c7 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -116,24 +116,19 @@
typedef struct batch_control {
grpc_call *call;
- grpc_cq_completion cq_completion;
+ union {
+ grpc_cq_completion cq_completion;
+ struct {
+ void *tag;
+ bool is_closure;
+ } notify_tag;
+ } completion_data;
grpc_closure finish_batch;
- void *notify_tag;
gpr_refcount steps_to_complete;
grpc_error *errors[MAX_ERRORS_PER_BATCH];
gpr_atm num_errors;
- uint8_t send_initial_metadata;
- uint8_t send_message;
- uint8_t send_final_op;
- uint8_t recv_initial_metadata;
- uint8_t recv_message;
- uint8_t recv_final_op;
- uint8_t is_notify_tag_closure;
-
- /* TODO(ctiller): now that this is inlined, figure out how much of the above
- state can be eliminated */
grpc_transport_stream_op op;
} batch_control;
@@ -167,6 +162,7 @@
bool has_initial_md_been_received;
batch_control active_batches[MAX_CONCURRENT_BATCHES];
+ grpc_transport_stream_op_payload stream_op_payload;
/* first idx: is_receiving, second idx: is_trailing */
grpc_metadata_batch metadata_batch[2][2];
@@ -283,6 +279,7 @@
/* Always support no compression */
GPR_BITSET(&call->encodings_accepted_by_peer, GRPC_COMPRESS_NONE);
call->is_client = args->server_transport_data == NULL;
+ call->stream_op_payload.context = call->context;
grpc_slice path = grpc_empty_slice();
if (call->is_client) {
GPR_ASSERT(args->add_initial_metadata_count <
@@ -516,7 +513,6 @@
GPR_TIMER_BEGIN("execute_op", 0);
elem = CALL_ELEM_FROM_CALL(call, 0);
- op->context = call->context;
elem->filter->start_transport_stream_op(exec_ctx, elem, op);
GPR_TIMER_END("execute_op", 0);
}
@@ -567,6 +563,7 @@
grpc_closure closure;
grpc_call *call;
grpc_transport_stream_op op;
+ grpc_transport_stream_op_payload payload;
} termination_closure;
static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp,
@@ -580,7 +577,9 @@
grpc_error *error) {
termination_closure *tc = tcp;
memset(&tc->op, 0, sizeof(tc->op));
- tc->op.cancel_error = GRPC_ERROR_REF(error);
+ tc->op.payload = &tc->payload;
+ tc->op.cancel_stream = true;
+ tc->op.payload->cancel_stream.cancel_error = GRPC_ERROR_REF(error);
/* reuse closure to catch completion */
tc->op.on_complete = grpc_closure_init(&tc->closure, done_termination, tc,
grpc_schedule_on_exec_ctx);
@@ -1034,16 +1033,13 @@
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);
- }
batch_control *bctl = &call->active_batches[slot];
if (bctl->call != NULL) {
return NULL;
}
memset(bctl, 0, sizeof(*bctl));
bctl->call = call;
+ bctl->op.payload = &call->stream_op_payload;
return bctl;
}
@@ -1085,20 +1081,20 @@
gpr_mu_lock(&call->mu);
- if (bctl->send_initial_metadata) {
+ if (bctl->op.send_initial_metadata) {
grpc_metadata_batch_destroy(
exec_ctx,
&call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]);
}
- if (bctl->send_message) {
+ if (bctl->op.send_message) {
call->sending_message = false;
}
- if (bctl->send_final_op) {
+ if (bctl->op.send_trailing_metadata) {
grpc_metadata_batch_destroy(
exec_ctx,
&call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]);
}
- if (bctl->recv_final_op) {
+ if (bctl->op.recv_trailing_metadata) {
grpc_metadata_batch *md =
&call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
recv_trailing_filter(exec_ctx, call, md);
@@ -1132,15 +1128,16 @@
}
gpr_mu_unlock(&call->mu);
- if (bctl->is_notify_tag_closure) {
+ if (bctl->completion_data.notify_tag.is_closure) {
/* unrefs bctl->error */
bctl->call = NULL;
- grpc_closure_run(exec_ctx, bctl->notify_tag, error);
+ grpc_closure_run(exec_ctx, bctl->completion_data.notify_tag.tag, error);
GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "completion");
} else {
/* unrefs bctl->error */
- grpc_cq_end_op(exec_ctx, bctl->call->cq, bctl->notify_tag, error,
- finish_batch_completion, bctl, &bctl->cq_completion);
+ grpc_cq_end_op(
+ exec_ctx, bctl->call->cq, bctl->completion_data.notify_tag.tag, error,
+ finish_batch_completion, bctl, &bctl->completion_data.cq_completion);
}
}
@@ -1390,12 +1387,14 @@
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);
+ bctl->completion_data.notify_tag.tag = notify_tag;
+ bctl->completion_data.notify_tag.is_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));
+ grpc_transport_stream_op_payload *stream_op_payload =
+ &call->stream_op_payload;
stream_op->covered_by_poller = true;
/* rewrite batch ops into a transport op */
@@ -1449,8 +1448,8 @@
error = GRPC_CALL_ERROR_INVALID_METADATA;
goto done_with_error;
}
- bctl->send_initial_metadata = 1;
- call->sent_initial_metadata = 1;
+ stream_op->send_initial_metadata = true;
+ call->sent_initial_metadata = true;
if (!prepare_application_metadata(
exec_ctx, call, (int)op->data.send_initial_metadata.count,
op->data.send_initial_metadata.metadata, 0, call->is_client,
@@ -1460,9 +1459,10 @@
}
/* TODO(ctiller): just make these the same variable? */
call->metadata_batch[0][0].deadline = call->send_deadline;
- stream_op->send_initial_metadata =
+ stream_op_payload->send_initial_metadata.send_initial_metadata =
&call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */];
- stream_op->send_initial_metadata_flags = op->flags;
+ stream_op_payload->send_initial_metadata.send_initial_metadata_flags =
+ op->flags;
break;
case GRPC_OP_SEND_MESSAGE:
if (!are_write_flags_valid(op->flags)) {
@@ -1477,8 +1477,8 @@
error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
goto done_with_error;
}
- bctl->send_message = 1;
- call->sending_message = 1;
+ stream_op->send_message = true;
+ call->sending_message = true;
grpc_slice_buffer_stream_init(
&call->sending_stream,
&op->data.send_message.send_message->data.raw.slice_buffer,
@@ -1490,7 +1490,8 @@
GRPC_COMPRESS_NONE) {
call->sending_stream.base.flags |= GRPC_WRITE_INTERNAL_COMPRESS;
}
- stream_op->send_message = &call->sending_stream.base;
+ stream_op_payload->send_message.send_message =
+ &call->sending_stream.base;
break;
case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
/* Flag validation: currently allow no flags */
@@ -1506,9 +1507,9 @@
error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
goto done_with_error;
}
- bctl->send_final_op = 1;
+ stream_op->send_trailing_metadata = true;
call->sent_final_op = 1;
- stream_op->send_trailing_metadata =
+ stream_op_payload->send_trailing_metadata.send_trailing_metadata =
&call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */];
break;
case GRPC_OP_SEND_STATUS_FROM_SERVER:
@@ -1530,7 +1531,7 @@
error = GRPC_CALL_ERROR_INVALID_METADATA;
goto done_with_error;
}
- bctl->send_final_op = 1;
+ stream_op->send_trailing_metadata = true;
call->sent_final_op = 1;
GPR_ASSERT(call->send_extra_metadata_count == 0);
call->send_extra_metadata_count = 1;
@@ -1568,7 +1569,7 @@
error = GRPC_CALL_ERROR_INVALID_METADATA;
goto done_with_error;
}
- stream_op->send_trailing_metadata =
+ stream_op_payload->send_trailing_metadata.send_trailing_metadata =
&call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */];
break;
case GRPC_OP_RECV_INITIAL_METADATA:
@@ -1591,10 +1592,10 @@
grpc_closure_init(&call->receiving_initial_metadata_ready,
receiving_initial_metadata_ready, bctl,
grpc_schedule_on_exec_ctx);
- bctl->recv_initial_metadata = 1;
- stream_op->recv_initial_metadata =
+ stream_op->recv_initial_metadata = true;
+ stream_op_payload->recv_initial_metadata.recv_initial_metadata =
&call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */];
- stream_op->recv_initial_metadata_ready =
+ stream_op_payload->recv_initial_metadata.recv_initial_metadata_ready =
&call->receiving_initial_metadata_ready;
num_completion_callbacks_needed++;
break;
@@ -1608,13 +1609,14 @@
error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
goto done_with_error;
}
- call->receiving_message = 1;
- bctl->recv_message = 1;
+ call->receiving_message = true;
+ stream_op->recv_message = true;
call->receiving_buffer = op->data.recv_message.recv_message;
- stream_op->recv_message = &call->receiving_stream;
+ stream_op_payload->recv_message.recv_message = &call->receiving_stream;
grpc_closure_init(&call->receiving_stream_ready, receiving_stream_ready,
bctl, grpc_schedule_on_exec_ctx);
- stream_op->recv_message_ready = &call->receiving_stream_ready;
+ stream_op_payload->recv_message.recv_message_ready =
+ &call->receiving_stream_ready;
num_completion_callbacks_needed++;
break;
case GRPC_OP_RECV_STATUS_ON_CLIENT:
@@ -1637,10 +1639,11 @@
call->final_op.client.status = op->data.recv_status_on_client.status;
call->final_op.client.status_details =
op->data.recv_status_on_client.status_details;
- bctl->recv_final_op = 1;
- stream_op->recv_trailing_metadata =
+ stream_op->recv_trailing_metadata = true;
+ stream_op->collect_stats = true;
+ stream_op_payload->recv_trailing_metadata.recv_trailing_metadata =
&call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
- stream_op->collect_stats =
+ stream_op_payload->collect_stats.collect_stats =
&call->final_info.stats.transport_stream_stats;
break;
case GRPC_OP_RECV_CLOSE_ON_SERVER:
@@ -1660,10 +1663,11 @@
call->requested_final_op = 1;
call->final_op.server.cancelled =
op->data.recv_close_on_server.cancelled;
- bctl->recv_final_op = 1;
- stream_op->recv_trailing_metadata =
+ stream_op->recv_trailing_metadata = true;
+ stream_op->collect_stats = true;
+ stream_op_payload->recv_trailing_metadata.recv_trailing_metadata =
&call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
- stream_op->collect_stats =
+ stream_op_payload->collect_stats.collect_stats =
&call->final_info.stats.transport_stream_stats;
break;
}
@@ -1675,7 +1679,6 @@
}
gpr_ref_init(&bctl->steps_to_complete, num_completion_callbacks_needed);
- stream_op->context = call->context;
grpc_closure_init(&bctl->finish_batch, finish_batch, bctl,
grpc_schedule_on_exec_ctx);
stream_op->on_complete = &bctl->finish_batch;
@@ -1690,25 +1693,25 @@
done_with_error:
/* reverse any mutations that occured */
- if (bctl->send_initial_metadata) {
+ if (stream_op->send_initial_metadata) {
call->sent_initial_metadata = 0;
grpc_metadata_batch_clear(exec_ctx, &call->metadata_batch[0][0]);
}
- if (bctl->send_message) {
+ if (stream_op->send_message) {
call->sending_message = 0;
grpc_byte_stream_destroy(exec_ctx, &call->sending_stream.base);
}
- if (bctl->send_final_op) {
+ if (stream_op->send_trailing_metadata) {
call->sent_final_op = 0;
grpc_metadata_batch_clear(exec_ctx, &call->metadata_batch[0][1]);
}
- if (bctl->recv_initial_metadata) {
+ if (stream_op->recv_initial_metadata) {
call->received_initial_metadata = 0;
}
- if (bctl->recv_message) {
+ if (stream_op->recv_message) {
call->receiving_message = 0;
}
- if (bctl->recv_final_op) {
+ if (stream_op->recv_trailing_metadata) {
call->requested_final_op = 0;
}
gpr_mu_unlock(&call->mu);
diff --git a/src/core/lib/surface/lame_client.c b/src/core/lib/surface/lame_client.c
index 49bc4c1..4c2c3df 100644
--- a/src/core/lib/surface/lame_client.c
+++ b/src/core/lib/surface/lame_client.c
@@ -84,10 +84,12 @@
grpc_call_element *elem,
grpc_transport_stream_op *op) {
GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
- if (op->recv_initial_metadata != NULL) {
- fill_metadata(exec_ctx, elem, op->recv_initial_metadata);
- } else if (op->recv_trailing_metadata != NULL) {
- fill_metadata(exec_ctx, elem, op->recv_trailing_metadata);
+ if (op->recv_initial_metadata) {
+ fill_metadata(exec_ctx, elem,
+ op->payload->recv_initial_metadata.recv_initial_metadata);
+ } else if (op->recv_trailing_metadata) {
+ fill_metadata(exec_ctx, elem,
+ op->payload->recv_trailing_metadata.recv_trailing_metadata);
}
grpc_transport_stream_op_finish_with_failure(
exec_ctx, op, GRPC_ERROR_CREATE("lame client channel"));
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index b360579..17b3f8c 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -154,8 +154,7 @@
grpc_completion_queue *cq_new;
grpc_metadata_batch *recv_initial_metadata;
- bool recv_idempotent_request;
- bool recv_cacheable_request;
+ uint32_t recv_initial_metadata_flags;
grpc_metadata_array initial_metadata;
request_matcher *request_matcher;
@@ -498,13 +497,7 @@
rc->data.batch.details->host = grpc_slice_ref_internal(calld->host);
rc->data.batch.details->method = grpc_slice_ref_internal(calld->path);
rc->data.batch.details->deadline = calld->deadline;
- rc->data.batch.details->flags =
- (calld->recv_idempotent_request
- ? GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST
- : 0) |
- (calld->recv_cacheable_request
- ? GRPC_INITIAL_METADATA_CACHEABLE_REQUEST
- : 0);
+ rc->data.batch.details->flags = calld->recv_initial_metadata_flags;
break;
case REGISTERED_CALL:
*rc->data.registered.deadline = calld->deadline;
@@ -632,7 +625,8 @@
if (!grpc_slice_eq(rm->host, calld->host)) continue;
if (!grpc_slice_eq(rm->method, calld->path)) continue;
if ((rm->flags & GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) &&
- !calld->recv_idempotent_request) {
+ 0 == (calld->recv_initial_metadata_flags &
+ GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST)) {
continue;
}
finish_start_new_rpc(exec_ctx, server, elem,
@@ -649,7 +643,8 @@
if (rm->has_host) continue;
if (!grpc_slice_eq(rm->method, calld->path)) continue;
if ((rm->flags & GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) &&
- !calld->recv_idempotent_request) {
+ 0 == (calld->recv_initial_metadata_flags &
+ GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST)) {
continue;
}
finish_start_new_rpc(exec_ctx, server, elem,
@@ -783,13 +778,16 @@
grpc_transport_stream_op *op) {
call_data *calld = elem->call_data;
- if (op->recv_initial_metadata != NULL) {
- GPR_ASSERT(op->recv_idempotent_request == NULL);
- calld->recv_initial_metadata = op->recv_initial_metadata;
- calld->on_done_recv_initial_metadata = op->recv_initial_metadata_ready;
- op->recv_initial_metadata_ready = &calld->server_on_recv_initial_metadata;
- op->recv_idempotent_request = &calld->recv_idempotent_request;
- op->recv_cacheable_request = &calld->recv_cacheable_request;
+ if (op->recv_initial_metadata) {
+ GPR_ASSERT(op->payload->recv_initial_metadata.recv_flags == NULL);
+ calld->recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata;
+ calld->on_done_recv_initial_metadata =
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready;
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready =
+ &calld->server_on_recv_initial_metadata;
+ op->payload->recv_initial_metadata.recv_flags =
+ &calld->recv_initial_metadata_flags;
}
}
diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c
index 165950e..3b0d18c 100644
--- a/src/core/lib/transport/transport.c
+++ b/src/core/lib/transport/transport.c
@@ -215,11 +215,20 @@
void grpc_transport_stream_op_finish_with_failure(grpc_exec_ctx *exec_ctx,
grpc_transport_stream_op *op,
grpc_error *error) {
- grpc_closure_sched(exec_ctx, op->recv_message_ready, GRPC_ERROR_REF(error));
- grpc_closure_sched(exec_ctx, op->recv_initial_metadata_ready,
- GRPC_ERROR_REF(error));
+ if (op->recv_message) {
+ grpc_closure_sched(exec_ctx, op->payload->recv_message.recv_message_ready,
+ GRPC_ERROR_REF(error));
+ }
+ if (op->recv_initial_metadata) {
+ grpc_closure_sched(
+ exec_ctx,
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready,
+ GRPC_ERROR_REF(error));
+ }
grpc_closure_sched(exec_ctx, op->on_complete, error);
- GRPC_ERROR_UNREF(op->cancel_error);
+ if (op->cancel_stream) {
+ GRPC_ERROR_UNREF(op->payload->cancel_stream.cancel_error);
+ }
}
typedef struct {
@@ -249,6 +258,7 @@
grpc_closure outer_on_complete;
grpc_closure *inner_on_complete;
grpc_transport_stream_op op;
+ grpc_transport_stream_op_payload payload;
} made_transport_stream_op;
static void destroy_made_transport_stream_op(grpc_exec_ctx *exec_ctx, void *arg,
@@ -260,11 +270,11 @@
grpc_transport_stream_op *grpc_make_transport_stream_op(
grpc_closure *on_complete) {
- made_transport_stream_op *op = gpr_malloc(sizeof(*op));
+ made_transport_stream_op *op = gpr_zalloc(sizeof(*op));
+ op->op.payload = &op->payload;
grpc_closure_init(&op->outer_on_complete, destroy_made_transport_stream_op,
op, grpc_schedule_on_exec_ctx);
op->inner_on_complete = on_complete;
- memset(&op->op, 0, sizeof(op->op));
op->op.on_complete = &op->outer_on_complete;
return &op->op;
}
diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h
index cc1c277..bfcce70 100644
--- a/src/core/lib/transport/transport.h
+++ b/src/core/lib/transport/transport.h
@@ -108,9 +108,12 @@
grpc_transport_stream_stats *to);
typedef struct {
+ void *extra_arg;
grpc_closure closure;
- void *args[2];
-} grpc_transport_private_op_data;
+} grpc_handler_private_op_data;
+
+typedef struct grpc_transport_stream_op_payload
+ grpc_transport_stream_op_payload;
/* Transport stream op: a set of operations to perform on a transport
against a single stream */
@@ -120,43 +123,83 @@
have been completed. */
grpc_closure *on_complete;
+ /** Values for the stream op (fields set are determined by flags above) */
+ grpc_transport_stream_op_payload *payload;
+
/** Is the completion of this op covered by a poller (if false: the op should
complete independently of some pollset being polled) */
- bool covered_by_poller;
+ bool covered_by_poller : 1;
- /** Send initial metadata to the peer, from the provided metadata batch.
- idempotent_request MUST be set if this is non-null */
- grpc_metadata_batch *send_initial_metadata;
- /** Iff send_initial_metadata != NULL, flags associated with
- send_initial_metadata: a bitfield of GRPC_INITIAL_METADATA_xxx */
- uint32_t send_initial_metadata_flags;
+ /** Send initial metadata to the peer, from the provided metadata batch. */
+ bool send_initial_metadata : 1;
/** Send trailing metadata to the peer, from the provided metadata batch. */
- grpc_metadata_batch *send_trailing_metadata;
+ bool send_trailing_metadata : 1;
/** Send message data to the peer, from the provided byte stream. */
- grpc_byte_stream *send_message;
+ bool send_message : 1;
/** Receive initial metadata from the stream, into provided metadata batch. */
- grpc_metadata_batch *recv_initial_metadata;
- bool *recv_idempotent_request;
- bool *recv_cacheable_request;
- /** Should be enqueued when initial metadata is ready to be processed. */
- grpc_closure *recv_initial_metadata_ready;
+ bool recv_initial_metadata : 1;
/** Receive message data from the stream, into provided byte stream. */
- grpc_byte_stream **recv_message;
- /** Should be enqueued when one message is ready to be processed. */
- grpc_closure *recv_message_ready;
+ bool recv_message : 1;
/** Receive trailing metadata from the stream, into provided metadata batch.
*/
- grpc_metadata_batch *recv_trailing_metadata;
+ bool recv_trailing_metadata : 1;
/** Collect any stats into provided buffer, zero internal stat counters */
- grpc_transport_stream_stats *collect_stats;
+ bool collect_stats : 1;
- /** If != GRPC_ERROR_NONE, forcefully close this stream.
+ /** Cancel this stream with the provided error */
+ bool cancel_stream : 1;
+
+ /***************************************************************************
+ * remaining fields are initialized and used at the discretion of the
+ * current handler of the op */
+
+ grpc_handler_private_op_data handler_private;
+} grpc_transport_stream_op;
+
+struct grpc_transport_stream_op_payload {
+ struct {
+ grpc_metadata_batch *send_initial_metadata;
+ /** Iff send_initial_metadata != NULL, flags associated with
+ send_initial_metadata: a bitfield of GRPC_INITIAL_METADATA_xxx */
+ uint32_t send_initial_metadata_flags;
+ } send_initial_metadata;
+
+ struct {
+ grpc_metadata_batch *send_trailing_metadata;
+ } send_trailing_metadata;
+
+ struct {
+ grpc_byte_stream *send_message;
+ } send_message;
+
+ struct {
+ grpc_metadata_batch *recv_initial_metadata;
+ uint32_t *recv_flags;
+ /** Should be enqueued when initial metadata is ready to be processed. */
+ grpc_closure *recv_initial_metadata_ready;
+ } recv_initial_metadata;
+
+ struct {
+ grpc_byte_stream **recv_message;
+ /** Should be enqueued when one message is ready to be processed. */
+ grpc_closure *recv_message_ready;
+ } recv_message;
+
+ struct {
+ grpc_metadata_batch *recv_trailing_metadata;
+ } recv_trailing_metadata;
+
+ struct {
+ grpc_transport_stream_stats *collect_stats;
+ } collect_stats;
+
+ /** Forcefully close this stream.
The HTTP2 semantics should be:
- server side: if cancel_error has GRPC_ERROR_INT_GRPC_STATUS, and
trailing metadata has not been sent, send trailing metadata with status
@@ -166,17 +209,13 @@
convert to a HTTP2 error code using
grpc_chttp2_grpc_status_to_http2_error. Send a RST_STREAM with this
error. */
- grpc_error *cancel_error;
+ struct {
+ grpc_error *cancel_error;
+ } cancel_stream;
/* Indexes correspond to grpc_context_index enum values */
grpc_call_context_element *context;
-
- /***************************************************************************
- * remaining fields are initialized and used at the discretion of the
- * current handler of the op */
-
- grpc_transport_private_op_data handler_private;
-} grpc_transport_stream_op;
+};
/** Transport op: a set of operations to perform on a transport as a whole */
typedef struct grpc_transport_op {
@@ -209,7 +248,7 @@
* remaining fields are initialized and used at the discretion of the
* transport implementation */
- grpc_transport_private_op_data transport_private;
+ grpc_handler_private_op_data handler_private;
} grpc_transport_op;
/* Returns the amount of memory required to store a grpc_stream for this
diff --git a/src/core/lib/transport/transport_op_string.c b/src/core/lib/transport/transport_op_string.c
index 28360e3..0ec6a6e 100644
--- a/src/core/lib/transport/transport_op_string.c
+++ b/src/core/lib/transport/transport_op_string.c
@@ -81,45 +81,49 @@
gpr_strvec_add(
&b, gpr_strdup(op->covered_by_poller ? "[COVERED]" : "[UNCOVERED]"));
- if (op->send_initial_metadata != NULL) {
+ if (op->send_initial_metadata) {
gpr_strvec_add(&b, gpr_strdup(" "));
gpr_strvec_add(&b, gpr_strdup("SEND_INITIAL_METADATA{"));
- put_metadata_list(&b, *op->send_initial_metadata);
+ put_metadata_list(
+ &b, *op->payload->send_initial_metadata.send_initial_metadata);
gpr_strvec_add(&b, gpr_strdup("}"));
}
- if (op->send_message != NULL) {
+ if (op->send_message) {
gpr_strvec_add(&b, gpr_strdup(" "));
gpr_asprintf(&tmp, "SEND_MESSAGE:flags=0x%08x:len=%d",
- op->send_message->flags, op->send_message->length);
+ op->payload->send_message.send_message->flags,
+ op->payload->send_message.send_message->length);
gpr_strvec_add(&b, tmp);
}
- if (op->send_trailing_metadata != NULL) {
+ if (op->send_trailing_metadata) {
gpr_strvec_add(&b, gpr_strdup(" "));
gpr_strvec_add(&b, gpr_strdup("SEND_TRAILING_METADATA{"));
- put_metadata_list(&b, *op->send_trailing_metadata);
+ put_metadata_list(
+ &b, *op->payload->send_trailing_metadata.send_trailing_metadata);
gpr_strvec_add(&b, gpr_strdup("}"));
}
- if (op->recv_initial_metadata != NULL) {
+ if (op->recv_initial_metadata) {
gpr_strvec_add(&b, gpr_strdup(" "));
gpr_strvec_add(&b, gpr_strdup("RECV_INITIAL_METADATA"));
}
- if (op->recv_message != NULL) {
+ if (op->recv_message) {
gpr_strvec_add(&b, gpr_strdup(" "));
gpr_strvec_add(&b, gpr_strdup("RECV_MESSAGE"));
}
- if (op->recv_trailing_metadata != NULL) {
+ if (op->recv_trailing_metadata) {
gpr_strvec_add(&b, gpr_strdup(" "));
gpr_strvec_add(&b, gpr_strdup("RECV_TRAILING_METADATA"));
}
- if (op->cancel_error != GRPC_ERROR_NONE) {
+ if (op->cancel_stream) {
gpr_strvec_add(&b, gpr_strdup(" "));
- const char *msg = grpc_error_string(op->cancel_error);
+ const char *msg =
+ grpc_error_string(op->payload->cancel_stream.cancel_error);
gpr_asprintf(&tmp, "CANCEL:%s", msg);
gpr_strvec_add(&b, tmp);
diff --git a/src/cpp/common/channel_filter.h b/src/cpp/common/channel_filter.h
index 79c4bab..cf913fe 100644
--- a/src/cpp/common/channel_filter.h
+++ b/src/cpp/common/channel_filter.h
@@ -149,10 +149,22 @@
/// long as the TransportStreamOp object does.
explicit TransportStreamOp(grpc_transport_stream_op *op)
: op_(op),
- send_initial_metadata_(op->send_initial_metadata),
- send_trailing_metadata_(op->send_trailing_metadata),
- recv_initial_metadata_(op->recv_initial_metadata),
- recv_trailing_metadata_(op->recv_trailing_metadata) {}
+ send_initial_metadata_(
+ op->send_initial_metadata
+ ? op->payload->send_initial_metadata.send_initial_metadata
+ : nullptr),
+ send_trailing_metadata_(
+ op->send_trailing_metadata
+ ? op->payload->send_trailing_metadata.send_trailing_metadata
+ : nullptr),
+ recv_initial_metadata_(
+ op->recv_initial_metadata
+ ? op->payload->recv_initial_metadata.recv_initial_metadata
+ : nullptr),
+ recv_trailing_metadata_(
+ op->recv_trailing_metadata
+ ? op->payload->recv_trailing_metadata.recv_trailing_metadata
+ : nullptr) {}
grpc_transport_stream_op *op() const { return op_; }
@@ -160,50 +172,57 @@
void set_on_complete(grpc_closure *closure) { op_->on_complete = closure; }
MetadataBatch *send_initial_metadata() {
- return op_->send_initial_metadata == nullptr ? nullptr
- : &send_initial_metadata_;
+ return op_->send_initial_metadata ? &send_initial_metadata_ : nullptr;
}
MetadataBatch *send_trailing_metadata() {
- return op_->send_trailing_metadata == nullptr ? nullptr
- : &send_trailing_metadata_;
+ return op_->send_trailing_metadata ? &send_trailing_metadata_ : nullptr;
}
MetadataBatch *recv_initial_metadata() {
- return op_->recv_initial_metadata == nullptr ? nullptr
- : &recv_initial_metadata_;
+ return op_->recv_initial_metadata ? &recv_initial_metadata_ : nullptr;
}
MetadataBatch *recv_trailing_metadata() {
- return op_->recv_trailing_metadata == nullptr ? nullptr
- : &recv_trailing_metadata_;
+ return op_->recv_trailing_metadata ? &recv_trailing_metadata_ : nullptr;
}
uint32_t *send_initial_metadata_flags() const {
- return &op_->send_initial_metadata_flags;
+ return op_->send_initial_metadata
+ ? &op_->payload->send_initial_metadata
+ .send_initial_metadata_flags
+ : nullptr;
}
grpc_closure *recv_initial_metadata_ready() const {
- return op_->recv_initial_metadata_ready;
+ return op_->recv_initial_metadata
+ ? op_->payload->recv_initial_metadata.recv_initial_metadata_ready
+ : nullptr;
}
void set_recv_initial_metadata_ready(grpc_closure *closure) {
- op_->recv_initial_metadata_ready = closure;
+ op_->payload->recv_initial_metadata.recv_initial_metadata_ready = closure;
}
- grpc_byte_stream *send_message() const { return op_->send_message; }
+ grpc_byte_stream *send_message() const {
+ return op_->send_message ? op_->payload->send_message.send_message
+ : nullptr;
+ }
void set_send_message(grpc_byte_stream *send_message) {
- op_->send_message = send_message;
+ op_->send_message = true;
+ op_->payload->send_message.send_message = send_message;
}
/// To be called only on clients and servers, respectively.
grpc_client_security_context *client_security_context() const {
- return (grpc_client_security_context *)op_->context[GRPC_CONTEXT_SECURITY]
+ return (grpc_client_security_context *)op_->payload
+ ->context[GRPC_CONTEXT_SECURITY]
.value;
}
grpc_server_security_context *server_security_context() const {
- return (grpc_server_security_context *)op_->context[GRPC_CONTEXT_SECURITY]
+ return (grpc_server_security_context *)op_->payload
+ ->context[GRPC_CONTEXT_SECURITY]
.value;
}
census_context *get_census_context() const {
- return (census_context *)op_->context[GRPC_CONTEXT_TRACING].value;
+ return (census_context *)op_->payload->context[GRPC_CONTEXT_TRACING].value;
}
private:
diff --git a/test/core/end2end/tests/filter_causes_close.c b/test/core/end2end/tests/filter_causes_close.c
index 25e6065..ebc6f0a 100644
--- a/test/core/end2end/tests/filter_causes_close.c
+++ b/test/core/end2end/tests/filter_causes_close.c
@@ -220,9 +220,10 @@
grpc_call_element *elem,
grpc_transport_stream_op *op) {
call_data *calld = elem->call_data;
- if (op->recv_initial_metadata != NULL) {
- calld->recv_im_ready = op->recv_initial_metadata_ready;
- op->recv_initial_metadata_ready =
+ if (op->recv_initial_metadata) {
+ calld->recv_im_ready =
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready;
+ op->payload->recv_initial_metadata.recv_initial_metadata_ready =
grpc_closure_create(recv_im_ready, elem, grpc_schedule_on_exec_ctx);
}
grpc_call_next_op(exec_ctx, elem, op);