diff options
Diffstat (limited to 'src/core/surface')
| -rw-r--r-- | src/core/surface/byte_buffer_queue.c | 97 | ||||
| -rw-r--r-- | src/core/surface/call.c | 1936 | ||||
| -rw-r--r-- | src/core/surface/call.h | 81 | ||||
| -rw-r--r-- | src/core/surface/call_log_batch.c | 27 | ||||
| -rw-r--r-- | src/core/surface/channel.c | 181 | ||||
| -rw-r--r-- | src/core/surface/channel.h | 13 | ||||
| -rw-r--r-- | src/core/surface/channel_connectivity.c | 41 | ||||
| -rw-r--r-- | src/core/surface/channel_create.c | 43 | ||||
| -rw-r--r-- | src/core/surface/channel_ping.c (renamed from src/core/surface/byte_buffer_queue.h) | 67 | ||||
| -rw-r--r-- | src/core/surface/completion_queue.c | 132 | ||||
| -rw-r--r-- | src/core/surface/completion_queue.h | 11 | ||||
| -rw-r--r-- | src/core/surface/init.c | 14 | ||||
| -rw-r--r-- | src/core/surface/lame_client.c | 82 | ||||
| -rw-r--r-- | src/core/surface/secure_channel_create.c | 64 | ||||
| -rw-r--r-- | src/core/surface/server.c | 251 | ||||
| -rw-r--r-- | src/core/surface/server.h | 2 | ||||
| -rw-r--r-- | src/core/surface/server_chttp2.c | 19 | ||||
| -rw-r--r-- | src/core/surface/server_create.c | 14 |
18 files changed, 1186 insertions, 1889 deletions
diff --git a/src/core/surface/byte_buffer_queue.c b/src/core/surface/byte_buffer_queue.c deleted file mode 100644 index e47dc4f4ce..0000000000 --- a/src/core/surface/byte_buffer_queue.c +++ /dev/null @@ -1,97 +0,0 @@ -/* - * - * Copyright 2015, 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 "src/core/surface/byte_buffer_queue.h" -#include <grpc/support/alloc.h> -#include <grpc/support/useful.h> - -static void bba_destroy(grpc_bbq_array *array, size_t start_pos) { - size_t i; - for (i = start_pos; i < array->count; i++) { - grpc_byte_buffer_destroy(array->data[i]); - } - gpr_free(array->data); -} - -/* Append an operation to an array, expanding as needed */ -static void bba_push(grpc_bbq_array *a, grpc_byte_buffer *buffer) { - if (a->count == a->capacity) { - a->capacity = GPR_MAX(a->capacity * 2, 8); - a->data = gpr_realloc(a->data, sizeof(grpc_byte_buffer *) * a->capacity); - } - a->data[a->count++] = buffer; -} - -void grpc_bbq_destroy(grpc_byte_buffer_queue *q) { - bba_destroy(&q->filling, 0); - bba_destroy(&q->draining, q->drain_pos); -} - -int grpc_bbq_empty(grpc_byte_buffer_queue *q) { - return (q->drain_pos == q->draining.count && q->filling.count == 0); -} - -void grpc_bbq_push(grpc_byte_buffer_queue *q, grpc_byte_buffer *buffer) { - q->bytes += grpc_byte_buffer_length(buffer); - bba_push(&q->filling, buffer); -} - -void grpc_bbq_flush(grpc_byte_buffer_queue *q) { - grpc_byte_buffer *bb; - while ((bb = grpc_bbq_pop(q))) { - grpc_byte_buffer_destroy(bb); - } -} - -size_t grpc_bbq_bytes(grpc_byte_buffer_queue *q) { return q->bytes; } - -grpc_byte_buffer *grpc_bbq_pop(grpc_byte_buffer_queue *q) { - grpc_bbq_array temp_array; - grpc_byte_buffer *out; - - if (q->drain_pos == q->draining.count) { - if (q->filling.count == 0) { - return NULL; - } - q->draining.count = 0; - q->drain_pos = 0; - /* swap arrays */ - temp_array = q->filling; - q->filling = q->draining; - q->draining = temp_array; - } - - out = q->draining.data[q->drain_pos++]; - q->bytes -= grpc_byte_buffer_length(out); - return out; -} diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 81ff215c0c..a162d99193 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -31,6 +31,7 @@ * */ #include <assert.h> +#include <limits.h> #include <stdio.h> #include <stdlib.h> #include <string.h> @@ -42,16 +43,17 @@ #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/byte_buffer_queue.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 completions possible. +/** The maximum number of concurrent batches possible. Based upon the maximum number of individually queueable ops in the batch api: - initial metadata send @@ -60,7 +62,7 @@ - initial metadata recv - message recv - status/close recv (depending on client/server) */ -#define MAX_CONCURRENT_COMPLETIONS 6 +#define MAX_CONCURRENT_BATCHES 6 typedef struct { grpc_ioreq_completion_func on_complete; @@ -68,24 +70,7 @@ typedef struct { int success; } completed_request; -/* See request_set in grpc_call below for a description */ -#define REQSET_EMPTY 'X' -#define REQSET_DONE 'Y' - -#define MAX_SEND_INITIAL_METADATA_COUNT 3 - -typedef struct { - /* Overall status of the operation: starts OK, may degrade to - non-OK */ - gpr_uint8 success; - /* a bit mask of which request ops are needed (1u << opid) */ - gpr_uint16 need_mask; - /* a bit mask of which request ops are now completed */ - gpr_uint16 complete_mask; - /* Completion function to call at the end of the operation */ - grpc_ioreq_completion_func on_complete; - void *user_data; -} reqinfo_master; +#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 @@ -130,106 +115,63 @@ typedef enum { 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; + + gpr_uint8 send_initial_metadata; + gpr_uint8 send_message; + gpr_uint8 send_final_op; + gpr_uint8 recv_initial_metadata; + gpr_uint8 recv_message; + gpr_uint8 recv_final_op; + gpr_uint8 is_notify_tag_closure; + gpr_uint8 success; +} batch_control; + struct grpc_call { grpc_completion_queue *cq; grpc_channel *channel; grpc_call *parent; grpc_call *first_child; - grpc_mdctx *metadata_context; /* TODO(ctiller): share with cq if possible? */ gpr_mu mu; - gpr_mu completion_mu; - /* how far through the stream have we read? */ - read_state read_state; - /* how far through the stream have we written? */ - write_state write_state; /* client or server call */ gpr_uint8 is_client; /* is the alarm set */ gpr_uint8 have_alarm; - /* are we currently performing a send operation */ - gpr_uint8 sending; - /* are we currently performing a recv operation */ - gpr_uint8 receiving; - /* are we currently completing requests */ - gpr_uint8 completing; /** has grpc_call_destroy been called */ gpr_uint8 destroy_called; - /* pairs with completed_requests */ - gpr_uint8 num_completed_requests; - /* are we currently reading a message? */ - gpr_uint8 reading_message; - /* have we bound a pollset yet? */ - gpr_uint8 bound_pollset; - /* is an error status set */ - gpr_uint8 error_status_set; - /** bitmask of allocated completion events in completions */ - gpr_uint8 allocated_completions; /** flag indicating that cancellation is inherited */ gpr_uint8 cancellation_is_inherited; + /** bitmask of live batches */ + gpr_uint8 used_batches; + /** which ops are in-flight */ + gpr_uint8 sent_initial_metadata; + gpr_uint8 sending_message; + gpr_uint8 sent_final_op; + gpr_uint8 received_initial_metadata; + gpr_uint8 receiving_message; + gpr_uint8 received_final_op; + + batch_control active_batches[MAX_CONCURRENT_BATCHES]; + + /* first idx: is_receiving, second idx: is_trailing */ + grpc_metadata_batch metadata_batch[2][2]; - /* flags with bits corresponding to write states allowing us to determine - what was sent */ - gpr_uint16 last_send_contains; - /* cancel with this status on the next outgoing transport op */ - grpc_status_code cancel_with_status; - - /* Active ioreqs. - request_set and request_data contain one element per active ioreq - operation. - - request_set[op] is an integer specifying a set of operations to which - the request belongs: - - if it is < GRPC_IOREQ_OP_COUNT, then this operation is pending - completion, and the integer represents to which group of operations - the ioreq belongs. Each group is represented by one master, and the - integer in request_set is an index into masters to find the master - data. - - if it is REQSET_EMPTY, the ioreq op is inactive and available to be - started - - finally, if request_set[op] is REQSET_DONE, then the operation is - complete and unavailable to be started again - - request_data[op] is the request data as supplied by the initiator of - a request, and is valid iff request_set[op] <= GRPC_IOREQ_OP_COUNT. - The set fields are as per the request type specified by op. - - Finally, one element of masters is set per active _set_ of ioreq - operations. It describes work left outstanding, result status, and - what work to perform upon operation completion. As one ioreq of each - op type can be active at once, by convention we choose the first element - of the group to be the master -- ie the master of in-progress operation - op is masters[request_set[op]]. This allows constant time allocation - and a strong upper bound of a count of masters to be calculated. */ - gpr_uint8 request_set[GRPC_IOREQ_OP_COUNT]; - grpc_ioreq_data request_data[GRPC_IOREQ_OP_COUNT]; - gpr_uint32 request_flags[GRPC_IOREQ_OP_COUNT]; - reqinfo_master masters[GRPC_IOREQ_OP_COUNT]; - - /* Dynamic array of ioreq's that have completed: the count of - elements is queued in num_completed_requests. - This list is built up under lock(), and flushed entirely during - unlock(). - We know the upper bound of the number of elements as we can only - have one ioreq of each type active at once. */ - completed_request completed_requests[GRPC_IOREQ_OP_COUNT]; - /* Incoming buffer of messages */ - grpc_byte_buffer_queue incoming_queue; /* 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]; - /* All metadata received - unreffed at once at the end of the call */ - grpc_mdelem **owned_metadata; - size_t owned_metadata_count; - size_t owned_metadata_capacity; + 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 */ gpr_uint32 encodings_accepted_by_peer; @@ -239,35 +181,36 @@ struct grpc_call { /* Deadline alarm - if have_alarm is non-zero */ grpc_timer alarm; - /* Call refcount - to keep the call alive during asynchronous operations */ - gpr_refcount internal_refcount; - - grpc_linked_mdelem send_initial_metadata[MAX_SEND_INITIAL_METADATA_COUNT]; - grpc_linked_mdelem status_link; - grpc_linked_mdelem details_link; - size_t send_initial_metadata_count; + /* 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; - grpc_stream_op_buffer send_ops; - grpc_stream_op_buffer recv_ops; - grpc_stream_state recv_state; - - gpr_slice_buffer incoming_message; - gpr_uint32 incoming_message_length; - gpr_uint32 incoming_message_flags; - grpc_closure destroy_closure; - grpc_closure on_done_recv; - grpc_closure on_done_send; - grpc_closure on_done_bind; - - /** completion events - for completion queue use */ - grpc_cq_completion completions[MAX_CONCURRENT_COMPLETIONS]; - /** siblings: children of the same parent form a list, and this list is protected under parent->mu */ grpc_call *sibling_next; grpc_call *sibling_prev; + + 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; + gpr_uint32 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; }; #define CALL_STACK_FROM_CALL(call) ((grpc_call_stack *)((call) + 1)) @@ -279,20 +222,15 @@ struct grpc_call { static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call, gpr_timespec deadline); -static void call_on_done_recv(grpc_exec_ctx *exec_ctx, void *call, int success); -static void call_on_done_send(grpc_exec_ctx *exec_ctx, void *call, int success); -static int fill_send_ops(grpc_call *call, grpc_transport_stream_op *op); static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call, grpc_transport_stream_op *op); -static void recv_metadata(grpc_exec_ctx *exec_ctx, grpc_call *call, - grpc_metadata_batch *metadata); -static void finish_read_ops(grpc_call *call); -static grpc_call_error cancel_with_status(grpc_call *c, grpc_status_code status, +static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c, + grpc_status_code status, const char *description); -static void finished_loose_op(grpc_exec_ctx *exec_ctx, void *call, int success); - -static void lock(grpc_call *call); -static void unlock(grpc_exec_ctx *exec_ctx, grpc_call *call); +static void destroy_call(grpc_exec_ctx *exec_ctx, void *call_stack, + int success); +static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp, + int success); grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call, gpr_uint32 propagation_mask, @@ -301,9 +239,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call, grpc_mdelem **add_initial_metadata, size_t add_initial_metadata_count, gpr_timespec send_deadline) { - size_t i; - grpc_transport_stream_op initial_op; - grpc_transport_stream_op *initial_op_ptr = NULL; + 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; @@ -311,51 +247,36 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call, call = gpr_malloc(sizeof(grpc_call) + channel_stack->call_stack_size); memset(call, 0, sizeof(grpc_call)); gpr_mu_init(&call->mu); - gpr_mu_init(&call->completion_mu); call->channel = channel; call->cq = cq; - if (cq != NULL) { - GRPC_CQ_INTERNAL_REF(cq, "bind"); - } call->parent = parent_call; call->is_client = server_transport_data == NULL; - for (i = 0; i < GRPC_IOREQ_OP_COUNT; i++) { - call->request_set[i] = REQSET_EMPTY; - } if (call->is_client) { - call->request_set[GRPC_IOREQ_SEND_TRAILING_METADATA] = REQSET_DONE; - call->request_set[GRPC_IOREQ_SEND_STATUS] = REQSET_DONE; + 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; } - GPR_ASSERT(add_initial_metadata_count < MAX_SEND_INITIAL_METADATA_COUNT); - for (i = 0; i < add_initial_metadata_count; i++) { - call->send_initial_metadata[i].md = add_initial_metadata[i]; + 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_initial_metadata_count = add_initial_metadata_count; call->send_deadline = send_deadline; GRPC_CHANNEL_INTERNAL_REF(channel, "call"); - call->metadata_context = grpc_channel_get_metadata_context(channel); - grpc_sopb_init(&call->send_ops); - grpc_sopb_init(&call->recv_ops); - gpr_slice_buffer_init(&call->incoming_message); - grpc_closure_init(&call->on_done_recv, call_on_done_recv, call); - grpc_closure_init(&call->on_done_send, call_on_done_send, call); - grpc_closure_init(&call->on_done_bind, finished_loose_op, call); - /* dropped in destroy and when READ_STATE_STREAM_CLOSED received */ - gpr_ref_init(&call->internal_refcount, 2); - /* server hack: start reads immediately so we can get initial metadata. - TODO(ctiller): figure out a cleaner solution */ - if (!call->is_client) { - memset(&initial_op, 0, sizeof(initial_op)); - initial_op.recv_ops = &call->recv_ops; - initial_op.recv_state = &call->recv_state; - initial_op.on_done_recv = &call->on_done_recv; - initial_op.context = call->context; - call->receiving = 1; - GRPC_CALL_INTERNAL_REF(call, "receiving"); - initial_op_ptr = &initial_op; + /* 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)); } - grpc_call_stack_init(&exec_ctx, channel_stack, server_transport_data, - initial_op_ptr, CALL_STACK_FROM_CALL(call)); if (parent_call != NULL) { GRPC_CALL_INTERNAL_REF(parent_call, "child"); GPR_ASSERT(call->is_client); @@ -408,90 +329,55 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call, void grpc_call_set_completion_queue(grpc_exec_ctx *exec_ctx, grpc_call *call, grpc_completion_queue *cq) { - lock(call); + GPR_ASSERT(cq); call->cq = cq; - if (cq) { - GRPC_CQ_INTERNAL_REF(cq, "bind"); - } - unlock(exec_ctx, call); + GRPC_CQ_INTERNAL_REF(cq, "bind"); + grpc_call_stack_set_pollset(exec_ctx, CALL_STACK_FROM_CALL(call), + grpc_cq_pollset(cq)); } -grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) { - return call->cq; -} - -static grpc_cq_completion *allocate_completion(grpc_call *call) { - gpr_uint8 i; - gpr_mu_lock(&call->completion_mu); - for (i = 0; i < GPR_ARRAY_SIZE(call->completions); i++) { - if (call->allocated_completions & (1u << i)) { - continue; - } - /* NB: the following integer arithmetic operation needs to be in its - * expanded form due to the "integral promotion" performed (see section - * 3.2.1.1 of the C89 draft standard). A cast to the smaller container type - * is then required to avoid the compiler warning */ - call->allocated_completions = - (gpr_uint8)(call->allocated_completions | (1u << i)); - gpr_mu_unlock(&call->completion_mu); - return &call->completions[i]; - } - GPR_UNREACHABLE_CODE(return NULL); - return NULL; -} - -static void done_completion(grpc_exec_ctx *exec_ctx, void *call, - grpc_cq_completion *completion) { - grpc_call *c = call; - gpr_mu_lock(&c->completion_mu); - c->allocated_completions &= - (gpr_uint8) ~(1u << (completion - c->completions)); - gpr_mu_unlock(&c->completion_mu); - GRPC_CALL_INTERNAL_UNREF(exec_ctx, c, "completion"); -} - -#ifdef GRPC_CALL_REF_COUNT_DEBUG -void grpc_call_internal_ref(grpc_call *c, const char *reason) { - gpr_log(GPR_DEBUG, "CALL: ref %p %d -> %d [%s]", c, - c->internal_refcount.count, c->internal_refcount.count + 1, reason); +#ifdef GRPC_STREAM_REFCOUNT_DEBUG +#define REF_REASON reason +#define REF_ARG , const char *reason #else -void grpc_call_internal_ref(grpc_call *c) { +#define REF_REASON "" +#define REF_ARG #endif - gpr_ref(&c->internal_refcount); +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, grpc_call *call) { +static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, int 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(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); - gpr_mu_destroy(&c->completion_mu); for (i = 0; i < STATUS_SOURCE_COUNT; i++) { if (c->status[i].details) { GRPC_MDSTR_UNREF(c->status[i].details); } } - for (i = 0; i < c->owned_metadata_count; i++) { - GRPC_MDELEM_UNREF(c->owned_metadata[i]); - } - gpr_free(c->owned_metadata); - for (i = 0; i < GPR_ARRAY_SIZE(c->buffered_metadata); i++) { - gpr_free(c->buffered_metadata[i].metadata); - } - for (i = 0; i < c->send_initial_metadata_count; i++) { - GRPC_MDELEM_UNREF(c->send_initial_metadata[i].md); + 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); } } - grpc_sopb_destroy(&c->send_ops); - grpc_sopb_destroy(&c->recv_ops); - grpc_bbq_destroy(&c->incoming_queue); - gpr_slice_buffer_destroy(&c->incoming_message); if (c->cq) { GRPC_CQ_INTERNAL_UNREF(c->cq, "bind"); } @@ -499,30 +385,14 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, grpc_call *call) { GPR_TIMER_END("destroy_call", 0); } -#ifdef GRPC_CALL_REF_COUNT_DEBUG -void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c, - const char *reason) { - gpr_log(GPR_DEBUG, "CALL: unref %p %d -> %d [%s]", c, - c->internal_refcount.count, c->internal_refcount.count - 1, reason); -#else -void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c) { -#endif - if (gpr_unref(&c->internal_refcount)) { - destroy_call(exec_ctx, c); - } -} - static void set_status_code(grpc_call *call, status_source source, gpr_uint32 status) { if (call->status[source].is_set) return; call->status[source].is_set = 1; call->status[source].code = (grpc_status_code)status; - call->error_status_set = status != GRPC_STATUS_OK; - if (status != GRPC_STATUS_OK && !grpc_bbq_empty(&call->incoming_queue)) { - grpc_bbq_flush(&call->incoming_queue); - } + /* TODO(ctiller): what to do about the flush that was previously here */ } static void set_compression_algorithm(grpc_call *call, @@ -539,6 +409,14 @@ grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm( return algorithm; } +gpr_uint32 grpc_call_test_only_get_message_flags(grpc_call *call) { + gpr_uint32 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) { @@ -596,14 +474,6 @@ gpr_uint32 grpc_call_test_only_get_encodings_accepted_by_peer(grpc_call *call) { return encodings_accepted_by_peer; } -gpr_uint32 grpc_call_test_only_get_message_flags(grpc_call *call) { - gpr_uint32 flags; - gpr_mu_lock(&call->mu); - flags = call->incoming_message_flags; - gpr_mu_unlock(&call->mu); - return flags; -} - static void set_status_details(grpc_call *call, status_source source, grpc_mdstr *status) { if (call->status[source].details != NULL) { @@ -612,149 +482,39 @@ static void set_status_details(grpc_call *call, status_source source, call->status[source].details = status; } -static int is_op_live(grpc_call *call, grpc_ioreq_op op) { - gpr_uint8 set = call->request_set[op]; - reqinfo_master *master; - if (set >= GRPC_IOREQ_OP_COUNT) return 0; - master = &call->masters[set]; - return (master->complete_mask & (1u << op)) == 0; -} - -static void lock(grpc_call *call) { gpr_mu_lock(&call->mu); } - -static int need_more_data(grpc_call *call) { - if (call->read_state == READ_STATE_STREAM_CLOSED) return 0; - /* TODO(ctiller): this needs some serious cleanup */ - return is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA) || - (is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) && - grpc_bbq_empty(&call->incoming_queue)) || - is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA) || - is_op_live(call, GRPC_IOREQ_RECV_STATUS) || - is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS) || - (is_op_live(call, GRPC_IOREQ_RECV_CLOSE) && - grpc_bbq_empty(&call->incoming_queue)) || - (call->write_state == WRITE_STATE_INITIAL && !call->is_client) || - (call->cancel_with_status != GRPC_STATUS_OK) || call->destroy_called; -} - -static void unlock(grpc_exec_ctx *exec_ctx, grpc_call *call) { - grpc_transport_stream_op op; - completed_request completed_requests[GRPC_IOREQ_OP_COUNT]; - int completing_requests = 0; - int start_op = 0; - int i; - const size_t MAX_RECV_PEEK_AHEAD = 65536; - size_t buffered_bytes; - - GPR_TIMER_BEGIN("unlock", 0); - - memset(&op, 0, sizeof(op)); - - op.cancel_with_status = call->cancel_with_status; - start_op = op.cancel_with_status != GRPC_STATUS_OK; - call->cancel_with_status = GRPC_STATUS_OK; /* reset */ - - if (!call->receiving && need_more_data(call)) { - if (grpc_bbq_empty(&call->incoming_queue) && call->reading_message) { - op.max_recv_bytes = call->incoming_message_length - - call->incoming_message.length + MAX_RECV_PEEK_AHEAD; - } else { - buffered_bytes = grpc_bbq_bytes(&call->incoming_queue); - if (buffered_bytes > MAX_RECV_PEEK_AHEAD) { - op.max_recv_bytes = 0; - } else { - op.max_recv_bytes = MAX_RECV_PEEK_AHEAD - buffered_bytes; - } - } - /* TODO(ctiller): 1024 is basically to cover a bug - I don't understand yet */ - if (op.max_recv_bytes > 1024) { - op.recv_ops = &call->recv_ops; - op.recv_state = &call->recv_state; - op.on_done_recv = &call->on_done_recv; - call->receiving = 1; - GRPC_CALL_INTERNAL_REF(call, "receiving"); - start_op = 1; - } - } - - if (!call->sending) { - if (fill_send_ops(call, &op)) { - call->sending = 1; - GRPC_CALL_INTERNAL_REF(call, "sending"); - start_op = 1; - } - } - - if (!call->bound_pollset && call->cq && (!call->is_client || start_op)) { - call->bound_pollset = 1; - op.bind_pollset = grpc_cq_pollset(call->cq); - start_op = 1; - } - - if (!call->completing && call->num_completed_requests != 0) { - completing_requests = call->num_completed_requests; - memcpy(completed_requests, call->completed_requests, - sizeof(completed_requests)); - call->num_completed_requests = 0; - call->completing = 1; - GRPC_CALL_INTERNAL_REF(call, "completing"); - } - - gpr_mu_unlock(&call->mu); - - if (start_op) { - execute_op(exec_ctx, call, &op); - } - - if (completing_requests > 0) { - for (i = 0; i < completing_requests; i++) { - completed_requests[i].on_complete(exec_ctx, call, - completed_requests[i].success, - completed_requests[i].user_data); - } - lock(call); - call->completing = 0; - unlock(exec_ctx, call); - GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "completing"); - } - - GPR_TIMER_END("unlock", 0); -} - -static void get_final_status(grpc_call *call, grpc_ioreq_data out) { +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) { - out.recv_status.set_value(call->status[i].code, - out.recv_status.user_data); + set_value(call->status[i].code, set_value_user_data); return; } } if (call->is_client) { - out.recv_status.set_value(GRPC_STATUS_UNKNOWN, out.recv_status.user_data); + set_value(GRPC_STATUS_UNKNOWN, set_value_user_data); } else { - out.recv_status.set_value(GRPC_STATUS_OK, out.recv_status.user_data); + set_value(GRPC_STATUS_OK, set_value_user_data); } } -static void get_final_details(grpc_call *call, grpc_ioreq_data out) { +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.recv_status_details.details_capacity) { - *out.recv_status_details.details_capacity = GPR_MAX( - len + 1, *out.recv_status_details.details_capacity * 3 / 2); - *out.recv_status_details.details = - gpr_realloc(*out.recv_status_details.details, - *out.recv_status_details.details_capacity); + 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.recv_status_details.details, GPR_SLICE_START_PTR(details), - len); - (*out.recv_status_details.details)[len] = 0; + memcpy(*out_details, GPR_SLICE_START_PTR(details), len); + (*out_details)[len] = 0; } else { goto no_details; } @@ -763,333 +523,45 @@ static void get_final_details(grpc_call *call, grpc_ioreq_data out) { } no_details: - if (0 == *out.recv_status_details.details_capacity) { - *out.recv_status_details.details_capacity = 8; - *out.recv_status_details.details = - gpr_malloc(*out.recv_status_details.details_capacity); - } - **out.recv_status_details.details = 0; -} - -static void finish_live_ioreq_op(grpc_call *call, grpc_ioreq_op op, - int success) { - completed_request *cr; - gpr_uint8 master_set = call->request_set[op]; - reqinfo_master *master; - size_t i; - /* ioreq is live: we need to do something */ - master = &call->masters[master_set]; - /* NB: the following integer arithmetic operation needs to be in its - * expanded form due to the "integral promotion" performed (see section - * 3.2.1.1 of the C89 draft standard). A cast to the smaller container type - * is then required to avoid the compiler warning */ - master->complete_mask = (gpr_uint16)(master->complete_mask | (1u << op)); - if (!success) { - master->success = 0; - } - if (master->complete_mask == master->need_mask) { - for (i = 0; i < GRPC_IOREQ_OP_COUNT; i++) { - if (call->request_set[i] != master_set) { - continue; - } - call->request_set[i] = REQSET_DONE; - switch ((grpc_ioreq_op)i) { - case GRPC_IOREQ_RECV_MESSAGE: - case GRPC_IOREQ_SEND_MESSAGE: - call->request_set[i] = REQSET_EMPTY; - if (!master->success) { - call->write_state = WRITE_STATE_WRITE_CLOSED; - } - break; - case GRPC_IOREQ_SEND_STATUS: - if (call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details != - NULL) { - GRPC_MDSTR_UNREF( - call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details); - call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details = - NULL; - } - break; - case GRPC_IOREQ_RECV_CLOSE: - case GRPC_IOREQ_SEND_INITIAL_METADATA: - case GRPC_IOREQ_SEND_TRAILING_METADATA: - case GRPC_IOREQ_SEND_CLOSE: - break; - case GRPC_IOREQ_RECV_STATUS: - get_final_status(call, call->request_data[GRPC_IOREQ_RECV_STATUS]); - break; - case GRPC_IOREQ_RECV_STATUS_DETAILS: - get_final_details(call, - call->request_data[GRPC_IOREQ_RECV_STATUS_DETAILS]); - break; - case GRPC_IOREQ_RECV_INITIAL_METADATA: - GPR_SWAP(grpc_metadata_array, call->buffered_metadata[0], - *call->request_data[GRPC_IOREQ_RECV_INITIAL_METADATA] - .recv_metadata); - break; - case GRPC_IOREQ_RECV_TRAILING_METADATA: - GPR_SWAP(grpc_metadata_array, call->buffered_metadata[1], - *call->request_data[GRPC_IOREQ_RECV_TRAILING_METADATA] - .recv_metadata); - break; - case GRPC_IOREQ_OP_COUNT: - abort(); - break; - } - } - cr = &call->completed_requests[call->num_completed_requests++]; - cr->success = master->success; - cr->on_complete = master->on_complete; - cr->user_data = master->user_data; - } -} - -static void finish_ioreq_op(grpc_call *call, grpc_ioreq_op op, int success) { - if (is_op_live(call, op)) { - finish_live_ioreq_op(call, op, success); + if (0 == *out_details_capacity) { + *out_details_capacity = 8; + *out_details = gpr_malloc(*out_details_capacity); } + **out_details = 0; } -static void early_out_write_ops(grpc_call *call) { - switch (call->write_state) { - case WRITE_STATE_WRITE_CLOSED: - finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, 0); - finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, 0); - finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, 0); - finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, 1); - /* fallthrough */ - case WRITE_STATE_STARTED: - finish_ioreq_op(call, GRPC_IOREQ_SEND_INITIAL_METADATA, 0); - /* fallthrough */ - case WRITE_STATE_INITIAL: - /* do nothing */ - break; - } +static grpc_linked_mdelem *linked_from_md(grpc_metadata *md) { + return (grpc_linked_mdelem *)&md->internal_data; } -static void call_on_done_send(grpc_exec_ctx *exec_ctx, void *pc, int success) { - grpc_call *call = pc; - GPR_TIMER_BEGIN("call_on_done_send", 0); - lock(call); - if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_INITIAL_METADATA)) { - finish_ioreq_op(call, GRPC_IOREQ_SEND_INITIAL_METADATA, success); - call->write_state = WRITE_STATE_STARTED; - } - if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_MESSAGE)) { - finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, success); - } - if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_CLOSE)) { - finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, success); - finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, success); - finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, 1); - call->write_state = WRITE_STATE_WRITE_CLOSED; - } - if (!success) { - call->write_state = WRITE_STATE_WRITE_CLOSED; - early_out_write_ops(call); - } - call->send_ops.nops = 0; - call->last_send_contains = 0; - call->sending = 0; - unlock(exec_ctx, call); - GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "sending"); - GPR_TIMER_END("call_on_done_send", 0); -} - -static void finish_message(grpc_call *call) { - GPR_TIMER_BEGIN("finish_message", 0); - if (call->error_status_set == 0) { - /* TODO(ctiller): this could be a lot faster if coded directly */ - grpc_byte_buffer *byte_buffer; - /* some aliases for readability */ - gpr_slice *slices = call->incoming_message.slices; - const size_t nslices = call->incoming_message.count; - - if ((call->incoming_message_flags & GRPC_WRITE_INTERNAL_COMPRESS) && - (call->compression_algorithm > GRPC_COMPRESS_NONE)) { - byte_buffer = grpc_raw_compressed_byte_buffer_create( - slices, nslices, call->compression_algorithm); +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 { - byte_buffer = grpc_raw_byte_buffer_create(slices, nslices); - } - grpc_bbq_push(&call->incoming_queue, byte_buffer); - } - gpr_slice_buffer_reset_and_unref(&call->incoming_message); - GPR_ASSERT(call->incoming_message.count == 0); - call->reading_message = 0; - GPR_TIMER_END("finish_message", 0); -} - -static int begin_message(grpc_call *call, grpc_begin_message msg) { - /* can't begin a message when we're still reading a message */ - if (call->reading_message) { - char *message = NULL; - gpr_asprintf( - &message, "Message terminated early; read %d bytes, expected %d", - (int)call->incoming_message.length, (int)call->incoming_message_length); - cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message); - gpr_free(message); - return 0; - } - /* sanity check: if message flags indicate a compressed message, the - * compression level should already be present in the call, as parsed off its - * corresponding metadata. */ - if ((msg.flags & GRPC_WRITE_INTERNAL_COMPRESS) && - (call->compression_algorithm == GRPC_COMPRESS_NONE)) { - char *message = NULL; - char *alg_name; - if (!grpc_compression_algorithm_name(call->compression_algorithm, - &alg_name)) { - /* This shouldn't happen, other than due to data corruption */ - alg_name = "<unknown>"; - } - gpr_asprintf(&message, - "Invalid compression algorithm (%s) for compressed message.", - alg_name); - cancel_with_status(call, GRPC_STATUS_INTERNAL, message); - gpr_free(message); - return 0; - } - /* stash away parameters, and prepare for incoming slices */ - if (msg.length > grpc_channel_get_max_message_length(call->channel)) { - char *message = NULL; - gpr_asprintf( - &message, - "Maximum message length of %d exceeded by a message of length %d", - grpc_channel_get_max_message_length(call->channel), msg.length); - cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message); - gpr_free(message); - return 0; - } else if (msg.length > 0) { - call->reading_message = 1; - call->incoming_message_length = msg.length; - call->incoming_message_flags = msg.flags; - return 1; - } else { - finish_message(call); - return 1; - } -} - -static int add_slice_to_message(grpc_call *call, gpr_slice slice) { - if (GPR_SLICE_LENGTH(slice) == 0) { - gpr_slice_unref(slice); - return 1; - } - /* we have to be reading a message to know what to do here */ - if (!call->reading_message) { - gpr_slice_unref(slice); - cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, - "Received payload data while not reading a message"); - return 0; - } - /* append the slice to the incoming buffer */ - gpr_slice_buffer_add(&call->incoming_message, slice); - if (call->incoming_message.length > call->incoming_message_length) { - /* if we got too many bytes, complain */ - char *message = NULL; - gpr_asprintf( - &message, "Receiving message overflow; read %d bytes, expected %d", - (int)call->incoming_message.length, (int)call->incoming_message_length); - cancel_with_status(call, GRPC_STATUS_INVALID_ARGUMENT, message); - gpr_free(message); - return 0; - } else if (call->incoming_message.length == call->incoming_message_length) { - finish_message(call); - return 1; - } else { - return 1; - } -} - -static void call_on_done_recv(grpc_exec_ctx *exec_ctx, void *pc, int success) { - grpc_call *call = pc; - grpc_call *child_call; - grpc_call *next_child_call; - size_t i; - GPR_TIMER_BEGIN("call_on_done_recv", 0); - lock(call); - call->receiving = 0; - if (success) { - for (i = 0; success && i < call->recv_ops.nops; i++) { - grpc_stream_op *op = &call->recv_ops.ops[i]; - switch (op->type) { - case GRPC_NO_OP: - break; - case GRPC_OP_METADATA: - GPR_TIMER_BEGIN("recv_metadata", 0); - recv_metadata(exec_ctx, call, &op->data.metadata); - GPR_TIMER_END("recv_metadata", 0); - break; - case GRPC_OP_BEGIN_MESSAGE: - GPR_TIMER_BEGIN("begin_message", 0); - success = begin_message(call, op->data.begin_message); - GPR_TIMER_END("begin_message", 0); - break; - case GRPC_OP_SLICE: - GPR_TIMER_BEGIN("add_slice_to_message", 0); - success = add_slice_to_message(call, op->data.slice); - GPR_TIMER_END("add_slice_to_message", 0); - break; + for (i = 0; i < call->send_extra_metadata_count; i++) { + GRPC_MDELEM_REF(call->send_extra_metadata[i].md); } - } - if (!success) { - grpc_stream_ops_unref_owned_objects(&call->recv_ops.ops[i], - call->recv_ops.nops - i); - } - if (call->recv_state == GRPC_STREAM_RECV_CLOSED) { - GPR_ASSERT(call->read_state <= READ_STATE_READ_CLOSED); - call->read_state = READ_STATE_READ_CLOSED; - } - if (call->recv_state == GRPC_STREAM_CLOSED) { - GPR_ASSERT(call->read_state <= READ_STATE_STREAM_CLOSED); - call->read_state = READ_STATE_STREAM_CLOSED; - if (call->have_alarm) { - grpc_timer_cancel(exec_ctx, &call->alarm); + for (i = 1; i < call->send_extra_metadata_count; i++) { + call->send_extra_metadata[i].prev = &call->send_extra_metadata[i - 1]; } - /* 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); + for (i = 0; i < call->send_extra_metadata_count - 1; i++) { + call->send_extra_metadata[i].next = &call->send_extra_metadata[i + 1]; } - GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "closed"); } - finish_read_ops(call); - } else { - finish_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, 0); - finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS, 0); - finish_ioreq_op(call, GRPC_IOREQ_RECV_CLOSE, 0); - finish_ioreq_op(call, GRPC_IOREQ_RECV_TRAILING_METADATA, 0); - finish_ioreq_op(call, GRPC_IOREQ_RECV_INITIAL_METADATA, 0); - finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS_DETAILS, 0); } - call->recv_ops.nops = 0; - unlock(exec_ctx, call); - - GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "receiving"); - GPR_TIMER_END("call_on_done_recv", 0); -} - -static int prepare_application_metadata(grpc_call *call, size_t count, - grpc_metadata *metadata) { - size_t i; for (i = 0; i < count; i++) { grpc_metadata *md = &metadata[i]; - grpc_metadata *next_md = (i == count - 1) ? NULL : &metadata[i + 1]; - grpc_metadata *prev_md = (i == 0) ? NULL : &metadata[i - 1]; 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(call->metadata_context, md->key, - (const gpr_uint8 *)md->value, - md->value_length); + l->md = grpc_mdelem_from_string_and_buffer( + md->key, (const gpr_uint8 *)md->value, md->value_length); if (!grpc_mdstr_is_legal_header(l->md->key)) { gpr_log(GPR_ERROR, "attempt to send invalid metadata key: %s", grpc_mdstr_as_c_string(l->md->key)); @@ -1099,243 +571,49 @@ static int prepare_application_metadata(grpc_call *call, size_t count, gpr_log(GPR_ERROR, "attempt to send invalid metadata value"); return 0; } - l->next = next_md ? (grpc_linked_mdelem *)&next_md->internal_data : NULL; - l->prev = prev_md ? (grpc_linked_mdelem *)&prev_md->internal_data : NULL; } - return 1; -} - -static grpc_mdelem_list chain_metadata_from_app(grpc_call *call, size_t count, - grpc_metadata *metadata) { - grpc_mdelem_list out; - if (count == 0) { - out.head = out.tail = NULL; - return out; + for (i = 1; i < count; i++) { + linked_from_md(&metadata[i])->prev = linked_from_md(&metadata[i - 1]); } - out.head = (grpc_linked_mdelem *)&(metadata[0].internal_data); - out.tail = (grpc_linked_mdelem *)&(metadata[count - 1].internal_data); - return out; -} - -/* Copy the contents of a byte buffer into stream ops */ -static void copy_byte_buffer_to_stream_ops(grpc_byte_buffer *byte_buffer, - grpc_stream_op_buffer *sopb) { - size_t i; - - switch (byte_buffer->type) { - case GRPC_BB_RAW: - for (i = 0; i < byte_buffer->data.raw.slice_buffer.count; i++) { - gpr_slice slice = byte_buffer->data.raw.slice_buffer.slices[i]; - gpr_slice_ref(slice); - grpc_sopb_add_slice(sopb, slice); - } - break; + for (i = 0; i < count - 1; i++) { + linked_from_md(&metadata[i])->next = linked_from_md(&metadata[i + 1]); } -} - -static int fill_send_ops(grpc_call *call, grpc_transport_stream_op *op) { - grpc_ioreq_data data; - gpr_uint32 flags; - grpc_metadata_batch mdb; - size_t i; - GPR_ASSERT(op->send_ops == NULL); - - switch (call->write_state) { - case WRITE_STATE_INITIAL: - if (!is_op_live(call, GRPC_IOREQ_SEND_INITIAL_METADATA)) { - break; - } - data = call->request_data[GRPC_IOREQ_SEND_INITIAL_METADATA]; - mdb.list = chain_metadata_from_app(call, data.send_metadata.count, - data.send_metadata.metadata); - mdb.garbage.head = mdb.garbage.tail = NULL; - mdb.deadline = call->send_deadline; - for (i = 0; i < call->send_initial_metadata_count; i++) { - grpc_metadata_batch_link_head(&mdb, &call->send_initial_metadata[i]); - } - grpc_sopb_add_metadata(&call->send_ops, mdb); - op->send_ops = &call->send_ops; - call->last_send_contains |= 1 << GRPC_IOREQ_SEND_INITIAL_METADATA; - call->send_initial_metadata_count = 0; - /* fall through intended */ - case WRITE_STATE_STARTED: - if (is_op_live(call, GRPC_IOREQ_SEND_MESSAGE)) { - size_t length; - data = call->request_data[GRPC_IOREQ_SEND_MESSAGE]; - flags = call->request_flags[GRPC_IOREQ_SEND_MESSAGE]; - length = grpc_byte_buffer_length(data.send_message); - GPR_ASSERT(length <= GPR_UINT32_MAX); - grpc_sopb_add_begin_message(&call->send_ops, (gpr_uint32)length, flags); - copy_byte_buffer_to_stream_ops(data.send_message, &call->send_ops); - op->send_ops = &call->send_ops; - call->last_send_contains |= 1 << GRPC_IOREQ_SEND_MESSAGE; - } - if (is_op_live(call, GRPC_IOREQ_SEND_CLOSE)) { - op->is_last_send = 1; - op->send_ops = &call->send_ops; - call->last_send_contains |= 1 << GRPC_IOREQ_SEND_CLOSE; - if (!call->is_client) { - /* send trailing metadata */ - data = call->request_data[GRPC_IOREQ_SEND_TRAILING_METADATA]; - mdb.list = chain_metadata_from_app(call, data.send_metadata.count, - data.send_metadata.metadata); - mdb.garbage.head = mdb.garbage.tail = NULL; - mdb.deadline = gpr_inf_future(GPR_CLOCK_REALTIME); - /* send status */ - /* TODO(ctiller): cache common status values */ - data = call->request_data[GRPC_IOREQ_SEND_STATUS]; - grpc_metadata_batch_add_tail( - &mdb, &call->status_link, - grpc_channel_get_reffed_status_elem(call->channel, - data.send_status.code)); - if (data.send_status.details) { - grpc_metadata_batch_add_tail( - &mdb, &call->details_link, - grpc_mdelem_from_metadata_strings( - call->metadata_context, - GRPC_MDSTR_REF( - grpc_channel_get_message_string(call->channel)), - data.send_status.details)); - call->request_data[GRPC_IOREQ_SEND_STATUS].send_status.details = - NULL; - } - grpc_sopb_add_metadata(&call->send_ops, mdb); - } - } + 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 WRITE_STATE_WRITE_CLOSED: + 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; - } - if (op->send_ops) { - op->on_done_send = &call->on_done_send; - } - return op->send_ops != NULL; -} - -static grpc_call_error start_ioreq_error(grpc_call *call, - gpr_uint32 mutated_ops, - grpc_call_error ret) { - size_t i; - for (i = 0; i < GRPC_IOREQ_OP_COUNT; i++) { - if (mutated_ops & (1u << i)) { - call->request_set[i] = REQSET_EMPTY; - } - } - return ret; -} - -static void finish_read_ops(grpc_call *call) { - int empty; - - if (is_op_live(call, GRPC_IOREQ_RECV_MESSAGE)) { - empty = - (NULL == (*call->request_data[GRPC_IOREQ_RECV_MESSAGE].recv_message = - grpc_bbq_pop(&call->incoming_queue))); - if (!empty) { - finish_live_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, 1); - empty = grpc_bbq_empty(&call->incoming_queue); - } - } else { - empty = grpc_bbq_empty(&call->incoming_queue); - } - - switch (call->read_state) { - case READ_STATE_STREAM_CLOSED: - if (empty && !call->have_alarm) { - finish_ioreq_op(call, GRPC_IOREQ_RECV_CLOSE, 1); - } - /* fallthrough */ - case READ_STATE_READ_CLOSED: - if (empty) { - finish_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, 1); - } - finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS, 1); - finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS_DETAILS, 1); - finish_ioreq_op(call, GRPC_IOREQ_RECV_TRAILING_METADATA, 1); - /* fallthrough */ - case READ_STATE_GOT_INITIAL_METADATA: - finish_ioreq_op(call, GRPC_IOREQ_RECV_INITIAL_METADATA, 1); - /* fallthrough */ - case READ_STATE_INITIAL: - /* do nothing */ + 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); } -} - -static grpc_call_error start_ioreq(grpc_call *call, const grpc_ioreq *reqs, - size_t nreqs, - grpc_ioreq_completion_func completion, - void *user_data) { - size_t i; - gpr_uint16 have_ops = 0; - grpc_ioreq_op op; - reqinfo_master *master; - grpc_ioreq_data data; - gpr_uint8 set; - - if (nreqs == 0) { - return GRPC_CALL_OK; - } - - set = reqs[0].op; - - for (i = 0; i < nreqs; i++) { - op = reqs[i].op; - if (call->request_set[op] < GRPC_IOREQ_OP_COUNT) { - return start_ioreq_error(call, have_ops, - GRPC_CALL_ERROR_TOO_MANY_OPERATIONS); - } else if (call->request_set[op] == REQSET_DONE) { - return start_ioreq_error(call, have_ops, GRPC_CALL_ERROR_ALREADY_INVOKED); - } - data = reqs[i].data; - if (op == GRPC_IOREQ_SEND_INITIAL_METADATA || - op == GRPC_IOREQ_SEND_TRAILING_METADATA) { - if (!prepare_application_metadata(call, data.send_metadata.count, - data.send_metadata.metadata)) { - return start_ioreq_error(call, have_ops, - GRPC_CALL_ERROR_INVALID_METADATA); - } - } - if (op == GRPC_IOREQ_SEND_STATUS) { - set_status_code(call, STATUS_FROM_SERVER_STATUS, - (gpr_uint32)reqs[i].data.send_status.code); - if (reqs[i].data.send_status.details) { - set_status_details(call, STATUS_FROM_SERVER_STATUS, - GRPC_MDSTR_REF(reqs[i].data.send_status.details)); - } - } - /* NB: the following integer arithmetic operation needs to be in its - * expanded form due to the "integral promotion" performed (see section - * 3.2.1.1 of the C89 draft standard). A cast to the smaller container type - * is then required to avoid the compiler warning */ - have_ops = (gpr_uint16)(have_ops | (1u << op)); - - call->request_data[op] = data; - call->request_flags[op] = reqs[i].flags; - call->request_set[op] = set; - } - - master = &call->masters[set]; - master->success = 1; - master->need_mask = have_ops; - master->complete_mask = 0; - master->on_complete = completion; - master->user_data = user_data; - - finish_read_ops(call); - early_out_write_ops(call); - - return GRPC_CALL_OK; -} -grpc_call_error grpc_call_start_ioreq_and_call_back( - grpc_exec_ctx *exec_ctx, grpc_call *call, const grpc_ioreq *reqs, - size_t nreqs, grpc_ioreq_completion_func on_complete, void *user_data) { - grpc_call_error err; - lock(call); - err = start_ioreq(call, reqs, nreqs, on_complete, user_data); - unlock(exec_ctx, call); - return err; + return 1; } void grpc_call_destroy(grpc_call *c) { @@ -1343,6 +621,7 @@ void grpc_call_destroy(grpc_call *c) { 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) { @@ -1359,17 +638,18 @@ void grpc_call_destroy(grpc_call *c) { GRPC_CALL_INTERNAL_UNREF(&exec_ctx, parent, "child"); } - lock(c); + 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->read_state != READ_STATE_STREAM_CLOSED; - unlock(&exec_ctx, c); + 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) { @@ -1390,73 +670,80 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c, "c=%p, status=%d, description=%s, reserved=%p)", 4, (c, (int)status, description, reserved)); GPR_ASSERT(reserved == NULL); - lock(c); - r = cancel_with_status(c, status, description); - unlock(&exec_ctx, c); + 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; } -static grpc_call_error cancel_with_status(grpc_call *c, grpc_status_code status, +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, int 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, int 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(c->metadata_context, description) - : NULL; + 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, (gpr_uint32)status); set_status_details(c, STATUS_FROM_API_OVERRIDE, details); - c->cancel_with_status = status; + 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, 1); return GRPC_CALL_OK; } -static void finished_loose_op(grpc_exec_ctx *exec_ctx, void *call, - int success_ignored) { - GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "loose-op"); -} - -typedef struct { - grpc_call *call; - grpc_closure closure; -} finished_loose_op_allocated_args; - -static void finished_loose_op_allocated(grpc_exec_ctx *exec_ctx, void *alloc, - int success) { - finished_loose_op_allocated_args *args = alloc; - finished_loose_op(exec_ctx, args->call, success); - gpr_free(args); -} - static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call, grpc_transport_stream_op *op) { grpc_call_element *elem; - GPR_ASSERT(op->on_consumed == NULL); - if (op->cancel_with_status != GRPC_STATUS_OK || op->bind_pollset) { - GRPC_CALL_INTERNAL_REF(call, "loose-op"); - if (op->bind_pollset) { - op->on_consumed = &call->on_done_bind; - } else { - finished_loose_op_allocated_args *args = gpr_malloc(sizeof(*args)); - args->call = call; - grpc_closure_init(&args->closure, finished_loose_op_allocated, args); - op->on_consumed = &args->closure; - } - } - + 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 = elem->filter->get_peer(&exec_ctx, elem); + 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; } @@ -1467,14 +754,13 @@ grpc_call *grpc_call_from_top_element(grpc_call_element *elem) { static void call_alarm(grpc_exec_ctx *exec_ctx, void *arg, int success) { grpc_call *call = arg; - lock(call); + gpr_mu_lock(&call->mu); call->have_alarm = 0; if (success) { - cancel_with_status(call, GRPC_STATUS_DEADLINE_EXCEEDED, + cancel_with_status(exec_ctx, call, GRPC_STATUS_DEADLINE_EXCEEDED, "Deadline Exceeded"); } - finish_read_ops(call); - unlock(exec_ctx, call); + gpr_mu_unlock(&call->mu); GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "alarm"); } @@ -1500,8 +786,12 @@ static void destroy_status(void *ignored) {} static gpr_uint32 decode_status(grpc_mdelem *md) { gpr_uint32 status; - void *user_data = grpc_mdelem_get_user_data(md, destroy_status); - if (user_data) { + 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 = ((gpr_uint32)(gpr_intptr)user_data) - STATUS_OFFSET; } else { if (!gpr_parse_bytes_to_uint32(grpc_mdstr_as_c_string(md->value), @@ -1515,101 +805,77 @@ static gpr_uint32 decode_status(grpc_mdelem *md) { return status; } -/* just as for status above, we need to offset: metadata userdata can't hold a - * zero (null), which in this case is used to signal no compression */ -#define COMPRESS_OFFSET 1 -static void destroy_compression(void *ignored) {} - static gpr_uint32 decode_compression(grpc_mdelem *md) { - grpc_compression_algorithm algorithm; - void *user_data = grpc_mdelem_get_user_data(md, destroy_compression); - if (user_data) { - algorithm = - ((grpc_compression_algorithm)(gpr_intptr)user_data) - COMPRESS_OFFSET; - } else { + 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); - if (!grpc_compression_algorithm_parse(md_c_str, strlen(md_c_str), - &algorithm)) { - gpr_log(GPR_ERROR, "Invalid compression algorithm: '%s'", md_c_str); - assert(0); - } - grpc_mdelem_set_user_data( - md, destroy_compression, - (void *)(gpr_intptr)(algorithm + COMPRESS_OFFSET)); + gpr_log(GPR_ERROR, "Invalid compression algorithm: '%s'", md_c_str); } return algorithm; } -static void recv_metadata(grpc_exec_ctx *exec_ctx, grpc_call *call, - grpc_metadata_batch *md) { - grpc_linked_mdelem *l; +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; - int is_trailing; - - is_trailing = call->read_state >= READ_STATE_GOT_INITIAL_METADATA; - for (l = md->list.head; l != NULL; l = l->next) { - grpc_mdelem *mdel = l->md; - grpc_mdstr *key = mdel->key; - if (key == grpc_channel_get_status_string(call->channel)) { - GPR_TIMER_BEGIN("status", 0); - set_status_code(call, STATUS_FROM_WIRE, decode_status(mdel)); - GPR_TIMER_END("status", 0); - } else if (key == grpc_channel_get_message_string(call->channel)) { - GPR_TIMER_BEGIN("status-details", 0); - set_status_details(call, STATUS_FROM_WIRE, GRPC_MDSTR_REF(mdel->value)); - GPR_TIMER_END("status-details", 0); - } else if (key == - grpc_channel_get_compression_algorithm_string(call->channel)) { - GPR_TIMER_BEGIN("compression_algorithm", 0); - set_compression_algorithm(call, decode_compression(mdel)); - GPR_TIMER_END("compression_algorithm", 0); - } else if (key == grpc_channel_get_encodings_accepted_by_peer_string( - call->channel)) { - GPR_TIMER_BEGIN("encodings_accepted_by_peer", 0); - set_encodings_accepted_by_peer(call, mdel); - GPR_TIMER_END("encodings_accepted_by_peer", 0); - } else { - GPR_TIMER_BEGIN("report_up", 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(mdel->key); - mdusr->value = grpc_mdstr_as_c_string(mdel->value); - mdusr->value_length = GPR_SLICE_LENGTH(mdel->value->slice); - if (call->owned_metadata_count == call->owned_metadata_capacity) { - call->owned_metadata_capacity = - GPR_MAX(call->owned_metadata_capacity + 8, - call->owned_metadata_capacity * 2); - call->owned_metadata = - gpr_realloc(call->owned_metadata, - sizeof(grpc_mdelem *) * call->owned_metadata_capacity); - } - call->owned_metadata[call->owned_metadata_count++] = mdel; - l->md = NULL; - GPR_TIMER_END("report_up", 0); - } - } - 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 (!is_trailing) { - call->read_state = READ_STATE_GOT_INITIAL_METADATA; - } + 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; +} - for (l = md->list.head; l; l = l->next) { - if (l->md) GRPC_MDELEM_UNREF(l->md); +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); } - for (l = md->garbage.head; l; l = l->next) { - GRPC_MDELEM_UNREF(l->md); +} + +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); } } @@ -1626,19 +892,7 @@ static void set_status_value_directly(grpc_status_code status, void *dest) { } static void set_cancelled_value(grpc_status_code status, void *dest) { - *(grpc_status_code *)dest = (status != GRPC_STATUS_OK); -} - -static void finish_batch(grpc_exec_ctx *exec_ctx, grpc_call *call, int success, - void *tag) { - grpc_cq_end_op(exec_ctx, call->cq, tag, success, done_completion, call, - allocate_completion(call)); -} - -static void finish_batch_with_close(grpc_exec_ctx *exec_ctx, grpc_call *call, - int success, void *tag) { - grpc_cq_end_op(exec_ctx, call->cq, tag, 1, done_completion, call, - allocate_completion(call)); + *(int *)dest = (status != GRPC_STATUS_OK); } static int are_write_flags_valid(gpr_uint32 flags) { @@ -1649,258 +903,504 @@ static int are_write_flags_valid(gpr_uint32 flags) { return !(flags & invalid_positions); } -grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, - size_t nops, void *tag, void *reserved) { - grpc_ioreq reqs[GRPC_IOREQ_OP_COUNT]; - size_t in; - size_t out; +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 = + (gpr_uint8)(call->used_batches | (gpr_uint8)(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 = (gpr_uint8)( + call->used_batches & ~(gpr_uint8)(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); + gpr_mu_lock(&call->mu); + bctl->call->used_batches = + (gpr_uint8)(bctl->call->used_batches & + ~(gpr_uint8)(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(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, + int success) { + batch_control *bctl = bctlp; + grpc_call *call = bctl->call; + + GPR_ASSERT(success); + gpr_slice_buffer_add(&(*call->receiving_buffer)->data.raw.slice_buffer, + call->receiving_slice); + + continue_receiving_slices(exec_ctx, bctl); +} + +static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, int 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_initial_metadata) { + grpc_metadata_batch *md = + &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */]; + grpc_metadata_batch_filter(md, recv_initial_filter, call); + + 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 (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 void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp, + int success) { + batch_control *bctl = bctlp; + grpc_call *call = bctl->call; + + if (call->receiving_stream == NULL) { + *call->receiving_buffer = NULL; + 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(call->receiving_stream); + call->receiving_stream = NULL; + *call->receiving_buffer = NULL; + 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 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; - grpc_ioreq *req; - void (*finish_func)(grpc_exec_ctx *, grpc_call *, int, void *) = finish_batch; - grpc_call_error error; - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + 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_API_TRACE( - "grpc_call_start_batch(call=%p, ops=%p, nops=%lu, tag=%p, reserved=%p)", - 5, (call, ops, (unsigned long)nops, tag, reserved)); + GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, notify_tag); - if (reserved != NULL) { - error = GRPC_CALL_ERROR; - goto done; - } + memset(&stream_op, 0, sizeof(stream_op)); - GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, tag); + /* 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 = (gpr_uint8)(is_notify_tag_closure != 0); if (nops == 0) { - grpc_cq_begin_op(call->cq); GRPC_CALL_INTERNAL_REF(call, "completion"); - grpc_cq_end_op(&exec_ctx, call->cq, tag, 1, done_completion, call, - allocate_completion(call)); + 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 ioreq ops */ - for (in = 0, out = 0; in < nops; in++) { - op = &ops[in]; + /* 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; + 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; + goto done_with_error; + } + if (call->sent_initial_metadata) { + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + goto done_with_error; } - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (op->data.send_initial_metadata.count > INT_MAX) { + error = GRPC_CALL_ERROR_INVALID_METADATA; + goto done_with_error; } - req->op = GRPC_IOREQ_SEND_INITIAL_METADATA; - req->data.send_metadata.count = op->data.send_initial_metadata.count; - req->data.send_metadata.metadata = - op->data.send_initial_metadata.metadata; - req->flags = op->flags; + 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; + goto done_with_error; } if (op->data.send_message == NULL) { error = GRPC_CALL_ERROR_INVALID_MESSAGE; - goto done; + goto done_with_error; } - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (call->sending_message) { + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + goto done_with_error; } - req->op = GRPC_IOREQ_SEND_MESSAGE; - req->data.send_message = op->data.send_message; - req->flags = op->flags; + 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; + goto done_with_error; } if (!call->is_client) { error = GRPC_CALL_ERROR_NOT_ON_SERVER; - goto done; + goto done_with_error; } - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (call->sent_final_op) { + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + goto done_with_error; } - req->op = GRPC_IOREQ_SEND_CLOSE; - req->flags = op->flags; + 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; + goto done_with_error; } if (call->is_client) { error = GRPC_CALL_ERROR_NOT_ON_CLIENT; - goto done; + goto done_with_error; + } + if (call->sent_final_op) { + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + goto done_with_error; } - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (op->data.send_status_from_server.trailing_metadata_count > + INT_MAX) { + error = GRPC_CALL_ERROR_INVALID_METADATA; + goto done_with_error; } - req->op = GRPC_IOREQ_SEND_TRAILING_METADATA; - req->flags = op->flags; - req->data.send_metadata.count = - op->data.send_status_from_server.trailing_metadata_count; - req->data.send_metadata.metadata = - op->data.send_status_from_server.trailing_metadata; - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + 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)); } - req->op = GRPC_IOREQ_SEND_STATUS; - req->data.send_status.code = op->data.send_status_from_server.status; - req->data.send_status.details = - op->data.send_status_from_server.status_details != NULL - ? grpc_mdstr_from_string( - call->metadata_context, - op->data.send_status_from_server.status_details) - : NULL; - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + set_status_code(call, STATUS_FROM_API_OVERRIDE, + (gpr_uint32)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; } - req->op = GRPC_IOREQ_SEND_CLOSE; + 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; - } - if (!call->is_client) { - error = GRPC_CALL_ERROR_NOT_ON_SERVER; - goto done; + goto done_with_error; } - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (call->received_initial_metadata) { + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + goto done_with_error; } - req->op = GRPC_IOREQ_RECV_INITIAL_METADATA; - req->data.recv_metadata = op->data.recv_initial_metadata; - req->data.recv_metadata->count = 0; - req->flags = op->flags; + call->received_initial_metadata = 1; + call->buffered_metadata[0] = op->data.recv_initial_metadata; + bctl->recv_initial_metadata = 1; + stream_op.recv_initial_metadata = + &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */]; break; case GRPC_OP_RECV_MESSAGE: /* Flag validation: currently allow no flags */ if (op->flags != 0) { error = GRPC_CALL_ERROR_INVALID_FLAGS; - goto done; + goto done_with_error; } - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (call->receiving_message) { + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + goto done_with_error; } - req->op = GRPC_IOREQ_RECV_MESSAGE; - req->data.recv_message = op->data.recv_message; - req->flags = op->flags; + 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; + goto done_with_error; } if (!call->is_client) { error = GRPC_CALL_ERROR_NOT_ON_SERVER; - goto done; - } - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + goto done_with_error; } - req->op = GRPC_IOREQ_RECV_STATUS; - req->flags = op->flags; - req->data.recv_status.set_value = set_status_value_directly; - req->data.recv_status.user_data = op->data.recv_status_on_client.status; - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (call->received_final_op) { + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + goto done_with_error; } - req->op = GRPC_IOREQ_RECV_STATUS_DETAILS; - req->data.recv_status_details.details = + 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; - req->data.recv_status_details.details_capacity = + call->final_op.client.status_details_capacity = op->data.recv_status_on_client.status_details_capacity; - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; - } - req->op = GRPC_IOREQ_RECV_TRAILING_METADATA; - req->data.recv_metadata = - op->data.recv_status_on_client.trailing_metadata; - req->data.recv_metadata->count = 0; - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; - } - req->op = GRPC_IOREQ_RECV_CLOSE; - finish_func = finish_batch_with_close; + 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; + goto done_with_error; } - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (call->is_client) { + error = GRPC_CALL_ERROR_NOT_ON_CLIENT; + goto done_with_error; } - req->op = GRPC_IOREQ_RECV_STATUS; - req->flags = op->flags; - req->data.recv_status.set_value = set_cancelled_value; - req->data.recv_status.user_data = - op->data.recv_close_on_server.cancelled; - req = &reqs[out++]; - if (out > GRPC_IOREQ_OP_COUNT) { - error = GRPC_CALL_ERROR_BATCH_TOO_BIG; - goto done; + if (call->received_final_op) { + error = GRPC_CALL_ERROR_TOO_MANY_OPERATIONS; + goto done_with_error; } - req->op = GRPC_IOREQ_RECV_CLOSE; - finish_func = finish_batch_with_close; + 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"); - grpc_cq_begin_op(call->cq); + 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); - error = grpc_call_start_ioreq_and_call_back(&exec_ctx, call, reqs, out, - finish_func, tag); done: - grpc_exec_ctx_finish(&exec_ctx); 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(&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, diff --git a/src/core/surface/call.h b/src/core/surface/call.h index 9b7c6f9bfb..b53340df8e 100644 --- a/src/core/surface/call.h +++ b/src/core/surface/call.h @@ -44,51 +44,6 @@ extern "C" { #endif -/* Primitive operation types - grpc_op's get rewritten into these */ -typedef enum { - GRPC_IOREQ_RECV_INITIAL_METADATA, - GRPC_IOREQ_RECV_MESSAGE, - GRPC_IOREQ_RECV_TRAILING_METADATA, - GRPC_IOREQ_RECV_STATUS, - GRPC_IOREQ_RECV_STATUS_DETAILS, - GRPC_IOREQ_RECV_CLOSE, - GRPC_IOREQ_SEND_INITIAL_METADATA, - GRPC_IOREQ_SEND_MESSAGE, - GRPC_IOREQ_SEND_TRAILING_METADATA, - GRPC_IOREQ_SEND_STATUS, - GRPC_IOREQ_SEND_CLOSE, - GRPC_IOREQ_OP_COUNT -} grpc_ioreq_op; - -typedef union { - grpc_metadata_array *recv_metadata; - grpc_byte_buffer **recv_message; - struct { - void (*set_value)(grpc_status_code status, void *user_data); - void *user_data; - } recv_status; - struct { - char **details; - size_t *details_capacity; - } recv_status_details; - struct { - size_t count; - grpc_metadata *metadata; - } send_metadata; - grpc_byte_buffer *send_message; - struct { - grpc_status_code code; - grpc_mdstr *details; - } send_status; -} grpc_ioreq_data; - -typedef struct { - grpc_ioreq_op op; - gpr_uint32 flags; - /**< A copy of the write flags from grpc_op */ - grpc_ioreq_data data; -} grpc_ioreq; - typedef void (*grpc_ioreq_completion_func)(grpc_exec_ctx *exec_ctx, grpc_call *call, int success, void *user_data); @@ -103,9 +58,8 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call, void grpc_call_set_completion_queue(grpc_exec_ctx *exec_ctx, grpc_call *call, grpc_completion_queue *cq); -grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call); -#ifdef GRPC_CALL_REF_COUNT_DEBUG +#ifdef GRPC_STREAM_REFCOUNT_DEBUG void grpc_call_internal_ref(grpc_call *call, const char *reason); void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *call, const char *reason); @@ -121,12 +75,14 @@ void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *call); grpc_call_internal_unref(exec_ctx, call) #endif -grpc_call_error grpc_call_start_ioreq_and_call_back( - grpc_exec_ctx *exec_ctx, grpc_call *call, const grpc_ioreq *reqs, - size_t nreqs, grpc_ioreq_completion_func on_complete, void *user_data); - grpc_call_stack *grpc_call_get_call_stack(grpc_call *call); +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); + /* Given the top call_element, get the call object. */ grpc_call *grpc_call_from_top_element(grpc_call_element *surface_element); @@ -134,19 +90,6 @@ void grpc_call_log_batch(char *file, int line, gpr_log_severity severity, grpc_call *call, const grpc_op *ops, size_t nops, void *tag); -void grpc_server_log_request_call(char *file, int line, - gpr_log_severity severity, - grpc_server *server, grpc_call **call, - grpc_call_details *details, - grpc_metadata_array *initial_metadata, - grpc_completion_queue *cq_bound_to_call, - grpc_completion_queue *cq_for_notification, - void *tag); - -void grpc_server_log_shutdown(char *file, int line, gpr_log_severity severity, - grpc_server *server, grpc_completion_queue *cq, - void *tag); - /* Set a context pointer. No thread safety guarantees are made wrt this value. */ void grpc_call_context_set(grpc_call *call, grpc_context_index elem, @@ -157,16 +100,6 @@ void *grpc_call_context_get(grpc_call *call, grpc_context_index elem); #define GRPC_CALL_LOG_BATCH(sev, call, ops, nops, tag) \ if (grpc_api_trace) grpc_call_log_batch(sev, call, ops, nops, tag) -#define GRPC_SERVER_LOG_REQUEST_CALL(sev, server, call, details, \ - initial_metadata, cq_bound_to_call, \ - cq_for_notifications, tag) \ - if (grpc_api_trace) \ - grpc_server_log_request_call(sev, server, call, details, initial_metadata, \ - cq_bound_to_call, cq_for_notifications, tag) - -#define GRPC_SERVER_LOG_SHUTDOWN(sev, server, cq, tag) \ - if (grpc_api_trace) grpc_server_log_shutdown(sev, server, cq, tag) - gpr_uint8 grpc_call_is_client(grpc_call *call); #ifdef __cplusplus diff --git a/src/core/surface/call_log_batch.c b/src/core/surface/call_log_batch.c index 2dd9737cf8..46756f418b 100644 --- a/src/core/surface/call_log_batch.c +++ b/src/core/surface/call_log_batch.c @@ -110,36 +110,9 @@ void grpc_call_log_batch(char *file, int line, gpr_log_severity severity, void *tag) { char *tmp; size_t i; - gpr_log(file, line, severity, - "grpc_call_start_batch(call=%p, ops=%p, nops=%d, tag=%p)", call, ops, - nops, tag); for (i = 0; i < nops; i++) { tmp = grpc_op_string(&ops[i]); gpr_log(file, line, severity, "ops[%d]: %s", i, tmp); gpr_free(tmp); } } - -void grpc_server_log_request_call(char *file, int line, - gpr_log_severity severity, - grpc_server *server, grpc_call **call, - grpc_call_details *details, - grpc_metadata_array *initial_metadata, - grpc_completion_queue *cq_bound_to_call, - grpc_completion_queue *cq_for_notification, - void *tag) { - gpr_log(file, line, severity, - "grpc_server_request_call(server=%p, call=%p, details=%p, " - "initial_metadata=%p, cq_bound_to_call=%p, cq_for_notification=%p, " - "tag=%p)", - server, call, details, initial_metadata, cq_bound_to_call, - cq_for_notification, tag); -} - -void grpc_server_log_shutdown(char *file, int line, gpr_log_severity severity, - grpc_server *server, grpc_completion_queue *cq, - void *tag) { - gpr_log(file, line, severity, - "grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", server, - cq, tag); -} diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c index a9a5f828f2..d0a8b0be09 100644 --- a/src/core/surface/channel.c +++ b/src/core/surface/channel.c @@ -46,6 +46,7 @@ #include "src/core/surface/api_trace.h" #include "src/core/surface/call.h" #include "src/core/surface/init.h" +#include "src/core/transport/static_metadata.h" /** Cache grpc-status: X mdelems for X = 0..NUM_CACHED_STATUS_ELEMS. * Avoids needing to take a metadata context lock for sending status @@ -62,19 +63,8 @@ typedef struct registered_call { struct grpc_channel { int is_client; - gpr_refcount refs; gpr_uint32 max_message_length; - grpc_mdctx *metadata_context; - /** mdstr for the grpc-status key */ - grpc_mdstr *grpc_status_string; - grpc_mdstr *grpc_compression_algorithm_string; - grpc_mdstr *grpc_encodings_accepted_by_peer_string; - grpc_mdstr *grpc_message_string; - grpc_mdstr *path_string; - grpc_mdstr *authority_string; grpc_mdelem *default_authority; - /** mdelem for grpc-status: 0 thru grpc-status: 2 */ - grpc_mdelem *grpc_status_elem[NUM_CACHED_STATUS_ELEMS]; gpr_mu registered_call_mu; registered_call *registered_calls; @@ -90,10 +80,12 @@ struct grpc_channel { /* the protobuf library will (by default) start warning at 100megs */ #define DEFAULT_MAX_MESSAGE_LENGTH (100 * 1024 * 1024) +static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, int success); + grpc_channel *grpc_channel_create_from_filters( grpc_exec_ctx *exec_ctx, const char *target, const grpc_channel_filter **filters, size_t num_filters, - const grpc_channel_args *args, grpc_mdctx *mdctx, int is_client) { + const grpc_channel_args *args, int is_client) { size_t i; size_t size = sizeof(grpc_channel) + grpc_channel_stack_size(filters, num_filters); @@ -102,24 +94,6 @@ grpc_channel *grpc_channel_create_from_filters( channel->target = gpr_strdup(target); GPR_ASSERT(grpc_is_initialized() && "call grpc_init()"); channel->is_client = is_client; - /* decremented by grpc_channel_destroy */ - gpr_ref_init(&channel->refs, 1); - channel->metadata_context = mdctx; - channel->grpc_status_string = grpc_mdstr_from_string(mdctx, "grpc-status"); - channel->grpc_compression_algorithm_string = - grpc_mdstr_from_string(mdctx, "grpc-encoding"); - channel->grpc_encodings_accepted_by_peer_string = - grpc_mdstr_from_string(mdctx, "grpc-accept-encoding"); - channel->grpc_message_string = grpc_mdstr_from_string(mdctx, "grpc-message"); - for (i = 0; i < NUM_CACHED_STATUS_ELEMS; i++) { - char buf[GPR_LTOA_MIN_BUFSIZE]; - gpr_ltoa((long)i, buf); - channel->grpc_status_elem[i] = grpc_mdelem_from_metadata_strings( - mdctx, GRPC_MDSTR_REF(channel->grpc_status_string), - grpc_mdstr_from_string(mdctx, buf)); - } - channel->path_string = grpc_mdstr_from_string(mdctx, ":path"); - channel->authority_string = grpc_mdstr_from_string(mdctx, ":authority"); gpr_mu_init(&channel->registered_call_mu); channel->registered_calls = NULL; @@ -138,7 +112,7 @@ grpc_channel *grpc_channel_create_from_filters( } } else if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) { if (args->args[i].type != GRPC_ARG_STRING) { - gpr_log(GPR_ERROR, "%s: must be an string", + gpr_log(GPR_ERROR, "%s ignored: it must be a string", GRPC_ARG_DEFAULT_AUTHORITY); } else { if (channel->default_authority) { @@ -146,21 +120,22 @@ grpc_channel *grpc_channel_create_from_filters( GRPC_MDELEM_UNREF(channel->default_authority); } channel->default_authority = grpc_mdelem_from_strings( - mdctx, ":authority", args->args[i].value.string); + ":authority", args->args[i].value.string); } } else if (0 == strcmp(args->args[i].key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)) { if (args->args[i].type != GRPC_ARG_STRING) { - gpr_log(GPR_ERROR, "%s: must be an string", + gpr_log(GPR_ERROR, "%s ignored: it must be a string", GRPC_SSL_TARGET_NAME_OVERRIDE_ARG); } else { if (channel->default_authority) { /* other ways of setting this (notably ssl) take precedence */ - gpr_log(GPR_ERROR, "%s: default host already set some other way", - GRPC_ARG_DEFAULT_AUTHORITY); + gpr_log(GPR_ERROR, + "%s ignored: default host already set some other way", + GRPC_SSL_TARGET_NAME_OVERRIDE_ARG); } else { channel->default_authority = grpc_mdelem_from_strings( - mdctx, ":authority", args->args[i].value.string); + ":authority", args->args[i].value.string); } } } @@ -171,14 +146,15 @@ grpc_channel *grpc_channel_create_from_filters( target != NULL) { char *default_authority = grpc_get_default_authority(target); if (default_authority) { - channel->default_authority = grpc_mdelem_from_strings( - channel->metadata_context, ":authority", default_authority); + channel->default_authority = + grpc_mdelem_from_strings(":authority", default_authority); } gpr_free(default_authority); } - grpc_channel_stack_init(exec_ctx, filters, num_filters, channel, args, - channel->metadata_context, + grpc_channel_stack_init(exec_ctx, 1, destroy_channel, channel, filters, + num_filters, args, + is_client ? "CLIENT_CHANNEL" : "SERVER_CHANNEL", CHANNEL_STACK_FROM_CHANNEL(channel)); return channel; @@ -219,21 +195,18 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel, "grpc_channel_create_call(" "channel=%p, parent_call=%p, propagation_mask=%x, cq=%p, method=%s, " "host=%s, " - "deadline=gpr_timespec { tv_sec: %ld, tv_nsec: %d, clock_type: %d }, " + "deadline=gpr_timespec { tv_sec: %lld, tv_nsec: %d, clock_type: %d }, " "reserved=%p)", 10, (channel, parent_call, (unsigned)propagation_mask, cq, method, host, - (long)deadline.tv_sec, deadline.tv_nsec, (int)deadline.clock_type, - reserved)); + (long long)deadline.tv_sec, (int)deadline.tv_nsec, + (int)deadline.clock_type, reserved)); GPR_ASSERT(!reserved); return grpc_channel_create_call_internal( channel, parent_call, propagation_mask, cq, - grpc_mdelem_from_metadata_strings( - channel->metadata_context, GRPC_MDSTR_REF(channel->path_string), - grpc_mdstr_from_string(channel->metadata_context, method)), - host ? grpc_mdelem_from_metadata_strings( - channel->metadata_context, - GRPC_MDSTR_REF(channel->authority_string), - grpc_mdstr_from_string(channel->metadata_context, host)) + grpc_mdelem_from_metadata_strings(GRPC_MDSTR_PATH, + grpc_mdstr_from_string(method)), + host ? grpc_mdelem_from_metadata_strings(GRPC_MDSTR_AUTHORITY, + grpc_mdstr_from_string(host)) : NULL, deadline); } @@ -245,15 +218,11 @@ void *grpc_channel_register_call(grpc_channel *channel, const char *method, "grpc_channel_register_call(channel=%p, method=%s, host=%s, reserved=%p)", 4, (channel, method, host, reserved)); GPR_ASSERT(!reserved); - rc->path = grpc_mdelem_from_metadata_strings( - channel->metadata_context, GRPC_MDSTR_REF(channel->path_string), - grpc_mdstr_from_string(channel->metadata_context, method)); - rc->authority = - host ? grpc_mdelem_from_metadata_strings( - channel->metadata_context, - GRPC_MDSTR_REF(channel->authority_string), - grpc_mdstr_from_string(channel->metadata_context, host)) - : NULL; + rc->path = grpc_mdelem_from_metadata_strings(GRPC_MDSTR_PATH, + grpc_mdstr_from_string(method)); + rc->authority = host ? grpc_mdelem_from_metadata_strings( + GRPC_MDSTR_AUTHORITY, grpc_mdstr_from_string(host)) + : NULL; gpr_mu_lock(&channel->registered_call_mu); rc->next = channel->registered_calls; channel->registered_calls = rc; @@ -270,11 +239,11 @@ grpc_call *grpc_channel_create_registered_call( "grpc_channel_create_registered_call(" "channel=%p, parent_call=%p, propagation_mask=%x, completion_queue=%p, " "registered_call_handle=%p, " - "deadline=gpr_timespec { tv_sec: %ld, tv_nsec: %d, clock_type: %d }, " + "deadline=gpr_timespec { tv_sec: %lld, tv_nsec: %d, clock_type: %d }, " "reserved=%p)", 9, (channel, parent_call, (unsigned)propagation_mask, completion_queue, - registered_call_handle, (long)deadline.tv_sec, deadline.tv_nsec, - (int)deadline.clock_type, reserved)); + registered_call_handle, (long long)deadline.tv_sec, + (int)deadline.tv_nsec, (int)deadline.clock_type, reserved)); GPR_ASSERT(!reserved); return grpc_channel_create_call_internal( channel, parent_call, propagation_mask, completion_queue, @@ -282,28 +251,26 @@ grpc_call *grpc_channel_create_registered_call( rc->authority ? GRPC_MDELEM_REF(rc->authority) : NULL, deadline); } -#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG -void grpc_channel_internal_ref(grpc_channel *c, const char *reason) { - gpr_log(GPR_DEBUG, "CHANNEL: ref %p %d -> %d [%s]", c, c->refs.count, - c->refs.count + 1, reason); +#ifdef GRPC_STREAM_REFCOUNT_DEBUG +#define REF_REASON reason +#define REF_ARG , const char *reason #else -void grpc_channel_internal_ref(grpc_channel *c) { +#define REF_REASON "" +#define REF_ARG #endif - gpr_ref(&c->refs); +void grpc_channel_internal_ref(grpc_channel *c REF_ARG) { + GRPC_CHANNEL_STACK_REF(CHANNEL_STACK_FROM_CHANNEL(c), REF_REASON); } -static void destroy_channel(grpc_exec_ctx *exec_ctx, grpc_channel *channel) { - size_t i; +void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx, + grpc_channel *c REF_ARG) { + GRPC_CHANNEL_STACK_UNREF(exec_ctx, CHANNEL_STACK_FROM_CHANNEL(c), REF_REASON); +} + +static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, + int iomgr_success) { + grpc_channel *channel = arg; grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CHANNEL(channel)); - for (i = 0; i < NUM_CACHED_STATUS_ELEMS; i++) { - GRPC_MDELEM_UNREF(channel->grpc_status_elem[i]); - } - GRPC_MDSTR_UNREF(channel->grpc_status_string); - GRPC_MDSTR_UNREF(channel->grpc_compression_algorithm_string); - GRPC_MDSTR_UNREF(channel->grpc_encodings_accepted_by_peer_string); - GRPC_MDSTR_UNREF(channel->grpc_message_string); - GRPC_MDSTR_UNREF(channel->path_string); - GRPC_MDSTR_UNREF(channel->authority_string); while (channel->registered_calls) { registered_call *rc = channel->registered_calls; channel->registered_calls = rc->next; @@ -316,26 +283,11 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, grpc_channel *channel) { if (channel->default_authority != NULL) { GRPC_MDELEM_UNREF(channel->default_authority); } - grpc_mdctx_unref(channel->metadata_context); gpr_mu_destroy(&channel->registered_call_mu); gpr_free(channel->target); gpr_free(channel); } -#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG -void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx, grpc_channel *channel, - const char *reason) { - gpr_log(GPR_DEBUG, "CHANNEL: unref %p %d -> %d [%s]", channel, - channel->refs.count, channel->refs.count - 1, reason); -#else -void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx, - grpc_channel *channel) { -#endif - if (gpr_unref(&channel->refs)) { - destroy_channel(exec_ctx, channel); - } -} - void grpc_channel_destroy(grpc_channel *channel) { grpc_transport_op op; grpc_channel_element *elem; @@ -355,38 +307,19 @@ grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel) { return CHANNEL_STACK_FROM_CHANNEL(channel); } -grpc_mdctx *grpc_channel_get_metadata_context(grpc_channel *channel) { - return channel->metadata_context; -} - -grpc_mdstr *grpc_channel_get_status_string(grpc_channel *channel) { - return channel->grpc_status_string; -} - -grpc_mdstr *grpc_channel_get_compression_algorithm_string( - grpc_channel *channel) { - return channel->grpc_compression_algorithm_string; -} - -grpc_mdstr *grpc_channel_get_encodings_accepted_by_peer_string( - grpc_channel *channel) { - return channel->grpc_encodings_accepted_by_peer_string; -} - grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) { - if (i >= 0 && i < NUM_CACHED_STATUS_ELEMS) { - return GRPC_MDELEM_REF(channel->grpc_status_elem[i]); - } else { - char tmp[GPR_LTOA_MIN_BUFSIZE]; - gpr_ltoa(i, tmp); - return grpc_mdelem_from_metadata_strings( - channel->metadata_context, GRPC_MDSTR_REF(channel->grpc_status_string), - grpc_mdstr_from_string(channel->metadata_context, tmp)); + char tmp[GPR_LTOA_MIN_BUFSIZE]; + switch (i) { + case 0: + return GRPC_MDELEM_GRPC_STATUS_0; + case 1: + return GRPC_MDELEM_GRPC_STATUS_1; + case 2: + return GRPC_MDELEM_GRPC_STATUS_2; } -} - -grpc_mdstr *grpc_channel_get_message_string(grpc_channel *channel) { - return channel->grpc_message_string; + gpr_ltoa(i, tmp); + return grpc_mdelem_from_metadata_strings(GRPC_MDSTR_GRPC_STATUS, + grpc_mdstr_from_string(tmp)); } gpr_uint32 grpc_channel_get_max_message_length(grpc_channel *channel) { diff --git a/src/core/surface/channel.h b/src/core/surface/channel.h index e5030d52d2..3d2ff23542 100644 --- a/src/core/surface/channel.h +++ b/src/core/surface/channel.h @@ -40,29 +40,20 @@ grpc_channel *grpc_channel_create_from_filters( grpc_exec_ctx *exec_ctx, const char *target, const grpc_channel_filter **filters, size_t count, - const grpc_channel_args *args, grpc_mdctx *mdctx, int is_client); + const grpc_channel_args *args, int is_client); /** Get a (borrowed) pointer to this channels underlying channel stack */ grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel); -/** Get a (borrowed) pointer to the channel wide metadata context */ -grpc_mdctx *grpc_channel_get_metadata_context(grpc_channel *channel); - /** Get a grpc_mdelem of grpc-status: X where X is the numeric value of status_code. The returned elem is owned by the caller. */ grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int status_code); -grpc_mdstr *grpc_channel_get_status_string(grpc_channel *channel); -grpc_mdstr *grpc_channel_get_compression_algorithm_string( - grpc_channel *channel); -grpc_mdstr *grpc_channel_get_encodings_accepted_by_peer_string( - grpc_channel *channel); -grpc_mdstr *grpc_channel_get_message_string(grpc_channel *channel); gpr_uint32 grpc_channel_get_max_message_length(grpc_channel *channel); -#ifdef GRPC_CHANNEL_REF_COUNT_DEBUG +#ifdef GRPC_STREAM_REFCOUNT_DEBUG void grpc_channel_internal_ref(grpc_channel *channel, const char *reason); void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx, grpc_channel *channel, const char *reason); diff --git a/src/core/surface/channel_connectivity.c b/src/core/surface/channel_connectivity.c index df2774b527..10f5c4da4d 100644 --- a/src/core/surface/channel_connectivity.c +++ b/src/core/surface/channel_connectivity.c @@ -83,7 +83,6 @@ typedef struct { gpr_mu mu; callback_phase phase; int success; - int removed; grpc_closure on_complete; grpc_timer alarm; grpc_connectivity_state state; @@ -135,30 +134,15 @@ static void finished_completion(grpc_exec_ctx *exec_ctx, void *pw, static void partly_done(grpc_exec_ctx *exec_ctx, state_watcher *w, int due_to_completion) { int delete = 0; - grpc_channel_element *client_channel_elem = NULL; - gpr_mu_lock(&w->mu); - if (w->removed == 0) { - w->removed = 1; - client_channel_elem = grpc_channel_stack_last_element( - grpc_channel_get_channel_stack(w->channel)); - if (client_channel_elem->filter == &grpc_client_channel_filter) { - grpc_client_channel_del_interested_party(exec_ctx, client_channel_elem, - grpc_cq_pollset(w->cq)); - } else { - grpc_client_uchannel_del_interested_party(exec_ctx, client_channel_elem, - grpc_cq_pollset(w->cq)); - } - } - gpr_mu_unlock(&w->mu); if (due_to_completion) { - gpr_mu_lock(&w->mu); - w->success = 1; - gpr_mu_unlock(&w->mu); grpc_timer_cancel(exec_ctx, &w->alarm); } gpr_mu_lock(&w->mu); + if (due_to_completion) { + w->success = 1; + } switch (w->phase) { case WAITING: w->phase = CALLING_BACK; @@ -200,19 +184,18 @@ void grpc_channel_watch_connectivity_state( GRPC_API_TRACE( "grpc_channel_watch_connectivity_state(" "channel=%p, last_observed_state=%d, " - "deadline=gpr_timespec { tv_sec: %ld, tv_nsec: %d, clock_type: %d }, " + "deadline=gpr_timespec { tv_sec: %lld, tv_nsec: %d, clock_type: %d }, " "cq=%p, tag=%p)", - 7, (channel, (int)last_observed_state, (long)deadline.tv_sec, - deadline.tv_nsec, (int)deadline.clock_type, cq, tag)); + 7, (channel, (int)last_observed_state, (long long)deadline.tv_sec, + (int)deadline.tv_nsec, (int)deadline.clock_type, cq, tag)); - grpc_cq_begin_op(cq); + grpc_cq_begin_op(cq, tag); gpr_mu_init(&w->mu); grpc_closure_init(&w->on_complete, watch_complete, w); w->phase = WAITING; w->state = last_observed_state; w->success = 0; - w->removed = 0; w->cq = cq; w->tag = tag; w->channel = channel; @@ -223,16 +206,14 @@ void grpc_channel_watch_connectivity_state( if (client_channel_elem->filter == &grpc_client_channel_filter) { GRPC_CHANNEL_INTERNAL_REF(channel, "watch_channel_connectivity"); - grpc_client_channel_add_interested_party(&exec_ctx, client_channel_elem, - grpc_cq_pollset(cq)); grpc_client_channel_watch_connectivity_state(&exec_ctx, client_channel_elem, - &w->state, &w->on_complete); + grpc_cq_pollset(cq), &w->state, + &w->on_complete); } else if (client_channel_elem->filter == &grpc_client_uchannel_filter) { GRPC_CHANNEL_INTERNAL_REF(channel, "watch_uchannel_connectivity"); - grpc_client_uchannel_add_interested_party(&exec_ctx, client_channel_elem, - grpc_cq_pollset(cq)); grpc_client_uchannel_watch_connectivity_state( - &exec_ctx, client_channel_elem, &w->state, &w->on_complete); + &exec_ctx, client_channel_elem, grpc_cq_pollset(cq), &w->state, + &w->on_complete); } grpc_exec_ctx_finish(&exec_ctx); diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c index 51d9130b63..97ec23408f 100644 --- a/src/core/surface/channel_create.c +++ b/src/core/surface/channel_create.c @@ -37,6 +37,8 @@ #include <string.h> #include <grpc/support/alloc.h> +#include <grpc/support/slice.h> +#include <grpc/support/slice_buffer.h> #include "src/core/census/grpc_filter.h" #include "src/core/channel/channel_args.h" @@ -56,11 +58,11 @@ typedef struct { grpc_closure *notify; grpc_connect_in_args args; grpc_connect_out_args *result; + grpc_closure initial_string_sent; + gpr_slice_buffer initial_string_buffer; grpc_endpoint *tcp; - grpc_mdctx *mdctx; - grpc_closure connected; } connector; @@ -72,18 +74,33 @@ static void connector_ref(grpc_connector *con) { static void connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) { connector *c = (connector *)con; if (gpr_unref(&c->refs)) { - grpc_mdctx_unref(c->mdctx); + /* c->initial_string_buffer does not need to be destroyed */ gpr_free(c); } } +static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg, + int success) { + connector_unref(exec_ctx, arg); +} + static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) { connector *c = arg; grpc_closure *notify; grpc_endpoint *tcp = c->tcp; if (tcp != NULL) { - c->result->transport = grpc_create_chttp2_transport( - exec_ctx, c->args.channel_args, tcp, c->mdctx, 1); + if (!GPR_SLICE_IS_EMPTY(c->args.initial_connect_string)) { + grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent, + c); + gpr_slice_buffer_init(&c->initial_string_buffer); + gpr_slice_buffer_add(&c->initial_string_buffer, + c->args.initial_connect_string); + connector_ref(arg); + grpc_endpoint_write(exec_ctx, tcp, &c->initial_string_buffer, + &c->initial_string_sent); + } + c->result->transport = + grpc_create_chttp2_transport(exec_ctx, c->args.channel_args, tcp, 1); grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport, NULL, 0); GPR_ASSERT(c->result->transport); @@ -123,7 +140,6 @@ static const grpc_connector_vtable connector_vtable = { typedef struct { grpc_subchannel_factory base; gpr_refcount refs; - grpc_mdctx *mdctx; grpc_channel_args *merge_args; grpc_channel *master; } subchannel_factory; @@ -139,7 +155,6 @@ static void subchannel_factory_unref(grpc_exec_ctx *exec_ctx, if (gpr_unref(&f->refs)) { GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, f->master, "subchannel_factory"); grpc_channel_args_destroy(f->merge_args); - grpc_mdctx_unref(f->mdctx); gpr_free(f); } } @@ -154,12 +169,8 @@ static grpc_subchannel *subchannel_factory_create_subchannel( grpc_subchannel *s; memset(c, 0, sizeof(*c)); c->base.vtable = &connector_vtable; - c->mdctx = f->mdctx; - grpc_mdctx_ref(c->mdctx); gpr_ref_init(&c->refs, 1); - args->mdctx = f->mdctx; args->args = final_args; - args->master = f->master; s = grpc_subchannel_create(&c->base, args); grpc_connector_unref(exec_ctx, &c->base); grpc_channel_args_destroy(final_args); @@ -182,7 +193,6 @@ grpc_channel *grpc_insecure_channel_create(const char *target, const grpc_channel_filter *filters[MAX_FILTERS]; grpc_resolver *resolver; subchannel_factory *f; - grpc_mdctx *mdctx = grpc_mdctx_create(); grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; size_t n = 0; GRPC_API_TRACE( @@ -196,19 +206,20 @@ grpc_channel *grpc_insecure_channel_create(const char *target, filters[n++] = &grpc_client_channel_filter; GPR_ASSERT(n <= MAX_FILTERS); - channel = grpc_channel_create_from_filters(&exec_ctx, target, filters, n, - args, mdctx, 1); + channel = + grpc_channel_create_from_filters(&exec_ctx, target, filters, n, args, 1); f = gpr_malloc(sizeof(*f)); f->base.vtable = &subchannel_factory_vtable; gpr_ref_init(&f->refs, 1); - grpc_mdctx_ref(mdctx); - f->mdctx = mdctx; f->merge_args = grpc_channel_args_copy(args); f->master = channel; GRPC_CHANNEL_INTERNAL_REF(f->master, "subchannel_factory"); resolver = grpc_resolver_create(target, &f->base); if (!resolver) { + GRPC_CHANNEL_INTERNAL_UNREF(&exec_ctx, f->master, "subchannel_factory"); + grpc_subchannel_factory_unref(&exec_ctx, &f->base); + grpc_exec_ctx_finish(&exec_ctx); return NULL; } diff --git a/src/core/surface/byte_buffer_queue.h b/src/core/surface/channel_ping.c index 2c3b22d24e..b4ce282787 100644 --- a/src/core/surface/byte_buffer_queue.h +++ b/src/core/surface/channel_ping.c @@ -31,32 +31,49 @@ * */ -#ifndef GRPC_INTERNAL_CORE_SURFACE_BYTE_BUFFER_QUEUE_H -#define GRPC_INTERNAL_CORE_SURFACE_BYTE_BUFFER_QUEUE_H +#include "src/core/surface/channel.h" -#include <grpc/byte_buffer.h> +#include <string.h> -/* TODO(ctiller): inline an element or two into this struct to avoid per-call - allocations */ -typedef struct { - grpc_byte_buffer **data; - size_t count; - size_t capacity; -} grpc_bbq_array; +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> + +#include "src/core/surface/api_trace.h" +#include "src/core/surface/completion_queue.h" -/* should be initialized by zeroing memory */ typedef struct { - size_t drain_pos; - grpc_bbq_array filling; - grpc_bbq_array draining; - size_t bytes; -} grpc_byte_buffer_queue; - -void grpc_bbq_destroy(grpc_byte_buffer_queue *q); -grpc_byte_buffer *grpc_bbq_pop(grpc_byte_buffer_queue *q); -void grpc_bbq_flush(grpc_byte_buffer_queue *q); -int grpc_bbq_empty(grpc_byte_buffer_queue *q); -void grpc_bbq_push(grpc_byte_buffer_queue *q, grpc_byte_buffer *bb); -size_t grpc_bbq_bytes(grpc_byte_buffer_queue *q); - -#endif /* GRPC_INTERNAL_CORE_SURFACE_BYTE_BUFFER_QUEUE_H */ + grpc_closure closure; + void *tag; + grpc_completion_queue *cq; + grpc_cq_completion completion_storage; +} ping_result; + +static void ping_destroy(grpc_exec_ctx *exec_ctx, void *arg, + grpc_cq_completion *storage) { + gpr_free(arg); +} + +static void ping_done(grpc_exec_ctx *exec_ctx, void *arg, int success) { + ping_result *pr = arg; + grpc_cq_end_op(exec_ctx, pr->cq, pr->tag, success, ping_destroy, pr, + &pr->completion_storage); +} + +void grpc_channel_ping(grpc_channel *channel, grpc_completion_queue *cq, + void *tag, void *reserved) { + grpc_transport_op op; + ping_result *pr = gpr_malloc(sizeof(*pr)); + grpc_channel_element *top_elem = + grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + GPR_ASSERT(reserved == NULL); + memset(&op, 0, sizeof(op)); + pr->tag = tag; + pr->cq = cq; + grpc_closure_init(&pr->closure, ping_done, pr); + op.send_ping = &pr->closure; + op.bind_pollset = grpc_cq_pollset(cq); + grpc_cq_begin_op(cq, tag); + top_elem->filter->start_transport_op(&exec_ctx, top_elem, &op); + grpc_exec_ctx_finish(&exec_ctx); +} diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index aa90b3f7f5..848a33adc3 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -71,9 +71,38 @@ struct grpc_completion_queue { int is_server_cq; int num_pluckers; plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS]; - grpc_closure pollset_destroy_done; + grpc_closure pollset_shutdown_done; + +#ifndef NDEBUG + void **outstanding_tags; + size_t outstanding_tag_count; + size_t outstanding_tag_capacity; +#endif + + grpc_completion_queue *next_free; }; +static gpr_mu g_freelist_mu; +grpc_completion_queue *g_freelist; + +static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc, + int success); + +void grpc_cq_global_init(void) { gpr_mu_init(&g_freelist_mu); } + +void grpc_cq_global_shutdown(void) { + gpr_mu_destroy(&g_freelist_mu); + while (g_freelist) { + grpc_completion_queue *next = g_freelist->next_free; + grpc_pollset_destroy(&g_freelist->pollset); +#ifndef NDEBUG + gpr_free(g_freelist->outstanding_tags); +#endif + gpr_free(g_freelist); + g_freelist = next; + } +} + struct grpc_cq_alarm { grpc_timer alarm; grpc_cq_completion completion; @@ -83,22 +112,48 @@ struct grpc_cq_alarm { void *tag; }; -static void on_pollset_destroy_done(grpc_exec_ctx *exec_ctx, void *cc, - int success); - grpc_completion_queue *grpc_completion_queue_create(void *reserved) { - grpc_completion_queue *cc = gpr_malloc(sizeof(grpc_completion_queue)); - GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved)); + grpc_completion_queue *cc; GPR_ASSERT(!reserved); - memset(cc, 0, sizeof(*cc)); + + GPR_TIMER_BEGIN("grpc_completion_queue_create", 0); + + GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved)); + + gpr_mu_lock(&g_freelist_mu); + if (g_freelist == NULL) { + gpr_mu_unlock(&g_freelist_mu); + + cc = gpr_malloc(sizeof(grpc_completion_queue)); + grpc_pollset_init(&cc->pollset); +#ifndef NDEBUG + cc->outstanding_tags = NULL; + cc->outstanding_tag_capacity = 0; +#endif + } else { + cc = g_freelist; + g_freelist = g_freelist->next_free; + gpr_mu_unlock(&g_freelist_mu); + /* pollset already initialized */ + } + /* Initial ref is dropped by grpc_completion_queue_shutdown */ gpr_ref_init(&cc->pending_events, 1); /* One for destroy(), one for pollset_shutdown */ gpr_ref_init(&cc->owning_refs, 2); - grpc_pollset_init(&cc->pollset); cc->completed_tail = &cc->completed_head; cc->completed_head.next = (gpr_uintptr)cc->completed_tail; - grpc_closure_init(&cc->pollset_destroy_done, on_pollset_destroy_done, cc); + cc->shutdown = 0; + cc->shutdown_called = 0; + cc->is_server_cq = 0; + cc->num_pluckers = 0; +#ifndef NDEBUG + cc->outstanding_tag_count = 0; +#endif + grpc_closure_init(&cc->pollset_shutdown_done, on_pollset_shutdown_done, cc); + + GPR_TIMER_END("grpc_completion_queue_create", 0); + return cc; } @@ -113,8 +168,8 @@ void grpc_cq_internal_ref(grpc_completion_queue *cc) { gpr_ref(&cc->owning_refs); } -static void on_pollset_destroy_done(grpc_exec_ctx *exec_ctx, void *arg, - int success) { +static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg, + int success) { grpc_completion_queue *cc = arg; GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy"); } @@ -129,15 +184,25 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) { #endif if (gpr_unref(&cc->owning_refs)) { GPR_ASSERT(cc->completed_head.next == (gpr_uintptr)&cc->completed_head); - grpc_pollset_destroy(&cc->pollset); - gpr_free(cc); + grpc_pollset_reset(&cc->pollset); + gpr_mu_lock(&g_freelist_mu); + cc->next_free = g_freelist; + g_freelist = cc; + gpr_mu_unlock(&g_freelist_mu); } } -void grpc_cq_begin_op(grpc_completion_queue *cc) { +void grpc_cq_begin_op(grpc_completion_queue *cc, void *tag) { #ifndef NDEBUG gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); GPR_ASSERT(!cc->shutdown_called); + if (cc->outstanding_tag_count == cc->outstanding_tag_capacity) { + cc->outstanding_tag_capacity = GPR_MAX(4, 2 * cc->outstanding_tag_capacity); + cc->outstanding_tags = + gpr_realloc(cc->outstanding_tags, sizeof(*cc->outstanding_tags) * + cc->outstanding_tag_capacity); + } + cc->outstanding_tags[cc->outstanding_tag_count++] = tag; gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); #endif gpr_ref(&cc->pending_events); @@ -154,6 +219,9 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, int shutdown; int i; grpc_pollset_worker *pluck_worker; +#ifndef NDEBUG + int found = 0; +#endif GPR_TIMER_BEGIN("grpc_cq_end_op", 0); @@ -164,6 +232,18 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, ((gpr_uintptr)&cc->completed_head) | ((gpr_uintptr)(success != 0)); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); +#ifndef NDEBUG + for (i = 0; i < (int)cc->outstanding_tag_count; i++) { + if (cc->outstanding_tags[i] == tag) { + cc->outstanding_tag_count--; + GPR_SWAP(void *, cc->outstanding_tags[i], + cc->outstanding_tags[cc->outstanding_tag_count]); + found = 1; + break; + } + } + GPR_ASSERT(found); +#endif shutdown = gpr_unref(&cc->pending_events); if (!shutdown) { cc->completed_tail->next = @@ -185,8 +265,8 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, GPR_ASSERT(!cc->shutdown); GPR_ASSERT(cc->shutdown_called); cc->shutdown = 1; + grpc_pollset_shutdown(exec_ctx, &cc->pollset, &cc->pollset_shutdown_done); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); - grpc_pollset_shutdown(exec_ctx, &cc->pollset, &cc->pollset_destroy_done); } GPR_TIMER_END("grpc_cq_end_op", 0); @@ -205,10 +285,10 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, GRPC_API_TRACE( "grpc_completion_queue_next(" "cc=%p, " - "deadline=gpr_timespec { tv_sec: %ld, tv_nsec: %d, clock_type: %d }, " + "deadline=gpr_timespec { tv_sec: %lld, tv_nsec: %d, clock_type: %d }, " "reserved=%p)", - 5, (cc, (long)deadline.tv_sec, deadline.tv_nsec, (int)deadline.clock_type, - reserved)); + 5, (cc, (long long)deadline.tv_sec, (int)deadline.tv_nsec, + (int)deadline.clock_type, reserved)); GPR_ASSERT(!reserved); deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); @@ -293,9 +373,9 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, GRPC_API_TRACE( "grpc_completion_queue_pluck(" "cc=%p, tag=%p, " - "deadline=gpr_timespec { tv_sec: %ld, tv_nsec: %d, clock_type: %d }, " + "deadline=gpr_timespec { tv_sec: %lld, tv_nsec: %d, clock_type: %d }, " "reserved=%p)", - 6, (cc, tag, (long)deadline.tv_sec, deadline.tv_nsec, + 6, (cc, tag, (long long)deadline.tv_sec, (int)deadline.tv_nsec, (int)deadline.clock_type, reserved)); GPR_ASSERT(!reserved); @@ -365,29 +445,31 @@ done: to zero here, then enter shutdown mode and wake up any waiters */ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + GPR_TIMER_BEGIN("grpc_completion_queue_shutdown", 0); GRPC_API_TRACE("grpc_completion_queue_shutdown(cc=%p)", 1, (cc)); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); if (cc->shutdown_called) { gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); + GPR_TIMER_END("grpc_completion_queue_shutdown", 0); return; } cc->shutdown_called = 1; - gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); - if (gpr_unref(&cc->pending_events)) { - gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); GPR_ASSERT(!cc->shutdown); cc->shutdown = 1; - gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); - grpc_pollset_shutdown(&exec_ctx, &cc->pollset, &cc->pollset_destroy_done); + grpc_pollset_shutdown(&exec_ctx, &cc->pollset, &cc->pollset_shutdown_done); } + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); grpc_exec_ctx_finish(&exec_ctx); + GPR_TIMER_END("grpc_completion_queue_shutdown", 0); } void grpc_completion_queue_destroy(grpc_completion_queue *cc) { GRPC_API_TRACE("grpc_completion_queue_destroy(cc=%p)", 1, (cc)); + GPR_TIMER_BEGIN("grpc_completion_queue_destroy", 0); grpc_completion_queue_shutdown(cc); GRPC_CQ_INTERNAL_UNREF(cc, "destroy"); + GPR_TIMER_END("grpc_completion_queue_destroy", 0); } grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) { diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h index 5f8282e542..1e40c48bea 100644 --- a/src/core/surface/completion_queue.h +++ b/src/core/surface/completion_queue.h @@ -68,10 +68,12 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc); #endif /* Flag that an operation is beginning: the completion channel will not finish - shutdown until a corrensponding grpc_cq_end_* call is made */ -void grpc_cq_begin_op(grpc_completion_queue *cc); + shutdown until a corrensponding grpc_cq_end_* call is made. + \a tag is currently used only in debug builds. */ +void grpc_cq_begin_op(grpc_completion_queue *cc, void *tag); -/* Queue a GRPC_OP_COMPLETED operation */ +/* Queue a GRPC_OP_COMPLETED operation; tag must correspond to the tag passed to + grpc_cq_begin_op */ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc, void *tag, int success, void (*done)(grpc_exec_ctx *exec_ctx, void *done_arg, @@ -83,4 +85,7 @@ grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc); void grpc_cq_mark_server_cq(grpc_completion_queue *cc); int grpc_cq_is_server_cq(grpc_completion_queue *cc); +void grpc_cq_global_init(void); +void grpc_cq_global_shutdown(void); + #endif /* GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H */ diff --git a/src/core/surface/init.c b/src/core/surface/init.c index b2e66a830e..82027af651 100644 --- a/src/core/surface/init.c +++ b/src/core/surface/init.c @@ -52,11 +52,16 @@ #include "src/core/profiling/timers.h" #include "src/core/surface/api_trace.h" #include "src/core/surface/call.h" +#include "src/core/surface/completion_queue.h" #include "src/core/surface/init.h" #include "src/core/surface/surface_trace.h" #include "src/core/transport/chttp2_transport.h" #include "src/core/transport/connectivity_state.h" +#ifndef GRPC_DEFAULT_NAME_PREFIX +#define GRPC_DEFAULT_NAME_PREFIX "dns:///" +#endif + #define MAX_PLUGINS 128 static gpr_once g_basic_init = GPR_ONCE_INIT; @@ -92,10 +97,11 @@ void grpc_init(void) { gpr_mu_lock(&g_init_mu); if (++g_initializations == 1) { gpr_time_init(); + grpc_mdctx_global_init(); grpc_lb_policy_registry_init(grpc_pick_first_lb_factory_create()); grpc_register_lb_policy(grpc_pick_first_lb_factory_create()); grpc_register_lb_policy(grpc_round_robin_lb_factory_create()); - grpc_resolver_registry_init("dns:///"); + grpc_resolver_registry_init(GRPC_DEFAULT_NAME_PREFIX); grpc_register_resolver_type(grpc_dns_resolver_factory_create()); grpc_register_resolver_type(grpc_ipv4_resolver_factory_create()); grpc_register_resolver_type(grpc_ipv6_resolver_factory_create()); @@ -118,6 +124,7 @@ void grpc_init(void) { } } gpr_timers_global_init(); + grpc_cq_global_init(); for (i = 0; i < g_number_of_plugins; i++) { if (g_all_of_the_plugins[i].init != NULL) { g_all_of_the_plugins[i].init(); @@ -133,17 +140,20 @@ void grpc_shutdown(void) { GRPC_API_TRACE("grpc_shutdown(void)", 0, ()); gpr_mu_lock(&g_init_mu); if (--g_initializations == 0) { - grpc_iomgr_shutdown(); grpc_executor_shutdown(); + grpc_cq_global_shutdown(); + grpc_iomgr_shutdown(); census_shutdown(); gpr_timers_global_destroy(); grpc_tracer_shutdown(); grpc_resolver_registry_shutdown(); + grpc_lb_policy_registry_shutdown(); for (i = 0; i < g_number_of_plugins; i++) { if (g_all_of_the_plugins[i].destroy != NULL) { g_all_of_the_plugins[i].destroy(); } } + grpc_mdctx_global_shutdown(); } gpr_mu_unlock(&g_init_mu); } diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c index e72264fbcd..a60e9d20da 100644 --- a/src/core/surface/lame_client.c +++ b/src/core/surface/lame_client.c @@ -49,49 +49,41 @@ typedef struct { } call_data; typedef struct { - grpc_mdctx *mdctx; - grpc_channel *master; grpc_status_code error_code; const char *error_message; } channel_data; +static void fill_metadata(grpc_call_element *elem, grpc_metadata_batch *mdb) { + call_data *calld = elem->call_data; + channel_data *chand = elem->channel_data; + char tmp[GPR_LTOA_MIN_BUFSIZE]; + gpr_ltoa(chand->error_code, tmp); + calld->status.md = grpc_mdelem_from_strings("grpc-status", tmp); + calld->details.md = + grpc_mdelem_from_strings("grpc-message", chand->error_message); + calld->status.prev = calld->details.next = NULL; + calld->status.next = &calld->details; + calld->details.prev = &calld->status; + mdb->list.head = &calld->status; + mdb->list.tail = &calld->details; + mdb->deadline = gpr_inf_future(GPR_CLOCK_REALTIME); +} + static void lame_start_transport_stream_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op *op) { - call_data *calld = elem->call_data; - channel_data *chand = elem->channel_data; GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - if (op->send_ops != NULL) { - grpc_stream_ops_unref_owned_objects(op->send_ops->ops, op->send_ops->nops); - op->on_done_send->cb(exec_ctx, op->on_done_send->cb_arg, 0); - } - if (op->recv_ops != NULL) { - char tmp[GPR_LTOA_MIN_BUFSIZE]; - grpc_metadata_batch mdb; - gpr_ltoa(chand->error_code, tmp); - calld->status.md = - grpc_mdelem_from_strings(chand->mdctx, "grpc-status", tmp); - calld->details.md = grpc_mdelem_from_strings(chand->mdctx, "grpc-message", - chand->error_message); - calld->status.prev = calld->details.next = NULL; - calld->status.next = &calld->details; - calld->details.prev = &calld->status; - mdb.list.head = &calld->status; - mdb.list.tail = &calld->details; - mdb.garbage.head = mdb.garbage.tail = NULL; - mdb.deadline = gpr_inf_future(GPR_CLOCK_REALTIME); - grpc_sopb_add_metadata(op->recv_ops, mdb); - *op->recv_state = GRPC_STREAM_CLOSED; - op->on_done_recv->cb(exec_ctx, op->on_done_recv->cb_arg, 1); - } - if (op->on_consumed != NULL) { - op->on_consumed->cb(exec_ctx, op->on_consumed->cb_arg, 0); + if (op->recv_initial_metadata != NULL) { + fill_metadata(elem, op->recv_initial_metadata); + } else if (op->recv_trailing_metadata != NULL) { + fill_metadata(elem, op->recv_trailing_metadata); } + grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, 0); + grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, 0); } static char *lame_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { - channel_data *chand = elem->channel_data; - return grpc_channel_get_target(chand->master); + return NULL; } static void lame_start_transport_op(grpc_exec_ctx *exec_ctx, @@ -109,25 +101,16 @@ static void lame_start_transport_op(grpc_exec_ctx *exec_ctx, } static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const void *transport_server_data, - grpc_transport_stream_op *initial_op) { - if (initial_op) { - grpc_transport_stream_op_finish_with_failure(exec_ctx, initial_op); - } -} + grpc_call_element_args *args) {} static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {} static void init_channel_elem(grpc_exec_ctx *exec_ctx, - grpc_channel_element *elem, grpc_channel *master, - const grpc_channel_args *args, grpc_mdctx *mdctx, - int is_first, int is_last) { - channel_data *chand = elem->channel_data; - GPR_ASSERT(is_first); - GPR_ASSERT(is_last); - chand->mdctx = mdctx; - chand->master = master; + grpc_channel_element *elem, + grpc_channel_element_args *args) { + GPR_ASSERT(args->is_first); + GPR_ASSERT(args->is_last); } static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, @@ -135,8 +118,9 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, static const grpc_channel_filter lame_filter = { lame_start_transport_stream_op, lame_start_transport_op, sizeof(call_data), - init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem, - destroy_channel_elem, lame_get_peer, "lame-client", + init_call_elem, grpc_call_stack_ignore_set_pollset, destroy_call_elem, + sizeof(channel_data), init_channel_elem, destroy_channel_elem, + lame_get_peer, "lame-client", }; #define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack *)((c) + 1)) @@ -149,8 +133,8 @@ grpc_channel *grpc_lame_client_channel_create(const char *target, channel_data *chand; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; static const grpc_channel_filter *filters[] = {&lame_filter}; - channel = grpc_channel_create_from_filters(&exec_ctx, target, filters, 1, - NULL, grpc_mdctx_create(), 1); + channel = + grpc_channel_create_from_filters(&exec_ctx, target, filters, 1, NULL, 1); elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0); GRPC_API_TRACE( "grpc_lame_client_channel_create(target=%s, error_code=%d, " diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index 0dd0a31169..92bd53411d 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -37,6 +37,8 @@ #include <string.h> #include <grpc/support/alloc.h> +#include <grpc/support/slice.h> +#include <grpc/support/slice_buffer.h> #include "src/core/census/grpc_filter.h" #include "src/core/channel/channel_args.h" @@ -61,14 +63,14 @@ typedef struct { grpc_closure *notify; grpc_connect_in_args args; grpc_connect_out_args *result; + grpc_closure initial_string_sent; + gpr_slice_buffer initial_string_buffer; gpr_mu mu; grpc_endpoint *connecting_endpoint; grpc_endpoint *newly_connecting_endpoint; grpc_closure connected_closure; - - grpc_mdctx *mdctx; } connector; static void connector_ref(grpc_connector *con) { @@ -79,14 +81,13 @@ static void connector_ref(grpc_connector *con) { static void connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) { connector *c = (connector *)con; if (gpr_unref(&c->refs)) { - grpc_mdctx_unref(c->mdctx); + /* c->initial_string_buffer does not need to be destroyed */ gpr_free(c); } } static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, grpc_security_status status, - grpc_endpoint *wrapped_endpoint, grpc_endpoint *secure_endpoint) { connector *c = arg; grpc_closure *notify; @@ -95,17 +96,15 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, memset(c->result, 0, sizeof(*c->result)); gpr_mu_unlock(&c->mu); } else if (status != GRPC_SECURITY_OK) { - GPR_ASSERT(c->connecting_endpoint == wrapped_endpoint); gpr_log(GPR_ERROR, "Secure handshake failed with error %d.", status); memset(c->result, 0, sizeof(*c->result)); c->connecting_endpoint = NULL; gpr_mu_unlock(&c->mu); } else { - GPR_ASSERT(c->connecting_endpoint == wrapped_endpoint); c->connecting_endpoint = NULL; gpr_mu_unlock(&c->mu); c->result->transport = grpc_create_chttp2_transport( - exec_ctx, c->args.channel_args, secure_endpoint, c->mdctx, 1); + exec_ctx, c->args.channel_args, secure_endpoint, 1); grpc_chttp2_transport_start_reading(exec_ctx, c->result->transport, NULL, 0); c->result->filters = gpr_malloc(sizeof(grpc_channel_filter *) * 2); @@ -118,6 +117,14 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, notify->cb(exec_ctx, notify->cb_arg, 1); } +static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg, + int success) { + connector *c = arg; + grpc_security_connector_do_handshake(exec_ctx, &c->security_connector->base, + c->connecting_endpoint, + on_secure_handshake_done, c); +} + static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) { connector *c = arg; grpc_closure *notify; @@ -127,8 +134,19 @@ static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) { GPR_ASSERT(c->connecting_endpoint == NULL); c->connecting_endpoint = tcp; gpr_mu_unlock(&c->mu); - grpc_security_connector_do_handshake(exec_ctx, &c->security_connector->base, - tcp, on_secure_handshake_done, c); + if (!GPR_SLICE_IS_EMPTY(c->args.initial_connect_string)) { + grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent, + c); + gpr_slice_buffer_init(&c->initial_string_buffer); + gpr_slice_buffer_add(&c->initial_string_buffer, + c->args.initial_connect_string); + grpc_endpoint_write(exec_ctx, tcp, &c->initial_string_buffer, + &c->initial_string_sent); + } else { + grpc_security_connector_do_handshake(exec_ctx, + &c->security_connector->base, tcp, + on_secure_handshake_done, c); + } } else { memset(c->result, 0, sizeof(*c->result)); notify = c->notify; @@ -174,7 +192,6 @@ static const grpc_connector_vtable connector_vtable = { typedef struct { grpc_subchannel_factory base; gpr_refcount refs; - grpc_mdctx *mdctx; grpc_channel_args *merge_args; grpc_channel_security_connector *security_connector; grpc_channel *master; @@ -193,7 +210,6 @@ static void subchannel_factory_unref(grpc_exec_ctx *exec_ctx, "subchannel_factory"); GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, f->master, "subchannel_factory"); grpc_channel_args_destroy(f->merge_args); - grpc_mdctx_unref(f->mdctx); gpr_free(f); } } @@ -209,13 +225,9 @@ static grpc_subchannel *subchannel_factory_create_subchannel( memset(c, 0, sizeof(*c)); c->base.vtable = &connector_vtable; c->security_connector = f->security_connector; - c->mdctx = f->mdctx; gpr_mu_init(&c->mu); - grpc_mdctx_ref(c->mdctx); gpr_ref_init(&c->refs, 1); args->args = final_args; - args->master = f->master; - args->mdctx = f->mdctx; s = grpc_subchannel_create(&c->base, args); grpc_connector_unref(exec_ctx, &c->base); grpc_channel_args_destroy(final_args); @@ -239,7 +251,6 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds, grpc_channel_args *args_copy; grpc_channel_args *new_args_from_connector; grpc_channel_security_connector *security_connector; - grpc_mdctx *mdctx; grpc_resolver *resolver; subchannel_factory *f; #define MAX_FILTERS 3 @@ -269,7 +280,6 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds, target, GRPC_STATUS_INVALID_ARGUMENT, "Failed to create security connector."); } - mdctx = grpc_mdctx_create(); connector_arg = grpc_security_connector_to_arg(&security_connector->base); args_copy = grpc_channel_args_copy_and_add( @@ -283,35 +293,33 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds, GPR_ASSERT(n <= MAX_FILTERS); channel = grpc_channel_create_from_filters(&exec_ctx, target, filters, n, - args_copy, mdctx, 1); + args_copy, 1); f = gpr_malloc(sizeof(*f)); f->base.vtable = &subchannel_factory_vtable; gpr_ref_init(&f->refs, 1); - grpc_mdctx_ref(mdctx); - f->mdctx = mdctx; GRPC_SECURITY_CONNECTOR_REF(&security_connector->base, "subchannel_factory"); f->security_connector = security_connector; f->merge_args = grpc_channel_args_copy(args_copy); f->master = channel; GRPC_CHANNEL_INTERNAL_REF(channel, "subchannel_factory"); resolver = grpc_resolver_create(target, &f->base); - if (!resolver) { - grpc_exec_ctx_finish(&exec_ctx); - return NULL; + if (resolver) { + grpc_client_channel_set_resolver( + &exec_ctx, grpc_channel_get_channel_stack(channel), resolver); + GRPC_RESOLVER_UNREF(&exec_ctx, resolver, "create"); } - - grpc_client_channel_set_resolver( - &exec_ctx, grpc_channel_get_channel_stack(channel), resolver); - GRPC_RESOLVER_UNREF(&exec_ctx, resolver, "create"); grpc_subchannel_factory_unref(&exec_ctx, &f->base); GRPC_SECURITY_CONNECTOR_UNREF(&security_connector->base, "channel_create"); - grpc_channel_args_destroy(args_copy); if (new_args_from_connector != NULL) { grpc_channel_args_destroy(new_args_from_connector); } + if (!resolver) { + GRPC_CHANNEL_INTERNAL_UNREF(&exec_ctx, channel, "subchannel_factory"); + channel = NULL; + } grpc_exec_ctx_finish(&exec_ctx); return channel; diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 819226278d..1e1cde3648 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -54,6 +54,7 @@ #include "src/core/surface/completion_queue.h" #include "src/core/surface/init.h" #include "src/core/transport/metadata.h" +#include "src/core/transport/static_metadata.h" typedef struct listener { void *arg; @@ -84,18 +85,18 @@ typedef struct requested_call { grpc_completion_queue *cq_for_notification; grpc_call **call; grpc_cq_completion completion; + grpc_metadata_array *initial_metadata; union { struct { grpc_call_details *details; - grpc_metadata_array *initial_metadata; } batch; struct { registered_method *registered_method; gpr_timespec *deadline; - grpc_metadata_array *initial_metadata; grpc_byte_buffer **optional_payload; } registered; } data; + grpc_closure publish; } requested_call; typedef struct channel_registered_method { @@ -108,8 +109,6 @@ struct channel_data { grpc_server *server; grpc_connectivity_state connectivity_state; grpc_channel *channel; - grpc_mdstr *path_key; - grpc_mdstr *authority_key; /* linked list of all channels on a server */ channel_data *next; channel_data *prev; @@ -150,16 +149,16 @@ struct call_data { grpc_mdstr *path; grpc_mdstr *host; gpr_timespec deadline; - int got_initial_metadata; grpc_completion_queue *cq_new; - grpc_stream_op_buffer *recv_ops; - grpc_stream_state *recv_state; - grpc_closure *on_done_recv; + grpc_metadata_batch *recv_initial_metadata; + grpc_metadata_array initial_metadata; - grpc_closure server_on_recv; + grpc_closure got_initial_metadata; + grpc_closure server_on_recv_initial_metadata; grpc_closure kill_zombie_closure; + grpc_closure *on_done_recv_initial_metadata; call_data *pending_next; }; @@ -396,7 +395,6 @@ static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd, int success) { channel_data *chand = cd; grpc_server *server = chand->server; - gpr_log(GPR_DEBUG, "finish_destroy_channel: %p", chand->channel); GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "server"); server_unref(exec_ctx, server); } @@ -559,91 +557,46 @@ static void maybe_finish_shutdown(grpc_exec_ctx *exec_ctx, static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { grpc_call_element *elem = user_data; - channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; - if (md->key == chand->path_key) { + if (md->key == GRPC_MDSTR_PATH) { calld->path = GRPC_MDSTR_REF(md->value); return NULL; - } else if (md->key == chand->authority_key) { + } else if (md->key == GRPC_MDSTR_AUTHORITY) { calld->host = GRPC_MDSTR_REF(md->value); return NULL; } return md; } -static void server_on_recv(grpc_exec_ctx *exec_ctx, void *ptr, int success) { +static void server_on_recv_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr, + int success) { grpc_call_element *elem = ptr; call_data *calld = elem->call_data; gpr_timespec op_deadline; - if (success && !calld->got_initial_metadata) { - size_t i; - size_t nops = calld->recv_ops->nops; - grpc_stream_op *ops = calld->recv_ops->ops; - for (i = 0; i < nops; i++) { - grpc_stream_op *op = &ops[i]; - if (op->type != GRPC_OP_METADATA) continue; - grpc_metadata_batch_filter(&op->data.metadata, server_filter, elem); - op_deadline = op->data.metadata.deadline; - if (0 != - gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) { - calld->deadline = op->data.metadata.deadline; - } - if (calld->host && calld->path) { - calld->got_initial_metadata = 1; - start_new_rpc(exec_ctx, elem); - } - break; - } + grpc_metadata_batch_filter(calld->recv_initial_metadata, server_filter, elem); + op_deadline = calld->recv_initial_metadata->deadline; + if (0 != gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) { + calld->deadline = op_deadline; } - - switch (*calld->recv_state) { - case GRPC_STREAM_OPEN: - break; - case GRPC_STREAM_SEND_CLOSED: - break; - case GRPC_STREAM_RECV_CLOSED: - gpr_mu_lock(&calld->mu_state); - if (calld->state == NOT_STARTED) { - calld->state = ZOMBIED; - gpr_mu_unlock(&calld->mu_state); - grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); - grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1); - } else { - gpr_mu_unlock(&calld->mu_state); - } - break; - case GRPC_STREAM_CLOSED: - gpr_mu_lock(&calld->mu_state); - if (calld->state == NOT_STARTED) { - calld->state = ZOMBIED; - gpr_mu_unlock(&calld->mu_state); - grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); - grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1); - } else if (calld->state == PENDING) { - calld->state = ZOMBIED; - gpr_mu_unlock(&calld->mu_state); - /* zombied call will be destroyed when it's removed from the pending - queue... later */ - } else { - gpr_mu_unlock(&calld->mu_state); - } - break; + if (calld->host && calld->path) { + /* do nothing */ + } else { + success = 0; } - calld->on_done_recv->cb(exec_ctx, calld->on_done_recv->cb_arg, success); + calld->on_done_recv_initial_metadata->cb( + exec_ctx, calld->on_done_recv_initial_metadata->cb_arg, success); } static void server_mutate_op(grpc_call_element *elem, grpc_transport_stream_op *op) { call_data *calld = elem->call_data; - if (op->recv_ops) { - /* substitute our callback for the higher callback */ - calld->recv_ops = op->recv_ops; - calld->recv_state = op->recv_state; - calld->on_done_recv = op->on_done_recv; - op->on_done_recv = &calld->server_on_recv; + if (op->recv_initial_metadata != NULL) { + calld->recv_initial_metadata = op->recv_initial_metadata; + calld->on_done_recv_initial_metadata = op->on_complete; + op->on_complete = &calld->server_on_recv_initial_metadata; } } @@ -655,12 +608,48 @@ static void server_start_transport_stream_op(grpc_exec_ctx *exec_ctx, grpc_call_next_op(exec_ctx, elem, op); } -static void accept_stream(void *cd, grpc_transport *transport, +static void got_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr, + int success) { + grpc_call_element *elem = ptr; + call_data *calld = elem->call_data; + if (success) { + start_new_rpc(exec_ctx, elem); + } else { + gpr_mu_lock(&calld->mu_state); + if (calld->state == NOT_STARTED) { + calld->state = ZOMBIED; + gpr_mu_unlock(&calld->mu_state); + grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem); + grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1); + } else if (calld->state == PENDING) { + calld->state = ZOMBIED; + gpr_mu_unlock(&calld->mu_state); + /* zombied call will be destroyed when it's removed from the pending + queue... later */ + } else { + gpr_mu_unlock(&calld->mu_state); + } + } +} + +static void accept_stream(grpc_exec_ctx *exec_ctx, void *cd, + grpc_transport *transport, const void *transport_server_data) { channel_data *chand = cd; /* create a call */ - grpc_call_create(chand->channel, NULL, 0, NULL, transport_server_data, NULL, - 0, gpr_inf_future(GPR_CLOCK_MONOTONIC)); + grpc_call *call = + grpc_call_create(chand->channel, NULL, 0, NULL, transport_server_data, + NULL, 0, gpr_inf_future(GPR_CLOCK_MONOTONIC)); + grpc_call_element *elem = + grpc_call_stack_element(grpc_call_get_call_stack(call), 0); + call_data *calld = elem->call_data; + grpc_op op; + memset(&op, 0, sizeof(op)); + op.op = GRPC_OP_RECV_INITIAL_METADATA; + op.data.recv_initial_metadata = &calld->initial_metadata; + grpc_closure_init(&calld->got_initial_metadata, got_initial_metadata, elem); + grpc_call_start_batch_and_execute(exec_ctx, call, &op, 1, + &calld->got_initial_metadata); } static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd, @@ -685,8 +674,7 @@ static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd, } static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const void *server_transport_data, - grpc_transport_stream_op *initial_op) { + grpc_call_element_args *args) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; memset(calld, 0, sizeof(call_data)); @@ -694,11 +682,10 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, calld->call = grpc_call_from_top_element(elem); gpr_mu_init(&calld->mu_state); - grpc_closure_init(&calld->server_on_recv, server_on_recv, elem); + grpc_closure_init(&calld->server_on_recv_initial_metadata, + server_on_recv_initial_metadata, elem); server_ref(chand->server); - - if (initial_op) server_mutate_op(elem, initial_op); } static void destroy_call_elem(grpc_exec_ctx *exec_ctx, @@ -714,6 +701,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, if (calld->path) { GRPC_MDSTR_UNREF(calld->path); } + grpc_metadata_array_destroy(&calld->initial_metadata); gpr_mu_destroy(&calld->mu_state); @@ -721,17 +709,13 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, } static void init_channel_elem(grpc_exec_ctx *exec_ctx, - grpc_channel_element *elem, grpc_channel *master, - const grpc_channel_args *args, - grpc_mdctx *metadata_context, int is_first, - int is_last) { + grpc_channel_element *elem, + grpc_channel_element_args *args) { channel_data *chand = elem->channel_data; - GPR_ASSERT(is_first); - GPR_ASSERT(!is_last); + GPR_ASSERT(args->is_first); + GPR_ASSERT(!args->is_last); chand->server = NULL; chand->channel = NULL; - chand->path_key = grpc_mdstr_from_string(metadata_context, ":path"); - chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority"); chand->next = chand->prev = chand; chand->registered_methods = NULL; chand->connectivity_state = GRPC_CHANNEL_IDLE; @@ -761,16 +745,15 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, chand->next = chand->prev = chand; maybe_finish_shutdown(exec_ctx, chand->server); gpr_mu_unlock(&chand->server->mu_global); - GRPC_MDSTR_UNREF(chand->path_key); - GRPC_MDSTR_UNREF(chand->authority_key); server_unref(exec_ctx, chand->server); } } static const grpc_channel_filter server_surface_filter = { server_start_transport_stream_op, grpc_channel_next_op, sizeof(call_data), - init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem, - destroy_channel_elem, grpc_call_next_get_peer, "server", + init_call_elem, grpc_call_stack_ignore_set_pollset, destroy_call_elem, + sizeof(channel_data), init_channel_elem, destroy_channel_elem, + grpc_call_next_get_peer, "server", }; void grpc_server_register_completion_queue(grpc_server *server, @@ -904,7 +887,7 @@ void grpc_server_start(grpc_server *server) { void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, grpc_transport *transport, grpc_channel_filter const **extra_filters, - size_t num_extra_filters, grpc_mdctx *mdctx, + size_t num_extra_filters, const grpc_channel_args *args) { size_t num_filters = s->channel_filter_count + num_extra_filters + 1; grpc_channel_filter const **filters = @@ -939,7 +922,7 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, } channel = grpc_channel_create_from_filters(exec_ctx, NULL, filters, - num_filters, args, mdctx, 0); + num_filters, args, 0); chand = (channel_data *)grpc_channel_stack_element( grpc_channel_get_channel_stack(channel), 0)->channel_data; chand->server = s; @@ -958,8 +941,8 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, chand->registered_methods = gpr_malloc(alloc); memset(chand->registered_methods, 0, alloc); for (rm = s->registered_methods; rm; rm = rm->next) { - host = rm->host ? grpc_mdstr_from_string(mdctx, rm->host) : NULL; - method = grpc_mdstr_from_string(mdctx, rm->method); + host = rm->host ? grpc_mdstr_from_string(rm->host) : NULL; + method = grpc_mdstr_from_string(rm->method); hash = GRPC_MDSTR_KV_HASH(host ? host->hash : 0, method->hash); for (probes = 0; chand->registered_methods[(hash + probes) % slots] .server_registered_method != NULL; @@ -1022,11 +1005,9 @@ void grpc_server_shutdown_and_notify(grpc_server *server, GRPC_API_TRACE("grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", 3, (server, cq, tag)); - GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag); - /* lock, and gather up some stuff to do */ gpr_mu_lock(&server->mu_global); - grpc_cq_begin_op(cq); + grpc_cq_begin_op(cq, tag); if (server->shutdown_published) { grpc_cq_end_op(&exec_ctx, cq, tag, 1, done_published_shutdown, NULL, gpr_malloc(sizeof(grpc_cq_completion))); @@ -1187,18 +1168,15 @@ grpc_call_error grpc_server_request_call( GRPC_API_TRACE( "grpc_server_request_call(" "server=%p, call=%p, details=%p, initial_metadata=%p, " - "cq_bound_to_call=%p, cq_for_notification=%p, tag%p)", + "cq_bound_to_call=%p, cq_for_notification=%p, tag=%p)", 7, (server, call, details, initial_metadata, cq_bound_to_call, cq_for_notification, tag)); - GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details, - initial_metadata, cq_bound_to_call, - cq_for_notification, tag); if (!grpc_cq_is_server_cq(cq_for_notification)) { gpr_free(rc); error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; goto done; } - grpc_cq_begin_op(cq_for_notification); + grpc_cq_begin_op(cq_for_notification, tag); details->reserved = NULL; rc->type = BATCH_CALL; rc->server = server; @@ -1207,7 +1185,7 @@ grpc_call_error grpc_server_request_call( rc->cq_for_notification = cq_for_notification; rc->call = call; rc->data.batch.details = details; - rc->data.batch.initial_metadata = initial_metadata; + rc->initial_metadata = initial_metadata; error = queue_call_request(&exec_ctx, server, rc); done: grpc_exec_ctx_finish(&exec_ctx); @@ -1235,7 +1213,7 @@ grpc_call_error grpc_server_request_registered_call( error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; goto done; } - grpc_cq_begin_op(cq_for_notification); + grpc_cq_begin_op(cq_for_notification, tag); rc->type = REGISTERED_CALL; rc->server = server; rc->tag = tag; @@ -1244,7 +1222,7 @@ grpc_call_error grpc_server_request_registered_call( rc->call = call; rc->data.registered.registered_method = rm; rc->data.registered.deadline = deadline; - rc->data.registered.initial_metadata = initial_metadata; + rc->initial_metadata = initial_metadata; rc->data.registered.optional_payload = optional_payload; error = queue_call_request(&exec_ctx, server, rc); done: @@ -1253,12 +1231,7 @@ done: } static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx, - grpc_call *call, int success, - void *tag); -static void publish_was_not_set(grpc_exec_ctx *exec_ctx, grpc_call *call, - int success, void *tag) { - abort(); -} + void *user_data, int success); static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) { gpr_slice slice = value->slice; @@ -1273,9 +1246,10 @@ static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) { static void begin_call(grpc_exec_ctx *exec_ctx, grpc_server *server, call_data *calld, requested_call *rc) { - grpc_ioreq_completion_func publish = publish_was_not_set; - grpc_ioreq req[2]; - grpc_ioreq *r = req; + grpc_op ops[1]; + grpc_op *op = ops; + + memset(ops, 0, sizeof(ops)); /* called once initial metadata has been read by the call, but BEFORE the ioreq to fetch it out of the call has been executed. @@ -1284,8 +1258,10 @@ static void begin_call(grpc_exec_ctx *exec_ctx, grpc_server *server, an ioreq op, that should complete immediately. */ grpc_call_set_completion_queue(exec_ctx, calld->call, rc->cq_bound_to_call); + grpc_closure_init(&rc->publish, publish_registered_or_batch, rc); *rc->call = calld->call; calld->cq_new = rc->cq_for_notification; + GPR_SWAP(grpc_metadata_array, *rc->initial_metadata, calld->initial_metadata); switch (rc->type) { case BATCH_CALL: GPR_ASSERT(calld->host != NULL); @@ -1295,31 +1271,22 @@ static void begin_call(grpc_exec_ctx *exec_ctx, grpc_server *server, cpstr(&rc->data.batch.details->method, &rc->data.batch.details->method_capacity, calld->path); rc->data.batch.details->deadline = calld->deadline; - r->op = GRPC_IOREQ_RECV_INITIAL_METADATA; - r->data.recv_metadata = rc->data.batch.initial_metadata; - r->flags = 0; - r++; - publish = publish_registered_or_batch; break; case REGISTERED_CALL: *rc->data.registered.deadline = calld->deadline; - r->op = GRPC_IOREQ_RECV_INITIAL_METADATA; - r->data.recv_metadata = rc->data.registered.initial_metadata; - r->flags = 0; - r++; if (rc->data.registered.optional_payload) { - r->op = GRPC_IOREQ_RECV_MESSAGE; - r->data.recv_message = rc->data.registered.optional_payload; - r->flags = 0; - r++; + op->op = GRPC_OP_RECV_MESSAGE; + op->data.recv_message = rc->data.registered.optional_payload; + op++; } - publish = publish_registered_or_batch; break; + default: + GPR_UNREACHABLE_CODE(return ); } GRPC_CALL_INTERNAL_REF(calld->call, "server"); - grpc_call_start_ioreq_and_call_back(exec_ctx, calld->call, req, - (size_t)(r - req), publish, rc); + grpc_call_start_batch_and_execute(exec_ctx, calld->call, ops, + (size_t)(op - ops), &rc->publish); } static void done_request_event(grpc_exec_ctx *exec_ctx, void *req, @@ -1342,25 +1309,19 @@ static void done_request_event(grpc_exec_ctx *exec_ctx, void *req, static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server, requested_call *rc) { *rc->call = NULL; - switch (rc->type) { - case BATCH_CALL: - rc->data.batch.initial_metadata->count = 0; - break; - case REGISTERED_CALL: - rc->data.registered.initial_metadata->count = 0; - break; - } + rc->initial_metadata->count = 0; + server_ref(server); grpc_cq_end_op(exec_ctx, rc->cq_for_notification, rc->tag, 0, done_request_event, rc, &rc->completion); } -static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx, - grpc_call *call, int success, - void *prc) { +static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx, void *prc, + int success) { + requested_call *rc = prc; + grpc_call *call = *rc->call; grpc_call_element *elem = grpc_call_stack_element(grpc_call_get_call_stack(call), 0); - requested_call *rc = prc; call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; server_ref(chand->server); diff --git a/src/core/surface/server.h b/src/core/surface/server.h index 4c46d07679..a957fdb360 100644 --- a/src/core/surface/server.h +++ b/src/core/surface/server.h @@ -57,7 +57,7 @@ void grpc_server_add_listener( void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *server, grpc_transport *transport, grpc_channel_filter const **extra_filters, - size_t num_extra_filters, grpc_mdctx *mdctx, + size_t num_extra_filters, const grpc_channel_args *args); const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server); diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c index 580b91573c..5ce7c1955b 100644 --- a/src/core/surface/server_chttp2.c +++ b/src/core/surface/server_chttp2.c @@ -44,11 +44,11 @@ #include <grpc/support/useful.h> static void setup_transport(grpc_exec_ctx *exec_ctx, void *server, - grpc_transport *transport, grpc_mdctx *mdctx) { + grpc_transport *transport) { static grpc_channel_filter const *extra_filters[] = { &grpc_http_server_filter}; grpc_server_setup_transport(exec_ctx, server, transport, extra_filters, - GPR_ARRAY_SIZE(extra_filters), mdctx, + GPR_ARRAY_SIZE(extra_filters), grpc_server_get_channel_args(server)); } @@ -61,10 +61,9 @@ static void new_transport(grpc_exec_ctx *exec_ctx, void *server, * (as in server_secure_chttp2.c) needs to add synchronization to avoid this * case. */ - grpc_mdctx *mdctx = grpc_mdctx_create(); grpc_transport *transport = grpc_create_chttp2_transport( - exec_ctx, grpc_server_get_channel_args(server), tcp, mdctx, 0); - setup_transport(exec_ctx, server, transport, mdctx); + exec_ctx, grpc_server_get_channel_args(server), tcp, 0); + setup_transport(exec_ctx, server, transport); grpc_chttp2_transport_start_reading(exec_ctx, transport, NULL, 0); } @@ -102,15 +101,15 @@ int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr) { } tcp = grpc_tcp_server_create(); - if (!tcp) { - goto error; - } + GPR_ASSERT(tcp); for (i = 0; i < resolved->naddrs; i++) { - port_temp = grpc_tcp_server_add_port( + grpc_tcp_listener *listener; + listener = grpc_tcp_server_add_port( tcp, (struct sockaddr *)&resolved->addrs[i].addr, resolved->addrs[i].len); - if (port_temp >= 0) { + port_temp = grpc_tcp_listener_get_port(listener); + if (port_temp > 0) { if (port_num == -1) { port_num = port_temp; } else { diff --git a/src/core/surface/server_create.c b/src/core/surface/server_create.c index c7811a6d88..f30093e06b 100644 --- a/src/core/surface/server_create.c +++ b/src/core/surface/server_create.c @@ -32,14 +32,20 @@ */ #include <grpc/grpc.h> +#include "src/core/census/grpc_filter.h" +#include "src/core/channel/channel_args.h" +#include "src/core/channel/compress_filter.h" #include "src/core/surface/api_trace.h" #include "src/core/surface/completion_queue.h" #include "src/core/surface/server.h" -#include "src/core/channel/compress_filter.h" grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) { - const grpc_channel_filter *filters[] = {&grpc_compress_filter}; + const grpc_channel_filter *filters[3]; + size_t num_filters = 0; + filters[num_filters++] = &grpc_compress_filter; + if (grpc_channel_args_is_census_enabled(args)) { + filters[num_filters++] = &grpc_server_census_filter; + } GRPC_API_TRACE("grpc_server_create(%p, %p)", 2, (args, reserved)); - return grpc_server_create_from_filters(filters, GPR_ARRAY_SIZE(filters), - args); + return grpc_server_create_from_filters(filters, num_filters, args); } |