| Index: third_party/grpc/src/core/surface/call.c
|
| diff --git a/third_party/grpc/src/core/surface/call.c b/third_party/grpc/src/core/surface/call.c
|
| new file mode 100644
|
| index 0000000000000000000000000000000000000000..1b117aa6b8ca752f7987688c474d159c9d60f2f5
|
| --- /dev/null
|
| +++ b/third_party/grpc/src/core/surface/call.c
|
| @@ -0,0 +1,1483 @@
|
| +/*
|
| + *
|
| + * Copyright 2015-2016, Google Inc.
|
| + * All rights reserved.
|
| + *
|
| + * Redistribution and use in source and binary forms, with or without
|
| + * modification, are permitted provided that the following conditions are
|
| + * met:
|
| + *
|
| + * * Redistributions of source code must retain the above copyright
|
| + * notice, this list of conditions and the following disclaimer.
|
| + * * Redistributions in binary form must reproduce the above
|
| + * copyright notice, this list of conditions and the following disclaimer
|
| + * in the documentation and/or other materials provided with the
|
| + * distribution.
|
| + * * Neither the name of Google Inc. nor the names of its
|
| + * contributors may be used to endorse or promote products derived from
|
| + * this software without specific prior written permission.
|
| + *
|
| + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
| + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
| + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
| + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
| + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
| + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
| + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
| + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
| + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
| + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
| + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
| + *
|
| + */
|
| +#include <assert.h>
|
| +#include <limits.h>
|
| +#include <stdio.h>
|
| +#include <stdlib.h>
|
| +#include <string.h>
|
| +
|
| +#include <grpc/compression.h>
|
| +#include <grpc/grpc.h>
|
| +#include <grpc/support/alloc.h>
|
| +#include <grpc/support/log.h>
|
| +#include <grpc/support/string_util.h>
|
| +#include <grpc/support/useful.h>
|
| +
|
| +#include "src/core/channel/channel_stack.h"
|
| +#include "src/core/compression/algorithm_metadata.h"
|
| +#include "src/core/iomgr/timer.h"
|
| +#include "src/core/profiling/timers.h"
|
| +#include "src/core/support/string.h"
|
| +#include "src/core/surface/api_trace.h"
|
| +#include "src/core/surface/call.h"
|
| +#include "src/core/surface/channel.h"
|
| +#include "src/core/surface/completion_queue.h"
|
| +#include "src/core/transport/static_metadata.h"
|
| +
|
| +/** The maximum number of concurrent batches possible.
|
| + Based upon the maximum number of individually queueable ops in the batch
|
| + api:
|
| + - initial metadata send
|
| + - message send
|
| + - status/close send (depending on client/server)
|
| + - initial metadata recv
|
| + - message recv
|
| + - status/close recv (depending on client/server) */
|
| +#define MAX_CONCURRENT_BATCHES 6
|
| +
|
| +typedef struct {
|
| + grpc_ioreq_completion_func on_complete;
|
| + void *user_data;
|
| + int success;
|
| +} completed_request;
|
| +
|
| +#define MAX_SEND_EXTRA_METADATA_COUNT 3
|
| +
|
| +/* 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 was created by some internal channel stack operation */
|
| + STATUS_FROM_CORE,
|
| + /* Status came from 'the wire' - or somewhere below the surface
|
| + layer */
|
| + STATUS_FROM_WIRE,
|
| + /* Status came from the server sending status */
|
| + STATUS_FROM_SERVER_STATUS,
|
| + STATUS_SOURCE_COUNT
|
| +} status_source;
|
| +
|
| +typedef struct {
|
| + uint8_t is_set;
|
| + grpc_status_code code;
|
| + grpc_mdstr *details;
|
| +} received_status;
|
| +
|
| +/* How far through the GRPC stream have we read? */
|
| +typedef enum {
|
| + /* We are still waiting for initial metadata to complete */
|
| + READ_STATE_INITIAL = 0,
|
| + /* We have gotten initial metadata, and are reading either
|
| + messages or trailing metadata */
|
| + READ_STATE_GOT_INITIAL_METADATA,
|
| + /* The stream is closed for reading */
|
| + READ_STATE_READ_CLOSED,
|
| + /* The stream is closed for reading & writing */
|
| + READ_STATE_STREAM_CLOSED
|
| +} read_state;
|
| +
|
| +typedef enum {
|
| + WRITE_STATE_INITIAL = 0,
|
| + WRITE_STATE_STARTED,
|
| + WRITE_STATE_WRITE_CLOSED
|
| +} write_state;
|
| +
|
| +typedef struct batch_control {
|
| + grpc_call *call;
|
| + grpc_cq_completion cq_completion;
|
| + grpc_closure finish_batch;
|
| + void *notify_tag;
|
| + gpr_refcount steps_to_complete;
|
| +
|
| + 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;
|
| + uint8_t success;
|
| +} batch_control;
|
| +
|
| +struct grpc_call {
|
| + grpc_completion_queue *cq;
|
| + grpc_channel *channel;
|
| + grpc_call *parent;
|
| + grpc_call *first_child;
|
| + /* TODO(ctiller): share with cq if possible? */
|
| + gpr_mu mu;
|
| +
|
| + /* client or server call */
|
| + uint8_t is_client;
|
| + /* is the alarm set */
|
| + uint8_t have_alarm;
|
| + /** has grpc_call_destroy been called */
|
| + uint8_t destroy_called;
|
| + /** flag indicating that cancellation is inherited */
|
| + uint8_t cancellation_is_inherited;
|
| + /** bitmask of live batches */
|
| + uint8_t used_batches;
|
| + /** which ops are in-flight */
|
| + uint8_t sent_initial_metadata;
|
| + uint8_t sending_message;
|
| + uint8_t sent_final_op;
|
| + uint8_t received_initial_metadata;
|
| + uint8_t receiving_message;
|
| + uint8_t received_final_op;
|
| +
|
| + /* have we received initial metadata */
|
| + bool has_initial_md_been_received;
|
| +
|
| + batch_control active_batches[MAX_CONCURRENT_BATCHES];
|
| +
|
| + /* first idx: is_receiving, second idx: is_trailing */
|
| + grpc_metadata_batch metadata_batch[2][2];
|
| +
|
| + /* Buffered read metadata waiting to be returned to the application.
|
| + Element 0 is initial metadata, element 1 is trailing metadata. */
|
| + grpc_metadata_array *buffered_metadata[2];
|
| +
|
| + /* Received call statuses from various sources */
|
| + received_status status[STATUS_SOURCE_COUNT];
|
| +
|
| + /* Compression algorithm for the call */
|
| + grpc_compression_algorithm compression_algorithm;
|
| + /* Supported encodings (compression algorithms), a bitset */
|
| + uint32_t encodings_accepted_by_peer;
|
| +
|
| + /* Contexts for various subsystems (security, tracing, ...). */
|
| + grpc_call_context_element context[GRPC_CONTEXT_COUNT];
|
| +
|
| + /* Deadline alarm - if have_alarm is non-zero */
|
| + grpc_timer alarm;
|
| +
|
| + /* for the client, extra metadata is initial metadata; for the
|
| + server, it's trailing metadata */
|
| + grpc_linked_mdelem send_extra_metadata[MAX_SEND_EXTRA_METADATA_COUNT];
|
| + int send_extra_metadata_count;
|
| + gpr_timespec send_deadline;
|
| +
|
| + /** 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;
|
| +
|
| + grpc_slice_buffer_stream sending_stream;
|
| + grpc_byte_stream *receiving_stream;
|
| + grpc_byte_buffer **receiving_buffer;
|
| + gpr_slice receiving_slice;
|
| + grpc_closure receiving_slice_ready;
|
| + grpc_closure receiving_stream_ready;
|
| + grpc_closure receiving_initial_metadata_ready;
|
| + uint32_t test_only_last_message_flags;
|
| +
|
| + union {
|
| + struct {
|
| + grpc_status_code *status;
|
| + char **status_details;
|
| + size_t *status_details_capacity;
|
| + } client;
|
| + struct {
|
| + int *cancelled;
|
| + } server;
|
| + } final_op;
|
| +
|
| + struct {
|
| + void *bctlp;
|
| + bool success;
|
| + } saved_receiving_stream_ready_ctx;
|
| +};
|
| +
|
| +#define CALL_STACK_FROM_CALL(call) ((grpc_call_stack *)((call) + 1))
|
| +#define CALL_FROM_CALL_STACK(call_stack) (((grpc_call *)(call_stack)) - 1)
|
| +#define CALL_ELEM_FROM_CALL(call, idx) \
|
| + grpc_call_stack_element(CALL_STACK_FROM_CALL(call), idx)
|
| +#define CALL_FROM_TOP_ELEM(top_elem) \
|
| + CALL_FROM_CALL_STACK(grpc_call_stack_from_top_element(top_elem))
|
| +
|
| +static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call,
|
| + gpr_timespec deadline);
|
| +static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
|
| + grpc_transport_stream_op *op);
|
| +static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
|
| + grpc_status_code status,
|
| + const char *description);
|
| +static void destroy_call(grpc_exec_ctx *exec_ctx, void *call_stack,
|
| + bool success);
|
| +static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
|
| + bool success);
|
| +
|
| +grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call,
|
| + uint32_t propagation_mask,
|
| + grpc_completion_queue *cq,
|
| + const void *server_transport_data,
|
| + grpc_mdelem **add_initial_metadata,
|
| + size_t add_initial_metadata_count,
|
| + gpr_timespec send_deadline) {
|
| + size_t i, j;
|
| + grpc_channel_stack *channel_stack = grpc_channel_get_channel_stack(channel);
|
| + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
|
| + grpc_call *call;
|
| + GPR_TIMER_BEGIN("grpc_call_create", 0);
|
| + call = gpr_malloc(sizeof(grpc_call) + channel_stack->call_stack_size);
|
| + memset(call, 0, sizeof(grpc_call));
|
| + gpr_mu_init(&call->mu);
|
| + call->channel = channel;
|
| + call->cq = cq;
|
| + call->parent = parent_call;
|
| + call->is_client = server_transport_data == NULL;
|
| + if (call->is_client) {
|
| + GPR_ASSERT(add_initial_metadata_count < MAX_SEND_EXTRA_METADATA_COUNT);
|
| + for (i = 0; i < add_initial_metadata_count; i++) {
|
| + call->send_extra_metadata[i].md = add_initial_metadata[i];
|
| + }
|
| + call->send_extra_metadata_count = (int)add_initial_metadata_count;
|
| + } else {
|
| + GPR_ASSERT(add_initial_metadata_count == 0);
|
| + call->send_extra_metadata_count = 0;
|
| + }
|
| + for (i = 0; i < 2; i++) {
|
| + for (j = 0; j < 2; j++) {
|
| + call->metadata_batch[i][j].deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
|
| + }
|
| + }
|
| + call->send_deadline = send_deadline;
|
| + GRPC_CHANNEL_INTERNAL_REF(channel, "call");
|
| + /* initial refcount dropped by grpc_call_destroy */
|
| + grpc_call_stack_init(&exec_ctx, channel_stack, 1, destroy_call, call,
|
| + call->context, server_transport_data,
|
| + CALL_STACK_FROM_CALL(call));
|
| + if (cq != NULL) {
|
| + GRPC_CQ_INTERNAL_REF(cq, "bind");
|
| + grpc_call_stack_set_pollset(&exec_ctx, CALL_STACK_FROM_CALL(call),
|
| + grpc_cq_pollset(cq));
|
| + }
|
| + 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(&exec_ctx, call, send_deadline);
|
| + }
|
| + grpc_exec_ctx_finish(&exec_ctx);
|
| + GPR_TIMER_END("grpc_call_create", 0);
|
| + return call;
|
| +}
|
| +
|
| +void grpc_call_set_completion_queue(grpc_exec_ctx *exec_ctx, grpc_call *call,
|
| + grpc_completion_queue *cq) {
|
| + GPR_ASSERT(cq);
|
| + call->cq = cq;
|
| + GRPC_CQ_INTERNAL_REF(cq, "bind");
|
| + grpc_call_stack_set_pollset(exec_ctx, CALL_STACK_FROM_CALL(call),
|
| + grpc_cq_pollset(cq));
|
| +}
|
| +
|
| +#ifdef GRPC_STREAM_REFCOUNT_DEBUG
|
| +#define REF_REASON reason
|
| +#define REF_ARG , const char *reason
|
| +#else
|
| +#define REF_REASON ""
|
| +#define REF_ARG
|
| +#endif
|
| +void grpc_call_internal_ref(grpc_call *c REF_ARG) {
|
| + GRPC_CALL_STACK_REF(CALL_STACK_FROM_CALL(c), REF_REASON);
|
| +}
|
| +void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c REF_ARG) {
|
| + GRPC_CALL_STACK_UNREF(exec_ctx, CALL_STACK_FROM_CALL(c), REF_REASON);
|
| +}
|
| +
|
| +static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, bool success) {
|
| + size_t i;
|
| + int ii;
|
| + grpc_call *c = call;
|
| + GPR_TIMER_BEGIN("destroy_call", 0);
|
| + for (i = 0; i < 2; i++) {
|
| + grpc_metadata_batch_destroy(
|
| + &c->metadata_batch[1 /* is_receiving */][i /* is_initial */]);
|
| + }
|
| + if (c->receiving_stream != NULL) {
|
| + grpc_byte_stream_destroy(exec_ctx, c->receiving_stream);
|
| + }
|
| + grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c));
|
| + GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->channel, "call");
|
| + gpr_mu_destroy(&c->mu);
|
| + for (i = 0; i < STATUS_SOURCE_COUNT; i++) {
|
| + if (c->status[i].details) {
|
| + GRPC_MDSTR_UNREF(c->status[i].details);
|
| + }
|
| + }
|
| + for (ii = 0; ii < c->send_extra_metadata_count; ii++) {
|
| + GRPC_MDELEM_UNREF(c->send_extra_metadata[ii].md);
|
| + }
|
| + for (i = 0; i < GRPC_CONTEXT_COUNT; i++) {
|
| + if (c->context[i].destroy) {
|
| + c->context[i].destroy(c->context[i].value);
|
| + }
|
| + }
|
| + if (c->cq) {
|
| + GRPC_CQ_INTERNAL_UNREF(c->cq, "bind");
|
| + }
|
| + gpr_free(c);
|
| + GPR_TIMER_END("destroy_call", 0);
|
| +}
|
| +
|
| +static void set_status_code(grpc_call *call, status_source source,
|
| + uint32_t status) {
|
| + if (call->status[source].is_set) return;
|
| +
|
| + call->status[source].is_set = 1;
|
| + call->status[source].code = (grpc_status_code)status;
|
| +
|
| + /* TODO(ctiller): what to do about the flush that was previously here */
|
| +}
|
| +
|
| +static void set_compression_algorithm(grpc_call *call,
|
| + grpc_compression_algorithm algo) {
|
| + call->compression_algorithm = algo;
|
| +}
|
| +
|
| +grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm(
|
| + grpc_call *call) {
|
| + grpc_compression_algorithm algorithm;
|
| + gpr_mu_lock(&call->mu);
|
| + algorithm = call->compression_algorithm;
|
| + gpr_mu_unlock(&call->mu);
|
| + return algorithm;
|
| +}
|
| +
|
| +uint32_t grpc_call_test_only_get_message_flags(grpc_call *call) {
|
| + uint32_t flags;
|
| + gpr_mu_lock(&call->mu);
|
| + flags = call->test_only_last_message_flags;
|
| + gpr_mu_unlock(&call->mu);
|
| + return flags;
|
| +}
|
| +
|
| +static void destroy_encodings_accepted_by_peer(void *p) { return; }
|
| +
|
| +static void set_encodings_accepted_by_peer(grpc_call *call, grpc_mdelem *mdel) {
|
| + size_t i;
|
| + grpc_compression_algorithm algorithm;
|
| + gpr_slice_buffer accept_encoding_parts;
|
| + gpr_slice accept_encoding_slice;
|
| + void *accepted_user_data;
|
| +
|
| + accepted_user_data =
|
| + grpc_mdelem_get_user_data(mdel, destroy_encodings_accepted_by_peer);
|
| + if (accepted_user_data != NULL) {
|
| + call->encodings_accepted_by_peer =
|
| + (uint32_t)(((uintptr_t)accepted_user_data) - 1);
|
| + return;
|
| + }
|
| +
|
| + accept_encoding_slice = mdel->value->slice;
|
| + gpr_slice_buffer_init(&accept_encoding_parts);
|
| + gpr_slice_split(accept_encoding_slice, ",", &accept_encoding_parts);
|
| +
|
| + /* No need to zero call->encodings_accepted_by_peer: grpc_call_create already
|
| + * zeroes the whole grpc_call */
|
| + /* Always support no compression */
|
| + GPR_BITSET(&call->encodings_accepted_by_peer, GRPC_COMPRESS_NONE);
|
| + for (i = 0; i < accept_encoding_parts.count; i++) {
|
| + const gpr_slice *accept_encoding_entry_slice =
|
| + &accept_encoding_parts.slices[i];
|
| + if (grpc_compression_algorithm_parse(
|
| + (const char *)GPR_SLICE_START_PTR(*accept_encoding_entry_slice),
|
| + GPR_SLICE_LENGTH(*accept_encoding_entry_slice), &algorithm)) {
|
| + GPR_BITSET(&call->encodings_accepted_by_peer, algorithm);
|
| + } else {
|
| + char *accept_encoding_entry_str =
|
| + gpr_dump_slice(*accept_encoding_entry_slice, GPR_DUMP_ASCII);
|
| + gpr_log(GPR_ERROR,
|
| + "Invalid entry in accept encoding metadata: '%s'. Ignoring.",
|
| + accept_encoding_entry_str);
|
| + gpr_free(accept_encoding_entry_str);
|
| + }
|
| + }
|
| +
|
| + gpr_slice_buffer_destroy(&accept_encoding_parts);
|
| +
|
| + grpc_mdelem_set_user_data(
|
| + mdel, destroy_encodings_accepted_by_peer,
|
| + (void *)(((uintptr_t)call->encodings_accepted_by_peer) + 1));
|
| +}
|
| +
|
| +uint32_t grpc_call_test_only_get_encodings_accepted_by_peer(grpc_call *call) {
|
| + uint32_t encodings_accepted_by_peer;
|
| + gpr_mu_lock(&call->mu);
|
| + encodings_accepted_by_peer = call->encodings_accepted_by_peer;
|
| + gpr_mu_unlock(&call->mu);
|
| + return encodings_accepted_by_peer;
|
| +}
|
| +
|
| +static void set_status_details(grpc_call *call, status_source source,
|
| + grpc_mdstr *status) {
|
| + if (call->status[source].details != NULL) {
|
| + GRPC_MDSTR_UNREF(call->status[source].details);
|
| + }
|
| + call->status[source].details = status;
|
| +}
|
| +
|
| +static void get_final_status(grpc_call *call,
|
| + void (*set_value)(grpc_status_code code,
|
| + void *user_data),
|
| + void *set_value_user_data) {
|
| + int i;
|
| + for (i = 0; i < STATUS_SOURCE_COUNT; i++) {
|
| + if (call->status[i].is_set) {
|
| + set_value(call->status[i].code, set_value_user_data);
|
| + return;
|
| + }
|
| + }
|
| + 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 get_final_details(grpc_call *call, char **out_details,
|
| + size_t *out_details_capacity) {
|
| + int i;
|
| + for (i = 0; i < STATUS_SOURCE_COUNT; i++) {
|
| + if (call->status[i].is_set) {
|
| + if (call->status[i].details) {
|
| + gpr_slice details = call->status[i].details->slice;
|
| + size_t len = GPR_SLICE_LENGTH(details);
|
| + if (len + 1 > *out_details_capacity) {
|
| + *out_details_capacity =
|
| + GPR_MAX(len + 1, *out_details_capacity * 3 / 2);
|
| + *out_details = gpr_realloc(*out_details, *out_details_capacity);
|
| + }
|
| + memcpy(*out_details, GPR_SLICE_START_PTR(details), len);
|
| + (*out_details)[len] = 0;
|
| + } else {
|
| + goto no_details;
|
| + }
|
| + return;
|
| + }
|
| + }
|
| +
|
| +no_details:
|
| + if (0 == *out_details_capacity) {
|
| + *out_details_capacity = 8;
|
| + *out_details = gpr_malloc(*out_details_capacity);
|
| + }
|
| + **out_details = 0;
|
| +}
|
| +
|
| +static grpc_linked_mdelem *linked_from_md(grpc_metadata *md) {
|
| + return (grpc_linked_mdelem *)&md->internal_data;
|
| +}
|
| +
|
| +static int prepare_application_metadata(grpc_call *call, int count,
|
| + grpc_metadata *metadata,
|
| + int is_trailing,
|
| + int prepend_extra_metadata) {
|
| + int i;
|
| + grpc_metadata_batch *batch =
|
| + &call->metadata_batch[0 /* is_receiving */][is_trailing];
|
| + if (prepend_extra_metadata) {
|
| + if (call->send_extra_metadata_count == 0) {
|
| + prepend_extra_metadata = 0;
|
| + } else {
|
| + for (i = 0; i < call->send_extra_metadata_count; i++) {
|
| + GRPC_MDELEM_REF(call->send_extra_metadata[i].md);
|
| + }
|
| + for (i = 1; i < call->send_extra_metadata_count; i++) {
|
| + call->send_extra_metadata[i].prev = &call->send_extra_metadata[i - 1];
|
| + }
|
| + for (i = 0; i < call->send_extra_metadata_count - 1; i++) {
|
| + call->send_extra_metadata[i].next = &call->send_extra_metadata[i + 1];
|
| + }
|
| + }
|
| + }
|
| + for (i = 0; i < count; i++) {
|
| + grpc_metadata *md = &metadata[i];
|
| + grpc_linked_mdelem *l = (grpc_linked_mdelem *)&md->internal_data;
|
| + GPR_ASSERT(sizeof(grpc_linked_mdelem) == sizeof(md->internal_data));
|
| + l->md = grpc_mdelem_from_string_and_buffer(
|
| + md->key, (const uint8_t *)md->value, md->value_length);
|
| + if (!grpc_header_key_is_legal(grpc_mdstr_as_c_string(l->md->key),
|
| + GRPC_MDSTR_LENGTH(l->md->key))) {
|
| + gpr_log(GPR_ERROR, "attempt to send invalid metadata key: %s",
|
| + grpc_mdstr_as_c_string(l->md->key));
|
| + return 0;
|
| + } else if (!grpc_is_binary_header(grpc_mdstr_as_c_string(l->md->key),
|
| + GRPC_MDSTR_LENGTH(l->md->key)) &&
|
| + !grpc_header_nonbin_value_is_legal(
|
| + grpc_mdstr_as_c_string(l->md->value),
|
| + GRPC_MDSTR_LENGTH(l->md->value))) {
|
| + gpr_log(GPR_ERROR, "attempt to send invalid metadata value");
|
| + return 0;
|
| + }
|
| + }
|
| + for (i = 1; i < count; i++) {
|
| + linked_from_md(&metadata[i])->prev = linked_from_md(&metadata[i - 1]);
|
| + }
|
| + for (i = 0; i < count - 1; i++) {
|
| + linked_from_md(&metadata[i])->next = linked_from_md(&metadata[i + 1]);
|
| + }
|
| + switch (prepend_extra_metadata * 2 + (count != 0)) {
|
| + case 0:
|
| + /* no prepend, no metadata => nothing to do */
|
| + batch->list.head = batch->list.tail = NULL;
|
| + break;
|
| + case 1:
|
| + /* metadata, but no prepend */
|
| + batch->list.head = linked_from_md(&metadata[0]);
|
| + batch->list.tail = linked_from_md(&metadata[count - 1]);
|
| + batch->list.head->prev = NULL;
|
| + batch->list.tail->next = NULL;
|
| + break;
|
| + case 2:
|
| + /* prepend, but no md */
|
| + batch->list.head = &call->send_extra_metadata[0];
|
| + batch->list.tail =
|
| + &call->send_extra_metadata[call->send_extra_metadata_count - 1];
|
| + batch->list.head->prev = NULL;
|
| + batch->list.tail->next = NULL;
|
| + break;
|
| + case 3:
|
| + /* prepend AND md */
|
| + batch->list.head = &call->send_extra_metadata[0];
|
| + call->send_extra_metadata[call->send_extra_metadata_count - 1].next =
|
| + linked_from_md(&metadata[0]);
|
| + linked_from_md(&metadata[0])->prev =
|
| + &call->send_extra_metadata[call->send_extra_metadata_count - 1];
|
| + batch->list.tail = linked_from_md(&metadata[count - 1]);
|
| + batch->list.head->prev = NULL;
|
| + batch->list.tail->next = NULL;
|
| + break;
|
| + default:
|
| + GPR_UNREACHABLE_CODE(return 0);
|
| + }
|
| +
|
| + return 1;
|
| +}
|
| +
|
| +void grpc_call_destroy(grpc_call *c) {
|
| + int cancel;
|
| + grpc_call *parent = c->parent;
|
| + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
|
| +
|
| + GPR_TIMER_BEGIN("grpc_call_destroy", 0);
|
| + GRPC_API_TRACE("grpc_call_destroy(c=%p)", 1, (c));
|
| +
|
| + 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(&exec_ctx, parent, "child");
|
| + }
|
| +
|
| + gpr_mu_lock(&c->mu);
|
| + GPR_ASSERT(!c->destroy_called);
|
| + c->destroy_called = 1;
|
| + if (c->have_alarm) {
|
| + grpc_timer_cancel(&exec_ctx, &c->alarm);
|
| + }
|
| + cancel = !c->received_final_op;
|
| + gpr_mu_unlock(&c->mu);
|
| + if (cancel) grpc_call_cancel(c, NULL);
|
| + GRPC_CALL_INTERNAL_UNREF(&exec_ctx, c, "destroy");
|
| + grpc_exec_ctx_finish(&exec_ctx);
|
| + GPR_TIMER_END("grpc_call_destroy", 0);
|
| +}
|
| +
|
| +grpc_call_error grpc_call_cancel(grpc_call *call, void *reserved) {
|
| + GRPC_API_TRACE("grpc_call_cancel(call=%p, reserved=%p)", 2, (call, reserved));
|
| + GPR_ASSERT(!reserved);
|
| + return grpc_call_cancel_with_status(call, GRPC_STATUS_CANCELLED, "Cancelled",
|
| + NULL);
|
| +}
|
| +
|
| +grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
|
| + grpc_status_code status,
|
| + const char *description,
|
| + void *reserved) {
|
| + grpc_call_error r;
|
| + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
|
| + GRPC_API_TRACE(
|
| + "grpc_call_cancel_with_status("
|
| + "c=%p, status=%d, description=%s, reserved=%p)",
|
| + 4, (c, (int)status, description, reserved));
|
| + GPR_ASSERT(reserved == NULL);
|
| + gpr_mu_lock(&c->mu);
|
| + r = cancel_with_status(&exec_ctx, c, status, description);
|
| + gpr_mu_unlock(&c->mu);
|
| + grpc_exec_ctx_finish(&exec_ctx);
|
| + return r;
|
| +}
|
| +
|
| +typedef struct cancel_closure {
|
| + grpc_closure closure;
|
| + grpc_call *call;
|
| + grpc_status_code status;
|
| +} cancel_closure;
|
| +
|
| +static void done_cancel(grpc_exec_ctx *exec_ctx, void *ccp, bool success) {
|
| + cancel_closure *cc = ccp;
|
| + GRPC_CALL_INTERNAL_UNREF(exec_ctx, cc->call, "cancel");
|
| + gpr_free(cc);
|
| +}
|
| +
|
| +static void send_cancel(grpc_exec_ctx *exec_ctx, void *ccp, bool success) {
|
| + grpc_transport_stream_op op;
|
| + cancel_closure *cc = ccp;
|
| + memset(&op, 0, sizeof(op));
|
| + op.cancel_with_status = cc->status;
|
| + /* reuse closure to catch completion */
|
| + grpc_closure_init(&cc->closure, done_cancel, cc);
|
| + op.on_complete = &cc->closure;
|
| + execute_op(exec_ctx, cc->call, &op);
|
| +}
|
| +
|
| +static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
|
| + grpc_status_code status,
|
| + const char *description) {
|
| + grpc_mdstr *details =
|
| + description ? grpc_mdstr_from_string(description) : NULL;
|
| + cancel_closure *cc = gpr_malloc(sizeof(*cc));
|
| +
|
| + GPR_ASSERT(status != GRPC_STATUS_OK);
|
| +
|
| + set_status_code(c, STATUS_FROM_API_OVERRIDE, (uint32_t)status);
|
| + set_status_details(c, STATUS_FROM_API_OVERRIDE, details);
|
| +
|
| + grpc_closure_init(&cc->closure, send_cancel, cc);
|
| + cc->call = c;
|
| + cc->status = status;
|
| + GRPC_CALL_INTERNAL_REF(c, "cancel");
|
| + grpc_exec_ctx_enqueue(exec_ctx, &cc->closure, true, NULL);
|
| +
|
| + return GRPC_CALL_OK;
|
| +}
|
| +
|
| +static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
|
| + grpc_transport_stream_op *op) {
|
| + grpc_call_element *elem;
|
| +
|
| + 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);
|
| +}
|
| +
|
| +char *grpc_call_get_peer(grpc_call *call) {
|
| + grpc_call_element *elem = CALL_ELEM_FROM_CALL(call, 0);
|
| + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
|
| + char *result;
|
| + GRPC_API_TRACE("grpc_call_get_peer(%p)", 1, (call));
|
| + result = elem->filter->get_peer(&exec_ctx, elem);
|
| + if (result == NULL) {
|
| + result = grpc_channel_get_target(call->channel);
|
| + }
|
| + if (result == NULL) {
|
| + result = gpr_strdup("unknown");
|
| + }
|
| + grpc_exec_ctx_finish(&exec_ctx);
|
| + return result;
|
| +}
|
| +
|
| +grpc_call *grpc_call_from_top_element(grpc_call_element *elem) {
|
| + return CALL_FROM_TOP_ELEM(elem);
|
| +}
|
| +
|
| +static void call_alarm(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
|
| + grpc_call *call = arg;
|
| + gpr_mu_lock(&call->mu);
|
| + call->have_alarm = 0;
|
| + if (success) {
|
| + cancel_with_status(exec_ctx, call, GRPC_STATUS_DEADLINE_EXCEEDED,
|
| + "Deadline Exceeded");
|
| + }
|
| + gpr_mu_unlock(&call->mu);
|
| + GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "alarm");
|
| +}
|
| +
|
| +static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call,
|
| + gpr_timespec deadline) {
|
| + if (call->have_alarm) {
|
| + gpr_log(GPR_ERROR, "Attempt to set deadline alarm twice");
|
| + assert(0);
|
| + return;
|
| + }
|
| + GRPC_CALL_INTERNAL_REF(call, "alarm");
|
| + call->have_alarm = 1;
|
| + call->send_deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
|
| + grpc_timer_init(exec_ctx, &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
|
| + as metadata cannot store a 0 value (which is used as OK for grpc_status_codes
|
| + */
|
| +#define STATUS_OFFSET 1
|
| +static void destroy_status(void *ignored) {}
|
| +
|
| +static uint32_t decode_status(grpc_mdelem *md) {
|
| + uint32_t status;
|
| + void *user_data;
|
| + if (md == GRPC_MDELEM_GRPC_STATUS_0) return 0;
|
| + if (md == GRPC_MDELEM_GRPC_STATUS_1) return 1;
|
| + if (md == GRPC_MDELEM_GRPC_STATUS_2) return 2;
|
| + user_data = grpc_mdelem_get_user_data(md, destroy_status);
|
| + if (user_data != NULL) {
|
| + status = ((uint32_t)(intptr_t)user_data) - STATUS_OFFSET;
|
| + } else {
|
| + if (!gpr_parse_bytes_to_uint32(grpc_mdstr_as_c_string(md->value),
|
| + GPR_SLICE_LENGTH(md->value->slice),
|
| + &status)) {
|
| + status = GRPC_STATUS_UNKNOWN; /* could not parse status code */
|
| + }
|
| + grpc_mdelem_set_user_data(md, destroy_status,
|
| + (void *)(intptr_t)(status + STATUS_OFFSET));
|
| + }
|
| + return status;
|
| +}
|
| +
|
| +static uint32_t decode_compression(grpc_mdelem *md) {
|
| + grpc_compression_algorithm algorithm =
|
| + grpc_compression_algorithm_from_mdstr(md->value);
|
| + if (algorithm == GRPC_COMPRESS_ALGORITHMS_COUNT) {
|
| + const char *md_c_str = grpc_mdstr_as_c_string(md->value);
|
| + gpr_log(GPR_ERROR, "Invalid compression algorithm: '%s'", md_c_str);
|
| + }
|
| + return algorithm;
|
| +}
|
| +
|
| +static grpc_mdelem *recv_common_filter(grpc_call *call, grpc_mdelem *elem) {
|
| + if (elem->key == GRPC_MDSTR_GRPC_STATUS) {
|
| + GPR_TIMER_BEGIN("status", 0);
|
| + set_status_code(call, STATUS_FROM_WIRE, decode_status(elem));
|
| + GPR_TIMER_END("status", 0);
|
| + return NULL;
|
| + } else if (elem->key == GRPC_MDSTR_GRPC_MESSAGE) {
|
| + GPR_TIMER_BEGIN("status-details", 0);
|
| + set_status_details(call, STATUS_FROM_WIRE, GRPC_MDSTR_REF(elem->value));
|
| + GPR_TIMER_END("status-details", 0);
|
| + return NULL;
|
| + }
|
| + return elem;
|
| +}
|
| +
|
| +static grpc_mdelem *publish_app_metadata(grpc_call *call, grpc_mdelem *elem,
|
| + int is_trailing) {
|
| + grpc_metadata_array *dest;
|
| + grpc_metadata *mdusr;
|
| + GPR_TIMER_BEGIN("publish_app_metadata", 0);
|
| + dest = call->buffered_metadata[is_trailing];
|
| + if (dest->count == dest->capacity) {
|
| + dest->capacity = GPR_MAX(dest->capacity + 8, dest->capacity * 2);
|
| + dest->metadata =
|
| + gpr_realloc(dest->metadata, sizeof(grpc_metadata) * dest->capacity);
|
| + }
|
| + mdusr = &dest->metadata[dest->count++];
|
| + mdusr->key = grpc_mdstr_as_c_string(elem->key);
|
| + mdusr->value = grpc_mdstr_as_c_string(elem->value);
|
| + mdusr->value_length = GPR_SLICE_LENGTH(elem->value->slice);
|
| + GPR_TIMER_END("publish_app_metadata", 0);
|
| + return elem;
|
| +}
|
| +
|
| +static grpc_mdelem *recv_initial_filter(void *callp, grpc_mdelem *elem) {
|
| + grpc_call *call = callp;
|
| + elem = recv_common_filter(call, elem);
|
| + if (elem == NULL) {
|
| + return NULL;
|
| + } else if (elem->key == GRPC_MDSTR_GRPC_ENCODING) {
|
| + GPR_TIMER_BEGIN("compression_algorithm", 0);
|
| + set_compression_algorithm(call, decode_compression(elem));
|
| + GPR_TIMER_END("compression_algorithm", 0);
|
| + return NULL;
|
| + } else if (elem->key == GRPC_MDSTR_GRPC_ACCEPT_ENCODING) {
|
| + GPR_TIMER_BEGIN("encodings_accepted_by_peer", 0);
|
| + set_encodings_accepted_by_peer(call, elem);
|
| + GPR_TIMER_END("encodings_accepted_by_peer", 0);
|
| + return NULL;
|
| + } else {
|
| + return publish_app_metadata(call, elem, 0);
|
| + }
|
| +}
|
| +
|
| +static grpc_mdelem *recv_trailing_filter(void *callp, grpc_mdelem *elem) {
|
| + grpc_call *call = callp;
|
| + elem = recv_common_filter(call, elem);
|
| + if (elem == NULL) {
|
| + return NULL;
|
| + } else {
|
| + return publish_app_metadata(call, elem, 1);
|
| + }
|
| +}
|
| +
|
| +grpc_call_stack *grpc_call_get_call_stack(grpc_call *call) {
|
| + return CALL_STACK_FROM_CALL(call);
|
| +}
|
| +
|
| +/*
|
| + * BATCH API IMPLEMENTATION
|
| + */
|
| +
|
| +static void set_status_value_directly(grpc_status_code status, void *dest) {
|
| + *(grpc_status_code *)dest = status;
|
| +}
|
| +
|
| +static void set_cancelled_value(grpc_status_code status, void *dest) {
|
| + *(int *)dest = (status != GRPC_STATUS_OK);
|
| +}
|
| +
|
| +static int 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 =
|
| + (GRPC_WRITE_USED_MASK | GRPC_WRITE_INTERNAL_USED_MASK);
|
| + const uint32_t invalid_positions = ~allowed_write_positions;
|
| + return !(flags & invalid_positions);
|
| +}
|
| +
|
| +static batch_control *allocate_batch_control(grpc_call *call) {
|
| + size_t i;
|
| + for (i = 0; i < MAX_CONCURRENT_BATCHES; i++) {
|
| + if ((call->used_batches & (1 << i)) == 0) {
|
| + call->used_batches = (uint8_t)(call->used_batches | (uint8_t)(1 << i));
|
| + return &call->active_batches[i];
|
| + }
|
| + }
|
| + return NULL;
|
| +}
|
| +
|
| +static void finish_batch_completion(grpc_exec_ctx *exec_ctx, void *user_data,
|
| + grpc_cq_completion *storage) {
|
| + batch_control *bctl = user_data;
|
| + grpc_call *call = bctl->call;
|
| + gpr_mu_lock(&call->mu);
|
| + call->used_batches = (uint8_t)(
|
| + call->used_batches & ~(uint8_t)(1 << (bctl - call->active_batches)));
|
| + gpr_mu_unlock(&call->mu);
|
| + GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "completion");
|
| +}
|
| +
|
| +static void post_batch_completion(grpc_exec_ctx *exec_ctx,
|
| + batch_control *bctl) {
|
| + grpc_call *call = bctl->call;
|
| + if (bctl->is_notify_tag_closure) {
|
| + grpc_exec_ctx_enqueue(exec_ctx, bctl->notify_tag, bctl->success, NULL);
|
| + gpr_mu_lock(&call->mu);
|
| + bctl->call->used_batches =
|
| + (uint8_t)(bctl->call->used_batches &
|
| + ~(uint8_t)(1 << (bctl - bctl->call->active_batches)));
|
| + gpr_mu_unlock(&call->mu);
|
| + GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "completion");
|
| + } else {
|
| + grpc_cq_end_op(exec_ctx, bctl->call->cq, bctl->notify_tag, bctl->success,
|
| + finish_batch_completion, bctl, &bctl->cq_completion);
|
| + }
|
| +}
|
| +
|
| +static void continue_receiving_slices(grpc_exec_ctx *exec_ctx,
|
| + batch_control *bctl) {
|
| + grpc_call *call = bctl->call;
|
| + for (;;) {
|
| + size_t remaining = call->receiving_stream->length -
|
| + (*call->receiving_buffer)->data.raw.slice_buffer.length;
|
| + if (remaining == 0) {
|
| + call->receiving_message = 0;
|
| + grpc_byte_stream_destroy(exec_ctx, call->receiving_stream);
|
| + call->receiving_stream = NULL;
|
| + if (gpr_unref(&bctl->steps_to_complete)) {
|
| + post_batch_completion(exec_ctx, bctl);
|
| + }
|
| + return;
|
| + }
|
| + if (grpc_byte_stream_next(exec_ctx, call->receiving_stream,
|
| + &call->receiving_slice, remaining,
|
| + &call->receiving_slice_ready)) {
|
| + gpr_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer,
|
| + call->receiving_slice);
|
| + } else {
|
| + return;
|
| + }
|
| + }
|
| +}
|
| +
|
| +static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
|
| + bool success) {
|
| + batch_control *bctl = bctlp;
|
| + grpc_call *call = bctl->call;
|
| +
|
| + if (success) {
|
| + gpr_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer,
|
| + call->receiving_slice);
|
| + continue_receiving_slices(exec_ctx, bctl);
|
| + } else {
|
| + grpc_byte_stream_destroy(exec_ctx, call->receiving_stream);
|
| + call->receiving_stream = NULL;
|
| + grpc_byte_buffer_destroy(*call->receiving_buffer);
|
| + *call->receiving_buffer = NULL;
|
| + if (gpr_unref(&bctl->steps_to_complete)) {
|
| + post_batch_completion(exec_ctx, bctl);
|
| + }
|
| + }
|
| +}
|
| +
|
| +static void process_data_after_md(grpc_exec_ctx *exec_ctx, batch_control *bctl,
|
| + bool success) {
|
| + grpc_call *call = bctl->call;
|
| + if (call->receiving_stream == NULL) {
|
| + *call->receiving_buffer = NULL;
|
| + call->receiving_message = 0;
|
| + if (gpr_unref(&bctl->steps_to_complete)) {
|
| + post_batch_completion(exec_ctx, bctl);
|
| + }
|
| + } else if (call->receiving_stream->length >
|
| + grpc_channel_get_max_message_length(call->channel)) {
|
| + cancel_with_status(exec_ctx, call, GRPC_STATUS_INTERNAL,
|
| + "Max message size exceeded");
|
| + grpc_byte_stream_destroy(exec_ctx, call->receiving_stream);
|
| + call->receiving_stream = NULL;
|
| + *call->receiving_buffer = NULL;
|
| + call->receiving_message = 0;
|
| + if (gpr_unref(&bctl->steps_to_complete)) {
|
| + post_batch_completion(exec_ctx, bctl);
|
| + }
|
| + } else {
|
| + call->test_only_last_message_flags = call->receiving_stream->flags;
|
| + if ((call->receiving_stream->flags & GRPC_WRITE_INTERNAL_COMPRESS) &&
|
| + (call->compression_algorithm > GRPC_COMPRESS_NONE)) {
|
| + *call->receiving_buffer = grpc_raw_compressed_byte_buffer_create(
|
| + NULL, 0, call->compression_algorithm);
|
| + } else {
|
| + *call->receiving_buffer = grpc_raw_byte_buffer_create(NULL, 0);
|
| + }
|
| + grpc_closure_init(&call->receiving_slice_ready, receiving_slice_ready,
|
| + bctl);
|
| + continue_receiving_slices(exec_ctx, bctl);
|
| + /* early out */
|
| + return;
|
| + }
|
| +}
|
| +
|
| +static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
|
| + bool success) {
|
| + batch_control *bctl = bctlp;
|
| + grpc_call *call = bctl->call;
|
| +
|
| + gpr_mu_lock(&bctl->call->mu);
|
| + if (bctl->call->has_initial_md_been_received) {
|
| + gpr_mu_unlock(&bctl->call->mu);
|
| + process_data_after_md(exec_ctx, bctlp, success);
|
| + } else {
|
| + call->saved_receiving_stream_ready_ctx.bctlp = bctlp;
|
| + call->saved_receiving_stream_ready_ctx.success = success;
|
| + gpr_mu_unlock(&bctl->call->mu);
|
| + }
|
| +}
|
| +
|
| +static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
|
| + void *bctlp, bool success) {
|
| + batch_control *bctl = bctlp;
|
| + grpc_call *call = bctl->call;
|
| +
|
| + gpr_mu_lock(&call->mu);
|
| +
|
| + grpc_metadata_batch *md =
|
| + &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */];
|
| + grpc_metadata_batch_filter(md, recv_initial_filter, call);
|
| + call->has_initial_md_been_received = true;
|
| +
|
| + if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) !=
|
| + 0 &&
|
| + !call->is_client) {
|
| + GPR_TIMER_BEGIN("set_deadline_alarm", 0);
|
| + set_deadline_alarm(exec_ctx, call, md->deadline);
|
| + GPR_TIMER_END("set_deadline_alarm", 0);
|
| + }
|
| +
|
| + if (call->saved_receiving_stream_ready_ctx.bctlp != NULL) {
|
| + grpc_closure *saved_rsr_closure = grpc_closure_create(
|
| + receiving_stream_ready, call->saved_receiving_stream_ready_ctx.bctlp);
|
| + grpc_exec_ctx_enqueue(exec_ctx, saved_rsr_closure,
|
| + call->saved_receiving_stream_ready_ctx.success, NULL);
|
| + call->saved_receiving_stream_ready_ctx.bctlp = NULL;
|
| + }
|
| +
|
| + gpr_mu_unlock(&call->mu);
|
| +
|
| + if (gpr_unref(&bctl->steps_to_complete)) {
|
| + post_batch_completion(exec_ctx, bctl);
|
| + }
|
| +}
|
| +
|
| +static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, bool success) {
|
| + batch_control *bctl = bctlp;
|
| + grpc_call *call = bctl->call;
|
| + grpc_call *child_call;
|
| + grpc_call *next_child_call;
|
| +
|
| + gpr_mu_lock(&call->mu);
|
| + if (bctl->send_initial_metadata) {
|
| + grpc_metadata_batch_destroy(
|
| + &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]);
|
| + }
|
| + if (bctl->send_message) {
|
| + call->sending_message = 0;
|
| + }
|
| + if (bctl->send_final_op) {
|
| + grpc_metadata_batch_destroy(
|
| + &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]);
|
| + }
|
| + if (bctl->recv_final_op) {
|
| + grpc_metadata_batch *md =
|
| + &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
|
| + grpc_metadata_batch_filter(md, recv_trailing_filter, call);
|
| +
|
| + if (call->have_alarm) {
|
| + grpc_timer_cancel(exec_ctx, &call->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, NULL);
|
| + GRPC_CALL_INTERNAL_UNREF(exec_ctx, child_call, "propagate_cancel");
|
| + }
|
| + child_call = next_child_call;
|
| + } while (child_call != call->first_child);
|
| + }
|
| +
|
| + if (call->is_client) {
|
| + get_final_status(call, set_status_value_directly,
|
| + call->final_op.client.status);
|
| + get_final_details(call, call->final_op.client.status_details,
|
| + call->final_op.client.status_details_capacity);
|
| + } else {
|
| + get_final_status(call, set_cancelled_value,
|
| + call->final_op.server.cancelled);
|
| + }
|
| +
|
| + success = 1;
|
| + }
|
| + bctl->success = success != 0;
|
| + gpr_mu_unlock(&call->mu);
|
| + if (gpr_unref(&bctl->steps_to_complete)) {
|
| + post_batch_completion(exec_ctx, bctl);
|
| + }
|
| +}
|
| +
|
| +static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
|
| + grpc_call *call, const grpc_op *ops,
|
| + size_t nops, void *notify_tag,
|
| + int is_notify_tag_closure) {
|
| + grpc_transport_stream_op stream_op;
|
| + size_t i;
|
| + const grpc_op *op;
|
| + batch_control *bctl;
|
| + int num_completion_callbacks_needed = 1;
|
| + grpc_call_error error = GRPC_CALL_OK;
|
| +
|
| + GPR_TIMER_BEGIN("grpc_call_start_batch", 0);
|
| +
|
| + GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, notify_tag);
|
| +
|
| + memset(&stream_op, 0, sizeof(stream_op));
|
| +
|
| + /* TODO(ctiller): this feels like it could be made lock-free */
|
| + gpr_mu_lock(&call->mu);
|
| + bctl = allocate_batch_control(call);
|
| + memset(bctl, 0, sizeof(*bctl));
|
| + bctl->call = call;
|
| + bctl->notify_tag = notify_tag;
|
| + bctl->is_notify_tag_closure = (uint8_t)(is_notify_tag_closure != 0);
|
| +
|
| + if (nops == 0) {
|
| + GRPC_CALL_INTERNAL_REF(call, "completion");
|
| + bctl->success = 1;
|
| + if (!is_notify_tag_closure) {
|
| + grpc_cq_begin_op(call->cq, notify_tag);
|
| + }
|
| + gpr_mu_unlock(&call->mu);
|
| + post_batch_completion(exec_ctx, bctl);
|
| + error = GRPC_CALL_OK;
|
| + goto done;
|
| + }
|
| +
|
| + /* rewrite batch ops into a transport op */
|
| + for (i = 0; i < nops; i++) {
|
| + op = &ops[i];
|
| + if (op->reserved != NULL) {
|
| + error = GRPC_CALL_ERROR;
|
| + goto done_with_error;
|
| + }
|
| + switch (op->op) {
|
| + case GRPC_OP_SEND_INITIAL_METADATA:
|
| + /* Flag validation: currently allow no flags */
|
| + if (op->flags != 0) {
|
| + error = GRPC_CALL_ERROR_INVALID_FLAGS;
|
| + goto done_with_error;
|
| + }
|
| + if (call->sent_initial_metadata) {
|
| + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
|
| + goto done_with_error;
|
| + }
|
| + if (op->data.send_initial_metadata.count > INT_MAX) {
|
| + error = GRPC_CALL_ERROR_INVALID_METADATA;
|
| + goto done_with_error;
|
| + }
|
| + bctl->send_initial_metadata = 1;
|
| + call->sent_initial_metadata = 1;
|
| + if (!prepare_application_metadata(
|
| + call, (int)op->data.send_initial_metadata.count,
|
| + op->data.send_initial_metadata.metadata, 0, call->is_client)) {
|
| + error = GRPC_CALL_ERROR_INVALID_METADATA;
|
| + goto done_with_error;
|
| + }
|
| + /* TODO(ctiller): just make these the same variable? */
|
| + call->metadata_batch[0][0].deadline = call->send_deadline;
|
| + stream_op.send_initial_metadata =
|
| + &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */];
|
| + break;
|
| + case GRPC_OP_SEND_MESSAGE:
|
| + if (!are_write_flags_valid(op->flags)) {
|
| + error = GRPC_CALL_ERROR_INVALID_FLAGS;
|
| + goto done_with_error;
|
| + }
|
| + if (op->data.send_message == NULL) {
|
| + error = GRPC_CALL_ERROR_INVALID_MESSAGE;
|
| + goto done_with_error;
|
| + }
|
| + if (call->sending_message) {
|
| + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
|
| + goto done_with_error;
|
| + }
|
| + bctl->send_message = 1;
|
| + call->sending_message = 1;
|
| + grpc_slice_buffer_stream_init(
|
| + &call->sending_stream,
|
| + &op->data.send_message->data.raw.slice_buffer, op->flags);
|
| + stream_op.send_message = &call->sending_stream.base;
|
| + break;
|
| + case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
|
| + /* Flag validation: currently allow no flags */
|
| + if (op->flags != 0) {
|
| + error = GRPC_CALL_ERROR_INVALID_FLAGS;
|
| + goto done_with_error;
|
| + }
|
| + if (!call->is_client) {
|
| + error = GRPC_CALL_ERROR_NOT_ON_SERVER;
|
| + goto done_with_error;
|
| + }
|
| + if (call->sent_final_op) {
|
| + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
|
| + goto done_with_error;
|
| + }
|
| + bctl->send_final_op = 1;
|
| + call->sent_final_op = 1;
|
| + stream_op.send_trailing_metadata =
|
| + &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */];
|
| + break;
|
| + case GRPC_OP_SEND_STATUS_FROM_SERVER:
|
| + /* Flag validation: currently allow no flags */
|
| + if (op->flags != 0) {
|
| + error = GRPC_CALL_ERROR_INVALID_FLAGS;
|
| + goto done_with_error;
|
| + }
|
| + if (call->is_client) {
|
| + error = GRPC_CALL_ERROR_NOT_ON_CLIENT;
|
| + goto done_with_error;
|
| + }
|
| + if (call->sent_final_op) {
|
| + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
|
| + goto done_with_error;
|
| + }
|
| + if (op->data.send_status_from_server.trailing_metadata_count >
|
| + INT_MAX) {
|
| + error = GRPC_CALL_ERROR_INVALID_METADATA;
|
| + goto done_with_error;
|
| + }
|
| + bctl->send_final_op = 1;
|
| + call->sent_final_op = 1;
|
| + 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);
|
| + if (op->data.send_status_from_server.status_details != NULL) {
|
| + call->send_extra_metadata[1].md = grpc_mdelem_from_metadata_strings(
|
| + GRPC_MDSTR_GRPC_MESSAGE,
|
| + grpc_mdstr_from_string(
|
| + op->data.send_status_from_server.status_details));
|
| + call->send_extra_metadata_count++;
|
| + set_status_details(
|
| + call, STATUS_FROM_API_OVERRIDE,
|
| + GRPC_MDSTR_REF(call->send_extra_metadata[1].md->value));
|
| + }
|
| + set_status_code(call, STATUS_FROM_API_OVERRIDE,
|
| + (uint32_t)op->data.send_status_from_server.status);
|
| + if (!prepare_application_metadata(
|
| + call,
|
| + (int)op->data.send_status_from_server.trailing_metadata_count,
|
| + op->data.send_status_from_server.trailing_metadata, 1, 1)) {
|
| + error = GRPC_CALL_ERROR_INVALID_METADATA;
|
| + goto done_with_error;
|
| + }
|
| + stream_op.send_trailing_metadata =
|
| + &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */];
|
| + break;
|
| + case GRPC_OP_RECV_INITIAL_METADATA:
|
| + /* Flag validation: currently allow no flags */
|
| + if (op->flags != 0) {
|
| + error = GRPC_CALL_ERROR_INVALID_FLAGS;
|
| + goto done_with_error;
|
| + }
|
| + if (call->received_initial_metadata) {
|
| + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
|
| + goto done_with_error;
|
| + }
|
| + call->received_initial_metadata = 1;
|
| + call->buffered_metadata[0] = op->data.recv_initial_metadata;
|
| + grpc_closure_init(&call->receiving_initial_metadata_ready,
|
| + receiving_initial_metadata_ready, bctl);
|
| + bctl->recv_initial_metadata = 1;
|
| + stream_op.recv_initial_metadata =
|
| + &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */];
|
| + stream_op.recv_initial_metadata_ready =
|
| + &call->receiving_initial_metadata_ready;
|
| + num_completion_callbacks_needed++;
|
| + break;
|
| + case GRPC_OP_RECV_MESSAGE:
|
| + /* Flag validation: currently allow no flags */
|
| + if (op->flags != 0) {
|
| + error = GRPC_CALL_ERROR_INVALID_FLAGS;
|
| + goto done_with_error;
|
| + }
|
| + if (call->receiving_message) {
|
| + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
|
| + goto done_with_error;
|
| + }
|
| + call->receiving_message = 1;
|
| + bctl->recv_message = 1;
|
| + call->receiving_buffer = op->data.recv_message;
|
| + stream_op.recv_message = &call->receiving_stream;
|
| + grpc_closure_init(&call->receiving_stream_ready, receiving_stream_ready,
|
| + bctl);
|
| + stream_op.recv_message_ready = &call->receiving_stream_ready;
|
| + num_completion_callbacks_needed++;
|
| + break;
|
| + case GRPC_OP_RECV_STATUS_ON_CLIENT:
|
| + /* Flag validation: currently allow no flags */
|
| + if (op->flags != 0) {
|
| + error = GRPC_CALL_ERROR_INVALID_FLAGS;
|
| + goto done_with_error;
|
| + }
|
| + if (!call->is_client) {
|
| + error = GRPC_CALL_ERROR_NOT_ON_SERVER;
|
| + goto done_with_error;
|
| + }
|
| + if (call->received_final_op) {
|
| + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
|
| + goto done_with_error;
|
| + }
|
| + call->received_final_op = 1;
|
| + call->buffered_metadata[1] =
|
| + op->data.recv_status_on_client.trailing_metadata;
|
| + 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;
|
| + call->final_op.client.status_details_capacity =
|
| + op->data.recv_status_on_client.status_details_capacity;
|
| + bctl->recv_final_op = 1;
|
| + stream_op.recv_trailing_metadata =
|
| + &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
|
| + break;
|
| + case GRPC_OP_RECV_CLOSE_ON_SERVER:
|
| + /* Flag validation: currently allow no flags */
|
| + if (op->flags != 0) {
|
| + error = GRPC_CALL_ERROR_INVALID_FLAGS;
|
| + goto done_with_error;
|
| + }
|
| + if (call->is_client) {
|
| + error = GRPC_CALL_ERROR_NOT_ON_CLIENT;
|
| + goto done_with_error;
|
| + }
|
| + if (call->received_final_op) {
|
| + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS;
|
| + goto done_with_error;
|
| + }
|
| + call->received_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 =
|
| + &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
|
| + break;
|
| + }
|
| + }
|
| +
|
| + GRPC_CALL_INTERNAL_REF(call, "completion");
|
| + if (!is_notify_tag_closure) {
|
| + grpc_cq_begin_op(call->cq, notify_tag);
|
| + }
|
| + 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);
|
| + stream_op.on_complete = &bctl->finish_batch;
|
| + gpr_mu_unlock(&call->mu);
|
| +
|
| + execute_op(exec_ctx, call, &stream_op);
|
| +
|
| +done:
|
| + GPR_TIMER_END("grpc_call_start_batch", 0);
|
| + return error;
|
| +
|
| +done_with_error:
|
| + /* reverse any mutations that occured */
|
| + if (bctl->send_initial_metadata) {
|
| + call->sent_initial_metadata = 0;
|
| + grpc_metadata_batch_clear(&call->metadata_batch[0][0]);
|
| + }
|
| + if (bctl->send_message) {
|
| + call->sending_message = 0;
|
| + grpc_byte_stream_destroy(exec_ctx, &call->sending_stream.base);
|
| + }
|
| + if (bctl->send_final_op) {
|
| + call->sent_final_op = 0;
|
| + grpc_metadata_batch_clear(&call->metadata_batch[0][1]);
|
| + }
|
| + if (bctl->recv_initial_metadata) {
|
| + call->received_initial_metadata = 0;
|
| + }
|
| + if (bctl->recv_message) {
|
| + call->receiving_message = 0;
|
| + }
|
| + if (bctl->recv_final_op) {
|
| + call->received_final_op = 0;
|
| + }
|
| + gpr_mu_unlock(&call->mu);
|
| + goto done;
|
| +}
|
| +
|
| +grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
|
| + size_t nops, void *tag, void *reserved) {
|
| + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
|
| + grpc_call_error err;
|
| +
|
| + GRPC_API_TRACE(
|
| + "grpc_call_start_batch(call=%p, ops=%p, nops=%lu, tag=%p, reserved=%p)",
|
| + 5, (call, ops, (unsigned long)nops, tag, reserved));
|
| +
|
| + if (reserved != NULL) {
|
| + err = GRPC_CALL_ERROR;
|
| + } else {
|
| + err = call_start_batch(&exec_ctx, call, ops, nops, tag, 0);
|
| + }
|
| +
|
| + grpc_exec_ctx_finish(&exec_ctx);
|
| + return err;
|
| +}
|
| +
|
| +grpc_call_error grpc_call_start_batch_and_execute(grpc_exec_ctx *exec_ctx,
|
| + grpc_call *call,
|
| + const grpc_op *ops,
|
| + size_t nops,
|
| + grpc_closure *closure) {
|
| + return call_start_batch(exec_ctx, call, ops, nops, closure, 1);
|
| +}
|
| +
|
| +void grpc_call_context_set(grpc_call *call, grpc_context_index elem,
|
| + void *value, void (*destroy)(void *value)) {
|
| + if (call->context[elem].destroy) {
|
| + call->context[elem].destroy(call->context[elem].value);
|
| + }
|
| + call->context[elem].value = value;
|
| + call->context[elem].destroy = destroy;
|
| +}
|
| +
|
| +void *grpc_call_context_get(grpc_call *call, grpc_context_index elem) {
|
| + return call->context[elem].value;
|
| +}
|
| +
|
| +uint8_t grpc_call_is_client(grpc_call *call) { return call->is_client; }
|
|
|