diff options
author | Robbie Shade <rjshade@google.com> | 2016-10-04 16:39:39 -0400 |
---|---|---|
committer | Robbie Shade <rjshade@google.com> | 2016-10-04 16:39:39 -0400 |
commit | 35d7a26c61a4db831b32d21fa50bd0a957bfa0ff (patch) | |
tree | 48025170d2df50df0a216b77fb39ffe67ba20f04 /src/core/lib | |
parent | 26f90888d6d98322974a75dc8938dc070edea4c5 (diff) | |
parent | db1f7bc012d570b5909ee911b762a7968e857aea (diff) |
Merge branch 'master' into enable_udp_testing
Diffstat (limited to 'src/core/lib')
84 files changed, 2966 insertions, 661 deletions
diff --git a/src/core/lib/channel/channel_args.c b/src/core/lib/channel/channel_args.c index 79ceeb66b3..3a56b1ff20 100644 --- a/src/core/lib/channel/channel_args.c +++ b/src/core/lib/channel/channel_args.c @@ -271,3 +271,21 @@ int grpc_channel_args_compare(const grpc_channel_args *a, } return 0; } + +int grpc_channel_arg_get_integer(grpc_arg *arg, grpc_integer_options options) { + if (arg->type != GRPC_ARG_INTEGER) { + gpr_log(GPR_ERROR, "%s ignored: it must be an integer", arg->key); + return options.default_value; + } + if (arg->value.integer < options.min_value) { + gpr_log(GPR_ERROR, "%s ignored: it must be >= %d", arg->key, + options.min_value); + return options.default_value; + } + if (arg->value.integer > options.max_value) { + gpr_log(GPR_ERROR, "%s ignored: it must be <= %d", arg->key, + options.max_value); + return options.default_value; + } + return arg->value.integer; +} diff --git a/src/core/lib/channel/channel_args.h b/src/core/lib/channel/channel_args.h index aec61ee7c6..586a296d1f 100644 --- a/src/core/lib/channel/channel_args.h +++ b/src/core/lib/channel/channel_args.h @@ -89,4 +89,12 @@ uint32_t grpc_channel_args_compression_algorithm_get_states( int grpc_channel_args_compare(const grpc_channel_args *a, const grpc_channel_args *b); +typedef struct grpc_integer_options { + int default_value; // Return this if value is outside of expected bounds. + int min_value; + int max_value; +} grpc_integer_options; +/** Returns the value of \a arg, subject to the contraints in \a options. */ +int grpc_channel_arg_get_integer(grpc_arg *arg, grpc_integer_options options); + #endif /* GRPC_CORE_LIB_CHANNEL_CHANNEL_ARGS_H */ diff --git a/src/core/lib/channel/channel_stack.c b/src/core/lib/channel/channel_stack.c index 87175d7943..57d34d9e9a 100644 --- a/src/core/lib/channel/channel_stack.c +++ b/src/core/lib/channel/channel_stack.c @@ -32,6 +32,7 @@ */ #include "src/core/lib/channel/channel_stack.h" +#include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <stdlib.h> @@ -157,12 +158,11 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx, } } -void grpc_call_stack_init(grpc_exec_ctx *exec_ctx, - grpc_channel_stack *channel_stack, int initial_refs, - grpc_iomgr_cb_func destroy, void *destroy_arg, - grpc_call_context_element *context, - const void *transport_server_data, - grpc_call_stack *call_stack) { +grpc_error *grpc_call_stack_init( + grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack, + int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg, + grpc_call_context_element *context, const void *transport_server_data, + gpr_timespec deadline, grpc_call_stack *call_stack) { grpc_channel_element *channel_elems = CHANNEL_ELEMS_FROM_STACK(channel_stack); grpc_call_element_args args; size_t count = channel_stack->count; @@ -178,17 +178,28 @@ void grpc_call_stack_init(grpc_exec_ctx *exec_ctx, ROUND_UP_TO_ALIGNMENT_SIZE(count * sizeof(grpc_call_element)); /* init per-filter data */ + grpc_error *first_error = GRPC_ERROR_NONE; for (i = 0; i < count; i++) { args.call_stack = call_stack; args.server_transport_data = transport_server_data; args.context = context; + args.deadline = deadline; call_elems[i].filter = channel_elems[i].filter; call_elems[i].channel_data = channel_elems[i].channel_data; call_elems[i].call_data = user_data; - call_elems[i].filter->init_call_elem(exec_ctx, &call_elems[i], &args); + grpc_error *error = + call_elems[i].filter->init_call_elem(exec_ctx, &call_elems[i], &args); + if (error != GRPC_ERROR_NONE) { + if (first_error == GRPC_ERROR_NONE) { + first_error = error; + } else { + GRPC_ERROR_UNREF(error); + } + } user_data += ROUND_UP_TO_ALIGNMENT_SIZE(call_elems[i].filter->sizeof_call_data); } + return first_error; } void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, @@ -217,7 +228,7 @@ void grpc_call_stack_ignore_set_pollset_or_pollset_set( grpc_polling_entity *pollent) {} void grpc_call_stack_destroy(grpc_exec_ctx *exec_ctx, grpc_call_stack *stack, - const grpc_call_stats *call_stats, + const grpc_call_final_info *final_info, void *and_free_memory) { grpc_call_element *elems = CALL_ELEMS_FROM_STACK(stack); size_t count = stack->count; @@ -225,7 +236,7 @@ void grpc_call_stack_destroy(grpc_exec_ctx *exec_ctx, grpc_call_stack *stack, /* destroy per-filter data */ for (i = 0; i < count; i++) { - elems[i].filter->destroy_call_elem(exec_ctx, &elems[i], call_stats, + elems[i].filter->destroy_call_elem(exec_ctx, &elems[i], final_info, i == count - 1 ? and_free_memory : NULL); } } @@ -259,21 +270,38 @@ grpc_call_stack *grpc_call_stack_from_top_element(grpc_call_element *elem) { sizeof(grpc_call_stack))); } +static void destroy_op(grpc_exec_ctx *exec_ctx, void *op, grpc_error *error) { + gpr_free(op); +} + void grpc_call_element_send_cancel(grpc_exec_ctx *exec_ctx, - grpc_call_element *cur_elem) { - grpc_transport_stream_op op; - memset(&op, 0, sizeof(op)); - op.cancel_error = GRPC_ERROR_CANCELLED; - grpc_call_next_op(exec_ctx, cur_elem, &op); + grpc_call_element *elem) { + grpc_transport_stream_op *op = gpr_malloc(sizeof(*op)); + memset(op, 0, sizeof(*op)); + op->cancel_error = GRPC_ERROR_CANCELLED; + op->on_complete = grpc_closure_create(destroy_op, op); + elem->filter->start_transport_stream_op(exec_ctx, elem, op); } void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx, - grpc_call_element *cur_elem, + grpc_call_element *elem, grpc_status_code status, gpr_slice *optional_message) { - grpc_transport_stream_op op; - memset(&op, 0, sizeof(op)); - grpc_transport_stream_op_add_cancellation_with_message(&op, status, + grpc_transport_stream_op *op = gpr_malloc(sizeof(*op)); + memset(op, 0, sizeof(*op)); + op->on_complete = grpc_closure_create(destroy_op, op); + grpc_transport_stream_op_add_cancellation_with_message(op, status, optional_message); - grpc_call_next_op(exec_ctx, cur_elem, &op); + elem->filter->start_transport_stream_op(exec_ctx, elem, op); +} + +void grpc_call_element_send_close_with_message(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_status_code status, + gpr_slice *optional_message) { + grpc_transport_stream_op *op = gpr_malloc(sizeof(*op)); + memset(op, 0, sizeof(*op)); + op->on_complete = grpc_closure_create(destroy_op, op); + grpc_transport_stream_op_add_close(op, status, optional_message); + elem->filter->start_transport_stream_op(exec_ctx, elem, op); } diff --git a/src/core/lib/channel/channel_stack.h b/src/core/lib/channel/channel_stack.h index d72c015b67..1cfe2885d8 100644 --- a/src/core/lib/channel/channel_stack.h +++ b/src/core/lib/channel/channel_stack.h @@ -51,6 +51,10 @@ #include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/transport/transport.h" +#ifdef __cplusplus +extern "C" { +#endif + typedef struct grpc_channel_element grpc_channel_element; typedef struct grpc_call_element grpc_call_element; @@ -70,14 +74,20 @@ typedef struct { grpc_call_stack *call_stack; const void *server_transport_data; grpc_call_context_element *context; + gpr_timespec deadline; } grpc_call_element_args; typedef struct { grpc_transport_stream_stats transport_stream_stats; gpr_timespec latency; /* From call creating to enqueing of received status */ - grpc_status_code final_status; } grpc_call_stats; +/** Information about the call upon completion. */ +typedef struct { + grpc_call_stats stats; + grpc_status_code final_status; +} grpc_call_final_info; + /* Channel filters specify: 1. the amount of memory needed in the channel & call (via the sizeof_XXX members) @@ -110,8 +120,9 @@ typedef struct { on a client; if it is non-NULL, then it points to memory owned by the transport and is on the server. Most filters want to ignore this argument. */ - void (*init_call_elem)(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args); + grpc_error *(*init_call_elem)(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args); void (*set_pollset_or_pollset_set)(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_polling_entity *pollent); @@ -119,16 +130,17 @@ typedef struct { The filter does not need to do any chaining. The bottom filter of a stack will be passed a non-NULL pointer to \a and_free_memory that should be passed to gpr_free when destruction - is complete. */ + is complete. \a final_info contains data about the completed call, mainly + for reporting purposes. */ void (*destroy_call_elem)(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, + const grpc_call_final_info *final_info, void *and_free_memory); /* sizeof(per channel data) */ size_t sizeof_channel_data; /* Initialize per-channel data. - elem is initialized at the start of the call, and elem->channel_data is - what needs initializing. + elem is initialized at the creating of the channel, and elem->channel_data + is what needs initializing. is_first, is_last designate this elements position in the stack, and are useful for asserting correct configuration by upper layer code. The filter does not need to do any chaining */ @@ -209,12 +221,11 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx, /* Initialize a call stack given a channel stack. transport_server_data is expected to be NULL on a client, or an opaque transport owned pointer on the server. */ -void grpc_call_stack_init(grpc_exec_ctx *exec_ctx, - grpc_channel_stack *channel_stack, int initial_refs, - grpc_iomgr_cb_func destroy, void *destroy_arg, - grpc_call_context_element *context, - const void *transport_server_data, - grpc_call_stack *call_stack); +grpc_error *grpc_call_stack_init( + grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack, + int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg, + grpc_call_context_element *context, const void *transport_server_data, + gpr_timespec deadline, grpc_call_stack *call_stack); /* Set a pollset or a pollset_set for a call stack: must occur before the first * op is started */ void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, @@ -243,7 +254,7 @@ void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, /* Destroy a call stack */ void grpc_call_stack_destroy(grpc_exec_ctx *exec_ctx, grpc_call_stack *stack, - const grpc_call_stats *call_stats, + const grpc_call_final_info *final_info, void *and_free_memory); /* Ignore set pollset{_set} - used by filters if they don't care about pollsets @@ -278,9 +289,18 @@ void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx, grpc_status_code status, gpr_slice *optional_message); +void grpc_call_element_send_close_with_message(grpc_exec_ctx *exec_ctx, + grpc_call_element *cur_elem, + grpc_status_code status, + gpr_slice *optional_message); + extern int grpc_trace_channel; #define GRPC_CALL_LOG_OP(sev, elem, op) \ if (grpc_trace_channel) grpc_call_log_op(sev, elem, op) +#ifdef __cplusplus +} +#endif + #endif /* GRPC_CORE_LIB_CHANNEL_CHANNEL_STACK_H */ diff --git a/src/core/lib/channel/channel_stack_builder.h b/src/core/lib/channel/channel_stack_builder.h index 0e6bfd9aa6..4a00f7bfdb 100644 --- a/src/core/lib/channel/channel_stack_builder.h +++ b/src/core/lib/channel/channel_stack_builder.h @@ -39,6 +39,10 @@ #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/channel_stack.h" +#ifdef __cplusplus +extern "C" { +#endif + /// grpc_channel_stack_builder offers a programmatic interface to selected /// and order channel filters typedef struct grpc_channel_stack_builder grpc_channel_stack_builder; @@ -158,4 +162,8 @@ void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder); extern int grpc_trace_channel_stack_builder; +#ifdef __cplusplus +} +#endif + #endif /* GRPC_CORE_LIB_CHANNEL_CHANNEL_STACK_BUILDER_H */ diff --git a/src/core/lib/channel/compress_filter.c b/src/core/lib/channel/compress_filter.c index 32ebe53ee6..0981d59f63 100644 --- a/src/core/lib/channel/compress_filter.c +++ b/src/core/lib/channel/compress_filter.c @@ -60,7 +60,7 @@ typedef struct call_data { /** If true, contents of \a compression_algorithm are authoritative */ int has_compression_algorithm; - grpc_transport_stream_op send_op; + grpc_transport_stream_op *send_op; uint32_t send_length; uint32_t send_flags; gpr_slice incoming_slice; @@ -199,11 +199,11 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx, grpc_slice_buffer_stream_init(&calld->replacement_stream, &calld->slices, calld->send_flags); - calld->send_op.send_message = &calld->replacement_stream.base; - calld->post_send = calld->send_op.on_complete; - calld->send_op.on_complete = &calld->send_done; + calld->send_op->send_message = &calld->replacement_stream.base; + calld->post_send = calld->send_op->on_complete; + calld->send_op->on_complete = &calld->send_done; - grpc_call_next_op(exec_ctx, elem, &calld->send_op); + grpc_call_next_op(exec_ctx, elem, calld->send_op); } static void got_slice(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { @@ -220,7 +220,7 @@ static void got_slice(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { static void continue_send_message(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { call_data *calld = elem->call_data; - while (grpc_byte_stream_next(exec_ctx, calld->send_op.send_message, + while (grpc_byte_stream_next(exec_ctx, calld->send_op->send_message, &calld->incoming_slice, ~(size_t)0, &calld->got_slice)) { gpr_slice_buffer_add(&calld->slices, calld->incoming_slice); @@ -243,7 +243,7 @@ static void compress_start_transport_stream_op(grpc_exec_ctx *exec_ctx, } if (op->send_message != NULL && !skip_compression(elem) && 0 == (op->send_message->flags & GRPC_WRITE_NO_COMPRESS)) { - calld->send_op = *op; + calld->send_op = op; calld->send_length = op->send_message->length; calld->send_flags = op->send_message->flags; continue_send_message(exec_ctx, elem); @@ -256,8 +256,9 @@ static void compress_start_transport_stream_op(grpc_exec_ctx *exec_ctx, } /* Constructor for call_data */ -static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { +static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; @@ -266,11 +267,14 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, calld->has_compression_algorithm = 0; grpc_closure_init(&calld->got_slice, got_slice, elem); grpc_closure_init(&calld->send_done, send_done, elem); + + return GRPC_ERROR_NONE; } /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, void *ignored) { + const grpc_call_final_info *final_info, + void *ignored) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; gpr_slice_buffer_destroy(&calld->slices); diff --git a/src/core/lib/channel/connected_channel.c b/src/core/lib/channel/connected_channel.c index 0a7d27a1dc..918379c845 100644 --- a/src/core/lib/channel/connected_channel.c +++ b/src/core/lib/channel/connected_channel.c @@ -81,16 +81,16 @@ static void con_start_transport_op(grpc_exec_ctx *exec_ctx, } /* Constructor for call_data */ -static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { +static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; - int r; - - r = grpc_transport_init_stream( + int r = grpc_transport_init_stream( exec_ctx, chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld), &args->call_stack->refcount, args->server_transport_data); - GPR_ASSERT(r == 0); + return r == 0 ? GRPC_ERROR_NONE + : GRPC_ERROR_CREATE("transport stream initialization failed"); } static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, @@ -104,7 +104,7 @@ static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, + const grpc_call_final_info *final_info, void *and_free_memory) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; diff --git a/src/core/lib/channel/context.h b/src/core/lib/channel/context.h index c50e84279d..071c5f695c 100644 --- a/src/core/lib/channel/context.h +++ b/src/core/lib/channel/context.h @@ -34,10 +34,19 @@ #ifndef GRPC_CORE_LIB_CHANNEL_CONTEXT_H #define GRPC_CORE_LIB_CHANNEL_CONTEXT_H -/* Call object context pointers */ +/// Call object context pointers. + +/// Call context is represented as an array of \a grpc_call_context_elements. +/// This enum represents the indexes into the array, where each index +/// contains a different type of value. typedef enum { + /// Value is either a \a grpc_client_security_context or a + /// \a grpc_server_security_context. GRPC_CONTEXT_SECURITY = 0, + + /// Value is a \a census_context. GRPC_CONTEXT_TRACING, + GRPC_CONTEXT_COUNT } grpc_context_index; diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c new file mode 100644 index 0000000000..079b98a2f8 --- /dev/null +++ b/src/core/lib/channel/deadline_filter.c @@ -0,0 +1,302 @@ +// +// Copyright 2016, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// + +#include "src/core/lib/channel/deadline_filter.h" + +#include <stdbool.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/sync.h> +#include <grpc/support/time.h> + +#include "src/core/lib/iomgr/exec_ctx.h" +#include "src/core/lib/iomgr/timer.h" + +// +// grpc_deadline_state +// + +// Timer callback. +static void timer_callback(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { + grpc_call_element* elem = arg; + grpc_deadline_state* deadline_state = elem->call_data; + gpr_mu_lock(&deadline_state->timer_mu); + deadline_state->timer_pending = false; + gpr_mu_unlock(&deadline_state->timer_mu); + if (error != GRPC_ERROR_CANCELLED) { + gpr_slice msg = gpr_slice_from_static_string("Deadline Exceeded"); + grpc_call_element_send_cancel_with_message( + exec_ctx, elem, GRPC_STATUS_DEADLINE_EXCEEDED, &msg); + gpr_slice_unref(msg); + } + GRPC_CALL_STACK_UNREF(exec_ctx, deadline_state->call_stack, "deadline_timer"); +} + +// Starts the deadline timer. +static void start_timer_if_needed(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + gpr_timespec deadline) { + grpc_deadline_state* deadline_state = elem->call_data; + deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); + if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_MONOTONIC)) != 0) { + // Take a reference to the call stack, to be owned by the timer. + GRPC_CALL_STACK_REF(deadline_state->call_stack, "deadline_timer"); + gpr_mu_lock(&deadline_state->timer_mu); + deadline_state->timer_pending = true; + grpc_timer_init(exec_ctx, &deadline_state->timer, deadline, timer_callback, + elem, gpr_now(GPR_CLOCK_MONOTONIC)); + gpr_mu_unlock(&deadline_state->timer_mu); + } +} + +// Cancels the deadline timer. +static void cancel_timer_if_needed(grpc_exec_ctx* exec_ctx, + grpc_deadline_state* deadline_state) { + gpr_mu_lock(&deadline_state->timer_mu); + if (deadline_state->timer_pending) { + grpc_timer_cancel(exec_ctx, &deadline_state->timer); + deadline_state->timer_pending = false; + } + gpr_mu_unlock(&deadline_state->timer_mu); +} + +// Callback run when the call is complete. +static void on_complete(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { + grpc_deadline_state* deadline_state = arg; + cancel_timer_if_needed(exec_ctx, deadline_state); + // Invoke the next callback. + deadline_state->next_on_complete->cb( + exec_ctx, deadline_state->next_on_complete->cb_arg, error); +} + +// Inject our own on_complete callback into op. +static void inject_on_complete_cb(grpc_deadline_state* deadline_state, + grpc_transport_stream_op* op) { + deadline_state->next_on_complete = op->on_complete; + grpc_closure_init(&deadline_state->on_complete, on_complete, deadline_state); + op->on_complete = &deadline_state->on_complete; +} + +// Callback and associated state for starting the timer after call stack +// initialization has been completed. +struct start_timer_after_init_state { + grpc_call_element* elem; + gpr_timespec deadline; + grpc_closure closure; +}; +static void start_timer_after_init(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { + struct start_timer_after_init_state* state = arg; + start_timer_if_needed(exec_ctx, state->elem, state->deadline); + gpr_free(state); +} + +void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + grpc_call_element_args* args) { + grpc_deadline_state* deadline_state = elem->call_data; + memset(deadline_state, 0, sizeof(*deadline_state)); + deadline_state->call_stack = args->call_stack; + gpr_mu_init(&deadline_state->timer_mu); + // Deadline will always be infinite on servers, so the timer will only be + // set on clients with a finite deadline. + const gpr_timespec deadline = + gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC); + if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_MONOTONIC)) != 0) { + // When the deadline passes, we indicate the failure by sending down + // an op with cancel_error set. However, we can't send down any ops + // until after the call stack is fully initialized. If we start the + // timer here, we have no guarantee that the timer won't pop before + // call stack initialization is finished. To avoid that problem, we + // create a closure to start the timer, and we schedule that closure + // to be run after call stack initialization is done. + struct start_timer_after_init_state* state = gpr_malloc(sizeof(*state)); + state->elem = elem; + state->deadline = deadline; + grpc_closure_init(&state->closure, start_timer_after_init, state); + grpc_exec_ctx_sched(exec_ctx, &state->closure, GRPC_ERROR_NONE, NULL); + } +} + +void grpc_deadline_state_destroy(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem) { + grpc_deadline_state* deadline_state = elem->call_data; + cancel_timer_if_needed(exec_ctx, deadline_state); + gpr_mu_destroy(&deadline_state->timer_mu); +} + +void grpc_deadline_state_client_start_transport_stream_op( + grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + grpc_transport_stream_op* op) { + grpc_deadline_state* deadline_state = elem->call_data; + if (op->cancel_error != GRPC_ERROR_NONE || + op->close_error != GRPC_ERROR_NONE) { + cancel_timer_if_needed(exec_ctx, deadline_state); + } else { + // Make sure we know when the call is complete, so that we can cancel + // the timer. + if (op->recv_trailing_metadata != NULL) { + inject_on_complete_cb(deadline_state, op); + } + } +} + +// +// filter code +// + +// Constructor for channel_data. Used for both client and server filters. +static void init_channel_elem(grpc_exec_ctx* exec_ctx, + grpc_channel_element* elem, + grpc_channel_element_args* args) { + GPR_ASSERT(!args->is_last); +} + +// Destructor for channel_data. Used for both client and server filters. +static void destroy_channel_elem(grpc_exec_ctx* exec_ctx, + grpc_channel_element* elem) {} + +// Call data used for both client and server filter. +typedef struct base_call_data { + grpc_deadline_state deadline_state; +} base_call_data; + +// Additional call data used only for the server filter. +typedef struct server_call_data { + base_call_data base; // Must be first. + // The closure for receiving initial metadata. + grpc_closure recv_initial_metadata_ready; + // Received initial metadata batch. + grpc_metadata_batch* recv_initial_metadata; + // The original recv_initial_metadata_ready closure, which we chain to + // after our own closure is invoked. + grpc_closure* next_recv_initial_metadata_ready; +} server_call_data; + +// Constructor for call_data. Used for both client and server filters. +static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + grpc_call_element_args* args) { + // Note: size of call data is different between client and server. + memset(elem->call_data, 0, elem->filter->sizeof_call_data); + grpc_deadline_state_init(exec_ctx, elem, args); + return GRPC_ERROR_NONE; +} + +// Destructor for call_data. Used for both client and server filters. +static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + const grpc_call_final_info* final_info, + void* and_free_memory) { + grpc_deadline_state_destroy(exec_ctx, elem); +} + +// Method for starting a call op for client filter. +static void client_start_transport_stream_op(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + grpc_transport_stream_op* op) { + grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op); + // Chain to next filter. + grpc_call_next_op(exec_ctx, elem, op); +} + +// Callback for receiving initial metadata on the server. +static void recv_initial_metadata_ready(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { + grpc_call_element* elem = arg; + server_call_data* calld = elem->call_data; + // Get deadline from metadata and start the timer if needed. + start_timer_if_needed(exec_ctx, elem, calld->recv_initial_metadata->deadline); + // Invoke the next callback. + calld->next_recv_initial_metadata_ready->cb( + exec_ctx, calld->next_recv_initial_metadata_ready->cb_arg, error); +} + +// Method for starting a call op for server filter. +static void server_start_transport_stream_op(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + grpc_transport_stream_op* op) { + server_call_data* calld = elem->call_data; + if (op->cancel_error != GRPC_ERROR_NONE || + op->close_error != GRPC_ERROR_NONE) { + cancel_timer_if_needed(exec_ctx, &calld->base.deadline_state); + } else { + // If we're receiving initial metadata, we need to get the deadline + // from the recv_initial_metadata_ready callback. So we inject our + // own callback into that hook. + if (op->recv_initial_metadata_ready != NULL) { + calld->next_recv_initial_metadata_ready = op->recv_initial_metadata_ready; + calld->recv_initial_metadata = op->recv_initial_metadata; + grpc_closure_init(&calld->recv_initial_metadata_ready, + recv_initial_metadata_ready, elem); + op->recv_initial_metadata_ready = &calld->recv_initial_metadata_ready; + } + // Make sure we know when the call is complete, so that we can cancel + // the timer. + // Note that we trigger this on recv_trailing_metadata, even though + // the client never sends trailing metadata, because this is the + // hook that tells us when the call is complete on the server side. + if (op->recv_trailing_metadata != NULL) { + inject_on_complete_cb(&calld->base.deadline_state, op); + } + } + // Chain to next filter. + grpc_call_next_op(exec_ctx, elem, op); +} + +const grpc_channel_filter grpc_client_deadline_filter = { + client_start_transport_stream_op, + grpc_channel_next_op, + sizeof(base_call_data), + init_call_elem, + grpc_call_stack_ignore_set_pollset_or_pollset_set, + destroy_call_elem, + 0, // sizeof(channel_data) + init_channel_elem, + destroy_channel_elem, + grpc_call_next_get_peer, + "deadline", +}; + +const grpc_channel_filter grpc_server_deadline_filter = { + server_start_transport_stream_op, + grpc_channel_next_op, + sizeof(server_call_data), + init_call_elem, + grpc_call_stack_ignore_set_pollset_or_pollset_set, + destroy_call_elem, + 0, // sizeof(channel_data) + init_channel_elem, + destroy_channel_elem, + grpc_call_next_get_peer, + "deadline", +}; diff --git a/src/core/lib/channel/deadline_filter.h b/src/core/lib/channel/deadline_filter.h new file mode 100644 index 0000000000..685df87761 --- /dev/null +++ b/src/core/lib/channel/deadline_filter.h @@ -0,0 +1,79 @@ +// +// Copyright 2016, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// + +#ifndef GRPC_CORE_LIB_CHANNEL_DEADLINE_FILTER_H +#define GRPC_CORE_LIB_CHANNEL_DEADLINE_FILTER_H + +#include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/iomgr/timer.h" + +// State used for filters that enforce call deadlines. +// Must be the first field in the filter's call_data. +typedef struct grpc_deadline_state { + // We take a reference to the call stack for the timer callback. + grpc_call_stack* call_stack; + // Guards access to timer_pending and timer. + gpr_mu timer_mu; + // True if the timer callback is currently pending. + bool timer_pending; + // The deadline timer. + grpc_timer timer; + // Closure to invoke when the call is complete. + // We use this to cancel the timer. + grpc_closure on_complete; + // The original on_complete closure, which we chain to after our own + // closure is invoked. + grpc_closure* next_on_complete; +} grpc_deadline_state; + +// To be used in a filter's init_call_elem(), destroy_call_elem(), and +// start_transport_stream_op() methods to enforce call deadlines. +// +// REQUIRES: The first field in elem->call_data is a grpc_deadline_state. +// +// For grpc_deadline_state_client_start_transport_stream_op(), it is the +// caller's responsibility to chain to the next filter if necessary +// after the function returns. +void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + grpc_call_element_args* args); +void grpc_deadline_state_destroy(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem); +void grpc_deadline_state_client_start_transport_stream_op( + grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + grpc_transport_stream_op* op); + +// Deadline filters for direct client channels and server channels. +// Note: Deadlines for non-direct client channels are handled by the +// client_channel filter. +extern const grpc_channel_filter grpc_client_deadline_filter; +extern const grpc_channel_filter grpc_server_deadline_filter; + +#endif /* GRPC_CORE_LIB_CHANNEL_DEADLINE_FILTER_H */ diff --git a/src/core/lib/channel/handshaker.c b/src/core/lib/channel/handshaker.c index 6c3ca198b7..8f9fb17a31 100644 --- a/src/core/lib/channel/handshaker.c +++ b/src/core/lib/channel/handshaker.c @@ -33,8 +33,8 @@ #include <string.h> -#include <grpc/impl/codegen/alloc.h> -#include <grpc/impl/codegen/log.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/handshaker.h" @@ -62,11 +62,13 @@ void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker, grpc_endpoint* endpoint, grpc_channel_args* args, + gpr_slice_buffer* read_buffer, gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor, grpc_handshaker_done_cb cb, void* user_data) { handshaker->vtable->do_handshake(exec_ctx, handshaker, endpoint, args, - deadline, acceptor, cb, user_data); + read_buffer, deadline, acceptor, cb, + user_data); } // @@ -143,7 +145,8 @@ void grpc_handshake_manager_shutdown(grpc_exec_ctx* exec_ctx, // handshakers together. static void call_next_handshaker(grpc_exec_ctx* exec_ctx, grpc_endpoint* endpoint, - grpc_channel_args* args, void* user_data, + grpc_channel_args* args, + gpr_slice_buffer* read_buffer, void* user_data, grpc_error* error) { grpc_handshake_manager* mgr = user_data; GPR_ASSERT(mgr->state != NULL); @@ -151,8 +154,8 @@ static void call_next_handshaker(grpc_exec_ctx* exec_ctx, // If we got an error, skip all remaining handshakers and invoke the // caller-supplied callback immediately. if (error != GRPC_ERROR_NONE) { - mgr->state->final_cb(exec_ctx, endpoint, args, mgr->state->final_user_data, - error); + mgr->state->final_cb(exec_ctx, endpoint, args, read_buffer, + mgr->state->final_user_data, error); return; } grpc_handshaker_done_cb cb = call_next_handshaker; @@ -163,9 +166,9 @@ static void call_next_handshaker(grpc_exec_ctx* exec_ctx, user_data = mgr->state->final_user_data; } // Invoke handshaker. - grpc_handshaker_do_handshake(exec_ctx, mgr->handshakers[mgr->state->index], - endpoint, args, mgr->state->deadline, - mgr->state->acceptor, cb, user_data); + grpc_handshaker_do_handshake( + exec_ctx, mgr->handshakers[mgr->state->index], endpoint, args, + read_buffer, mgr->state->deadline, mgr->state->acceptor, cb, user_data); ++mgr->state->index; // If this is the last handshaker, clean up state. if (mgr->state->index == mgr->count) { @@ -180,10 +183,12 @@ void grpc_handshake_manager_do_handshake( gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor, grpc_handshaker_done_cb cb, void* user_data) { grpc_channel_args* args_copy = grpc_channel_args_copy(args); + gpr_slice_buffer* read_buffer = malloc(sizeof(*read_buffer)); + gpr_slice_buffer_init(read_buffer); if (mgr->count == 0) { // No handshakers registered, so we just immediately call the done // callback with the passed-in endpoint. - cb(exec_ctx, endpoint, args_copy, user_data, GRPC_ERROR_NONE); + cb(exec_ctx, endpoint, args_copy, read_buffer, user_data, GRPC_ERROR_NONE); } else { GPR_ASSERT(mgr->state == NULL); mgr->state = gpr_malloc(sizeof(struct grpc_handshaker_state)); @@ -192,6 +197,7 @@ void grpc_handshake_manager_do_handshake( mgr->state->acceptor = acceptor; mgr->state->final_cb = cb; mgr->state->final_user_data = user_data; - call_next_handshaker(exec_ctx, endpoint, args_copy, mgr, GRPC_ERROR_NONE); + call_next_handshaker(exec_ctx, endpoint, args_copy, read_buffer, mgr, + GRPC_ERROR_NONE); } } diff --git a/src/core/lib/channel/handshaker.h b/src/core/lib/channel/handshaker.h index dfc469c417..d574b46242 100644 --- a/src/core/lib/channel/handshaker.h +++ b/src/core/lib/channel/handshaker.h @@ -35,7 +35,6 @@ #define GRPC_CORE_LIB_CHANNEL_HANDSHAKER_H #include <grpc/impl/codegen/grpc_types.h> -#include <grpc/impl/codegen/time.h> #include "src/core/lib/iomgr/closure.h" #include "src/core/lib/iomgr/endpoint.h" @@ -56,10 +55,11 @@ typedef struct grpc_handshaker grpc_handshaker; /// Callback type invoked when a handshaker is done. -/// Takes ownership of \a args. +/// Takes ownership of \a args and \a read_buffer. typedef void (*grpc_handshaker_done_cb)(grpc_exec_ctx* exec_ctx, grpc_endpoint* endpoint, grpc_channel_args* args, + gpr_slice_buffer* read_buffer, void* user_data, grpc_error* error); struct grpc_handshaker_vtable { @@ -72,10 +72,12 @@ struct grpc_handshaker_vtable { /// Performs handshaking. When finished, calls \a cb with \a user_data. /// Takes ownership of \a args. + /// Takes ownership of \a read_buffer, which contains leftover bytes read + /// from the endpoint by the previous handshaker. /// \a acceptor will be NULL for client-side handshakers. void (*do_handshake)(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker, grpc_endpoint* endpoint, grpc_channel_args* args, - gpr_timespec deadline, + gpr_slice_buffer* read_buffer, gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor, grpc_handshaker_done_cb cb, void* user_data); }; @@ -101,6 +103,7 @@ void grpc_handshaker_do_handshake(grpc_exec_ctx* exec_ctx, grpc_handshaker* handshaker, grpc_endpoint* endpoint, grpc_channel_args* args, + gpr_slice_buffer* read_buffer, gpr_timespec deadline, grpc_tcp_server_acceptor* acceptor, grpc_handshaker_done_cb cb, void* user_data); diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c index 8057e251f0..1dc05fb20d 100644 --- a/src/core/lib/channel/http_client_filter.c +++ b/src/core/lib/channel/http_client_filter.c @@ -43,6 +43,9 @@ #define EXPECTED_CONTENT_TYPE "application/grpc" #define EXPECTED_CONTENT_TYPE_LENGTH sizeof(EXPECTED_CONTENT_TYPE) - 1 +/* default maximum size of payload eligable for GET request */ +static const size_t kMaxPayloadSizeForGet = 2048; + typedef struct call_data { grpc_linked_mdelem method; grpc_linked_mdelem scheme; @@ -50,20 +53,39 @@ typedef struct call_data { grpc_linked_mdelem te_trailers; grpc_linked_mdelem content_type; grpc_linked_mdelem user_agent; + grpc_linked_mdelem payload_bin; grpc_metadata_batch *recv_initial_metadata; + uint8_t *payload_bytes; + + /* Vars to read data off of send_message */ + grpc_transport_stream_op send_op; + uint32_t send_length; + uint32_t send_flags; + gpr_slice incoming_slice; + grpc_slice_buffer_stream replacement_stream; + gpr_slice_buffer slices; + /* flag that indicates that all slices of send_messages aren't availble */ + bool send_message_blocked; /** Closure to call when finished with the hc_on_recv hook */ grpc_closure *on_done_recv; + grpc_closure *on_complete; + grpc_closure *post_send; + /** Receive closures are chained: we inject this closure as the on_done_recv up-call on transport_op, and remember to call our on_done_recv member after handling it. */ grpc_closure hc_on_recv; + grpc_closure hc_on_complete; + grpc_closure got_slice; + grpc_closure send_done; } call_data; typedef struct channel_data { grpc_mdelem *static_scheme; grpc_mdelem *user_agent; + size_t max_payload_size_for_get; } channel_data; typedef struct { @@ -81,8 +103,8 @@ static grpc_mdelem *client_recv_filter(void *user_data, grpc_mdelem *md) { grpc_mdstr_as_c_string(md->value)); gpr_slice message = gpr_slice_from_copied_string(message_string); gpr_free(message_string); - grpc_call_element_send_cancel_with_message(a->exec_ctx, a->elem, - GRPC_STATUS_CANCELLED, &message); + grpc_call_element_send_close_with_message(a->exec_ctx, a->elem, + GRPC_STATUS_CANCELLED, &message); return NULL; } else if (md == GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC) { return NULL; @@ -119,6 +141,24 @@ static void hc_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, calld->on_done_recv->cb(exec_ctx, calld->on_done_recv->cb_arg, error); } +static void hc_on_complete(grpc_exec_ctx *exec_ctx, void *user_data, + grpc_error *error) { + grpc_call_element *elem = user_data; + call_data *calld = elem->call_data; + if (calld->payload_bytes) { + gpr_free(calld->payload_bytes); + calld->payload_bytes = NULL; + } + calld->on_complete->cb(exec_ctx, calld->on_complete->cb_arg, error); +} + +static void send_done(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { + grpc_call_element *elem = elemp; + call_data *calld = elem->call_data; + gpr_slice_buffer_reset_and_unref(&calld->slices); + calld->post_send->cb(exec_ctx, calld->post_send->cb_arg, error); +} + static grpc_mdelem *client_strip_filter(void *user_data, grpc_mdelem *md) { /* eat the things we'd like to set ourselves */ if (md->key == GRPC_MDSTR_METHOD) return NULL; @@ -129,22 +169,104 @@ static grpc_mdelem *client_strip_filter(void *user_data, grpc_mdelem *md) { return md; } -static void hc_mutate_op(grpc_call_element *elem, +static void continue_send_message(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem) { + call_data *calld = elem->call_data; + uint8_t *wrptr = calld->payload_bytes; + while (grpc_byte_stream_next(exec_ctx, calld->send_op.send_message, + &calld->incoming_slice, ~(size_t)0, + &calld->got_slice)) { + memcpy(wrptr, GPR_SLICE_START_PTR(calld->incoming_slice), + GPR_SLICE_LENGTH(calld->incoming_slice)); + wrptr += GPR_SLICE_LENGTH(calld->incoming_slice); + gpr_slice_buffer_add(&calld->slices, calld->incoming_slice); + if (calld->send_length == calld->slices.length) { + calld->send_message_blocked = false; + break; + } + } +} + +static void got_slice(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) { + grpc_call_element *elem = elemp; + call_data *calld = elem->call_data; + calld->send_message_blocked = false; + gpr_slice_buffer_add(&calld->slices, calld->incoming_slice); + if (calld->send_length == calld->slices.length) { + /* Pass down the original send_message op that was blocked.*/ + grpc_slice_buffer_stream_init(&calld->replacement_stream, &calld->slices, + calld->send_flags); + calld->send_op.send_message = &calld->replacement_stream.base; + calld->post_send = calld->send_op.on_complete; + calld->send_op.on_complete = &calld->send_done; + grpc_call_next_op(exec_ctx, elem, &calld->send_op); + } else { + continue_send_message(exec_ctx, elem); + } +} + +static void hc_mutate_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op *op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; + if (op->send_initial_metadata != NULL) { + /* Decide which HTTP VERB to use. We use GET if the request is marked + cacheable, and the operation contains both initial metadata and send + message, and the payload is below the size threshold, and all the data + for this request is immediately available. */ + grpc_mdelem *method = GRPC_MDELEM_METHOD_POST; + calld->send_message_blocked = false; + if ((op->send_initial_metadata_flags & + GRPC_INITIAL_METADATA_CACHEABLE_REQUEST) && + op->send_message != NULL && + op->send_message->length < channeld->max_payload_size_for_get) { + method = GRPC_MDELEM_METHOD_GET; + calld->send_message_blocked = true; + } else if (op->send_initial_metadata_flags & + GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST) { + method = GRPC_MDELEM_METHOD_PUT; + } + + /* Attempt to read the data from send_message and create a header field. */ + if (method == GRPC_MDELEM_METHOD_GET) { + /* allocate memory to hold the entire payload */ + calld->payload_bytes = gpr_malloc(op->send_message->length); + + /* read slices of send_message and copy into payload_bytes */ + calld->send_op = *op; + calld->send_length = op->send_message->length; + calld->send_flags = op->send_message->flags; + continue_send_message(exec_ctx, elem); + + if (calld->send_message_blocked == false) { + /* when all the send_message data is available, then create a MDELEM and + append to headers */ + grpc_mdelem *payload_bin = grpc_mdelem_from_metadata_strings( + GRPC_MDSTR_GRPC_PAYLOAD_BIN, + grpc_mdstr_from_buffer(calld->payload_bytes, + op->send_message->length)); + grpc_metadata_batch_add_tail(op->send_initial_metadata, + &calld->payload_bin, payload_bin); + calld->on_complete = op->on_complete; + op->on_complete = &calld->hc_on_complete; + op->send_message = NULL; + } else { + /* Not all data is available. Fall back to POST. */ + gpr_log(GPR_DEBUG, + "Request is marked Cacheable but not all data is available.\ + Falling back to POST"); + method = GRPC_MDELEM_METHOD_POST; + } + } + grpc_metadata_batch_filter(op->send_initial_metadata, client_strip_filter, elem); /* Send : prefixed headers, which have to be before any application layer headers. */ - grpc_metadata_batch_add_head( - op->send_initial_metadata, &calld->method, - op->send_initial_metadata_flags & - GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST - ? GRPC_MDELEM_METHOD_PUT - : GRPC_MDELEM_METHOD_POST); + grpc_metadata_batch_add_head(op->send_initial_metadata, &calld->method, + method); grpc_metadata_batch_add_head(op->send_initial_metadata, &calld->scheme, channeld->static_scheme); grpc_metadata_batch_add_tail(op->send_initial_metadata, &calld->te_trailers, @@ -169,22 +291,41 @@ static void hc_start_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op) { GPR_TIMER_BEGIN("hc_start_transport_op", 0); GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - hc_mutate_op(elem, op); + hc_mutate_op(exec_ctx, elem, op); GPR_TIMER_END("hc_start_transport_op", 0); - grpc_call_next_op(exec_ctx, elem, op); + call_data *calld = elem->call_data; + if (op->send_message != NULL && calld->send_message_blocked) { + /* Don't forward the op. send_message contains slices that aren't ready + yet. The call will be forwarded by the op_complete of slice read call. + */ + } else { + grpc_call_next_op(exec_ctx, elem, op); + } } /* Constructor for call_data */ -static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { +static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { call_data *calld = elem->call_data; calld->on_done_recv = NULL; + calld->on_complete = NULL; + calld->payload_bytes = NULL; + gpr_slice_buffer_init(&calld->slices); grpc_closure_init(&calld->hc_on_recv, hc_on_recv, elem); + grpc_closure_init(&calld->hc_on_complete, hc_on_complete, elem); + grpc_closure_init(&calld->got_slice, got_slice, elem); + grpc_closure_init(&calld->send_done, send_done, elem); + return GRPC_ERROR_NONE; } /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, void *ignored) {} + const grpc_call_final_info *final_info, + void *ignored) { + call_data *calld = elem->call_data; + gpr_slice_buffer_destroy(&calld->slices); +} static grpc_mdelem *scheme_from_args(const grpc_channel_args *args) { unsigned i; @@ -207,6 +348,22 @@ static grpc_mdelem *scheme_from_args(const grpc_channel_args *args) { return GRPC_MDELEM_SCHEME_HTTP; } +static size_t max_payload_size_from_args(const grpc_channel_args *args) { + if (args != NULL) { + for (size_t i = 0; i < args->num_args; ++i) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_MAX_PAYLOAD_SIZE_FOR_GET)) { + if (args->args[i].type != GRPC_ARG_INTEGER) { + gpr_log(GPR_ERROR, "%s: must be an integer", + GRPC_ARG_MAX_PAYLOAD_SIZE_FOR_GET); + } else { + return (size_t)args->args[i].value.integer; + } + } + } + } + return kMaxPayloadSizeForGet; +} + static grpc_mdstr *user_agent_from_args(const grpc_channel_args *args, const char *transport_name) { gpr_strvec v; @@ -230,8 +387,9 @@ static grpc_mdstr *user_agent_from_args(const grpc_channel_args *args, } } - gpr_asprintf(&tmp, "%sgrpc-c/%s (%s; %s)", is_first ? "" : " ", - grpc_version_string(), GPR_PLATFORM_STRING, transport_name); + gpr_asprintf(&tmp, "%sgrpc-c/%s (%s; %s; %s)", is_first ? "" : " ", + grpc_version_string(), GPR_PLATFORM_STRING, transport_name, + grpc_g_stands_for()); is_first = 0; gpr_strvec_add(&v, tmp); @@ -264,6 +422,8 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx, GPR_ASSERT(!args->is_last); GPR_ASSERT(args->optional_transport != NULL); chand->static_scheme = scheme_from_args(args->channel_args); + chand->max_payload_size_for_get = + max_payload_size_from_args(args->channel_args); chand->user_agent = grpc_mdelem_from_metadata_strings( GRPC_MDSTR_USER_AGENT, user_agent_from_args(args->channel_args, diff --git a/src/core/lib/channel/http_client_filter.h b/src/core/lib/channel/http_client_filter.h index 47081175ea..9e6e106e9c 100644 --- a/src/core/lib/channel/http_client_filter.h +++ b/src/core/lib/channel/http_client_filter.h @@ -41,4 +41,7 @@ extern const grpc_channel_filter grpc_http_client_filter; /* Channel arg to override the http2 :scheme header */ #define GRPC_ARG_HTTP2_SCHEME "grpc.http2_scheme" +/* Channel arg to determine maximum size of payload eligable for GET request */ +#define GRPC_ARG_MAX_PAYLOAD_SIZE_FOR_GET "grpc.max_payload_size_for_get" + #endif /* GRPC_CORE_LIB_CHANNEL_HTTP_CLIENT_FILTER_H */ diff --git a/src/core/lib/channel/http_server_filter.c b/src/core/lib/channel/http_server_filter.c index d0beebd817..0f2bf97824 100644 --- a/src/core/lib/channel/http_server_filter.c +++ b/src/core/lib/channel/http_server_filter.c @@ -49,17 +49,32 @@ typedef struct call_data { uint8_t seen_scheme; uint8_t seen_te_trailers; uint8_t seen_authority; + uint8_t seen_payload_bin; grpc_linked_mdelem status; grpc_linked_mdelem content_type; + /* flag to ensure payload_bin is delivered only once */ + uint8_t payload_bin_delivered; + grpc_metadata_batch *recv_initial_metadata; bool *recv_idempotent_request; + bool *recv_cacheable_request; /** Closure to call when finished with the hs_on_recv hook */ grpc_closure *on_done_recv; + /** Closure to call when we retrieve read message from the payload-bin header + */ + grpc_closure *recv_message_ready; + grpc_closure *on_complete; + grpc_byte_stream **pp_recv_message; + gpr_slice_buffer read_slice_buffer; + grpc_slice_buffer_stream read_stream; + /** Receive closures are chained: we inject this closure as the on_done_recv up-call on transport_op, and remember to call our on_done_recv member after handling it. */ grpc_closure hs_on_recv; + grpc_closure hs_on_complete; + grpc_closure hs_recv_message_ready; } call_data; typedef struct channel_data { uint8_t unused; } channel_data; @@ -76,16 +91,20 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { /* Check if it is one of the headers we care about. */ if (md == GRPC_MDELEM_TE_TRAILERS || md == GRPC_MDELEM_METHOD_POST || - md == GRPC_MDELEM_METHOD_PUT || md == GRPC_MDELEM_SCHEME_HTTP || - md == GRPC_MDELEM_SCHEME_HTTPS || + md == GRPC_MDELEM_METHOD_PUT || md == GRPC_MDELEM_METHOD_GET || + md == GRPC_MDELEM_SCHEME_HTTP || md == GRPC_MDELEM_SCHEME_HTTPS || md == GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC) { /* swallow it */ if (md == GRPC_MDELEM_METHOD_POST) { calld->seen_method = 1; *calld->recv_idempotent_request = false; + *calld->recv_cacheable_request = false; } else if (md == GRPC_MDELEM_METHOD_PUT) { calld->seen_method = 1; *calld->recv_idempotent_request = true; + } else if (md == GRPC_MDELEM_METHOD_GET) { + calld->seen_method = 1; + *calld->recv_cacheable_request = true; } else if (md->key == GRPC_MDSTR_SCHEME) { calld->seen_scheme = 1; } else if (md == GRPC_MDELEM_TE_TRAILERS) { @@ -137,6 +156,16 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { GRPC_MDSTR_AUTHORITY, GRPC_MDSTR_REF(md->value)); calld->seen_authority = 1; return authority; + } else if (md->key == GRPC_MDSTR_GRPC_PAYLOAD_BIN) { + /* Retrieve the payload from the value of the 'grpc-internal-payload-bin' + header field */ + calld->seen_payload_bin = 1; + gpr_slice_buffer_init(&calld->read_slice_buffer); + gpr_slice_buffer_add(&calld->read_slice_buffer, + gpr_slice_ref(md->value->slice)); + grpc_slice_buffer_stream_init(&calld->read_stream, + &calld->read_slice_buffer, 0); + return NULL; } else { return md; } @@ -189,6 +218,36 @@ static void hs_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, GRPC_ERROR_UNREF(err); } +static void hs_on_complete(grpc_exec_ctx *exec_ctx, void *user_data, + grpc_error *err) { + grpc_call_element *elem = user_data; + call_data *calld = elem->call_data; + /* Call recv_message_ready if we got the payload via the header field */ + if (calld->seen_payload_bin && calld->recv_message_ready != NULL) { + *calld->pp_recv_message = calld->payload_bin_delivered + ? NULL + : (grpc_byte_stream *)&calld->read_stream; + calld->recv_message_ready->cb(exec_ctx, calld->recv_message_ready->cb_arg, + err); + calld->recv_message_ready = NULL; + calld->payload_bin_delivered = true; + } + calld->on_complete->cb(exec_ctx, calld->on_complete->cb_arg, err); +} + +static void hs_recv_message_ready(grpc_exec_ctx *exec_ctx, void *user_data, + grpc_error *err) { + grpc_call_element *elem = user_data; + call_data *calld = elem->call_data; + if (calld->seen_payload_bin) { + /* do nothing. This is probably a GET request, and payload will be returned + in hs_on_complete callback. */ + } else { + calld->recv_message_ready->cb(exec_ctx, calld->recv_message_ready->cb_arg, + err); + } +} + static void hs_mutate_op(grpc_call_element *elem, grpc_transport_stream_op *op) { /* grab pointers to our data from the call element */ @@ -206,11 +265,25 @@ static void hs_mutate_op(grpc_call_element *elem, if (op->recv_initial_metadata) { /* substitute our callback for the higher callback */ GPR_ASSERT(op->recv_idempotent_request != NULL); + GPR_ASSERT(op->recv_cacheable_request != NULL); calld->recv_initial_metadata = op->recv_initial_metadata; calld->recv_idempotent_request = op->recv_idempotent_request; + calld->recv_cacheable_request = op->recv_cacheable_request; calld->on_done_recv = op->recv_initial_metadata_ready; op->recv_initial_metadata_ready = &calld->hs_on_recv; } + + if (op->recv_message) { + calld->recv_message_ready = op->recv_message_ready; + calld->pp_recv_message = op->recv_message; + if (op->recv_message_ready) { + op->recv_message_ready = &calld->hs_recv_message_ready; + } + if (op->on_complete) { + calld->on_complete = op->on_complete; + op->on_complete = &calld->hs_on_complete; + } + } } static void hs_start_transport_op(grpc_exec_ctx *exec_ctx, @@ -224,18 +297,23 @@ static void hs_start_transport_op(grpc_exec_ctx *exec_ctx, } /* Constructor for call_data */ -static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { +static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; /* initialize members */ memset(calld, 0, sizeof(*calld)); grpc_closure_init(&calld->hs_on_recv, hs_on_recv, elem); + grpc_closure_init(&calld->hs_on_complete, hs_on_complete, elem); + grpc_closure_init(&calld->hs_recv_message_ready, hs_recv_message_ready, elem); + return GRPC_ERROR_NONE; } /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, void *ignored) {} + const grpc_call_final_info *final_info, + void *ignored) {} /* Constructor for channel_data */ static void init_channel_elem(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c new file mode 100644 index 0000000000..02fc68fc3a --- /dev/null +++ b/src/core/lib/channel/message_size_filter.c @@ -0,0 +1,169 @@ +// +// Copyright 2016, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// + +#include "src/core/lib/channel/message_size_filter.h" + +#include <limits.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + +#include "src/core/lib/channel/channel_args.h" + +#define DEFAULT_MAX_SEND_MESSAGE_LENGTH -1 // Unlimited. +// The protobuf library will (by default) start warning at 100 megs. +#define DEFAULT_MAX_RECV_MESSAGE_LENGTH (4 * 1024 * 1024) + +typedef struct call_data { + // Receive closures are chained: we inject this closure as the + // recv_message_ready up-call on transport_stream_op, and remember to + // call our next_recv_message_ready member after handling it. + grpc_closure recv_message_ready; + // Used by recv_message_ready. + grpc_byte_stream** recv_message; + // Original recv_message_ready callback, invoked after our own. + grpc_closure* next_recv_message_ready; +} call_data; + +typedef struct channel_data { + int max_send_size; + int max_recv_size; +} channel_data; + +// Callback invoked when we receive a message. Here we check the max +// receive message size. +static void recv_message_ready(grpc_exec_ctx* exec_ctx, void* user_data, + grpc_error* error) { + grpc_call_element* elem = user_data; + call_data* calld = elem->call_data; + channel_data* chand = elem->channel_data; + if (*calld->recv_message != NULL && chand->max_recv_size >= 0 && + (*calld->recv_message)->length > (size_t)chand->max_recv_size) { + char* message_string; + gpr_asprintf(&message_string, + "Received message larger than max (%u vs. %d)", + (*calld->recv_message)->length, chand->max_recv_size); + gpr_slice message = gpr_slice_from_copied_string(message_string); + gpr_free(message_string); + grpc_call_element_send_close_with_message( + exec_ctx, elem, GRPC_STATUS_INVALID_ARGUMENT, &message); + } + // Invoke the next callback. + grpc_exec_ctx_sched(exec_ctx, calld->next_recv_message_ready, error, NULL); +} + +// Start transport op. +static void 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; + // Check max send message size. + if (op->send_message != NULL && chand->max_send_size >= 0 && + op->send_message->length > (size_t)chand->max_send_size) { + char* message_string; + gpr_asprintf(&message_string, "Sent message larger than max (%u vs. %d)", + op->send_message->length, chand->max_send_size); + gpr_slice message = gpr_slice_from_copied_string(message_string); + gpr_free(message_string); + grpc_call_element_send_close_with_message( + exec_ctx, elem, GRPC_STATUS_INVALID_ARGUMENT, &message); + } + // Inject callback for receiving a message. + if (op->recv_message_ready != NULL) { + calld->next_recv_message_ready = op->recv_message_ready; + calld->recv_message = op->recv_message; + op->recv_message_ready = &calld->recv_message_ready; + } + // Chain to the next filter. + grpc_call_next_op(exec_ctx, elem, op); +} + +// Constructor for call_data. +static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + grpc_call_element_args* args) { + call_data* calld = elem->call_data; + calld->next_recv_message_ready = NULL; + grpc_closure_init(&calld->recv_message_ready, recv_message_ready, elem); + return GRPC_ERROR_NONE; +} + +// Destructor for call_data. +static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + const grpc_call_final_info* final_info, + void* ignored) {} + +// Constructor for channel_data. +static void init_channel_elem(grpc_exec_ctx* exec_ctx, + grpc_channel_element* elem, + grpc_channel_element_args* args) { + GPR_ASSERT(!args->is_last); + channel_data* chand = elem->channel_data; + memset(chand, 0, sizeof(*chand)); + chand->max_send_size = DEFAULT_MAX_SEND_MESSAGE_LENGTH; + chand->max_recv_size = DEFAULT_MAX_RECV_MESSAGE_LENGTH; + for (size_t i = 0; i < args->channel_args->num_args; ++i) { + if (strcmp(args->channel_args->args[i].key, + GRPC_ARG_MAX_SEND_MESSAGE_LENGTH) == 0) { + const grpc_integer_options options = {DEFAULT_MAX_SEND_MESSAGE_LENGTH, 0, + INT_MAX}; + chand->max_send_size = + grpc_channel_arg_get_integer(&args->channel_args->args[i], options); + } + if (strcmp(args->channel_args->args[i].key, + GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH) == 0) { + const grpc_integer_options options = {DEFAULT_MAX_RECV_MESSAGE_LENGTH, 0, + INT_MAX}; + chand->max_recv_size = + grpc_channel_arg_get_integer(&args->channel_args->args[i], options); + } + } +} + +// Destructor for channel_data. +static void destroy_channel_elem(grpc_exec_ctx* exec_ctx, + grpc_channel_element* elem) {} + +const grpc_channel_filter grpc_message_size_filter = { + start_transport_stream_op, + grpc_channel_next_op, + sizeof(call_data), + init_call_elem, + grpc_call_stack_ignore_set_pollset_or_pollset_set, + destroy_call_elem, + sizeof(channel_data), + init_channel_elem, + destroy_channel_elem, + grpc_call_next_get_peer, + "message_size"}; diff --git a/src/core/lib/channel/message_size_filter.h b/src/core/lib/channel/message_size_filter.h new file mode 100644 index 0000000000..a88ff7f81a --- /dev/null +++ b/src/core/lib/channel/message_size_filter.h @@ -0,0 +1,39 @@ +// +// Copyright 2016, Google Inc. +// All rights reserved. +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// * Redistributions in binary form must reproduce the above +// copyright notice, this list of conditions and the following disclaimer +// in the documentation and/or other materials provided with the +// distribution. +// * Neither the name of Google Inc. nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// + +#ifndef GRPC_CORE_LIB_CHANNEL_MESSAGE_SIZE_FILTER_H +#define GRPC_CORE_LIB_CHANNEL_MESSAGE_SIZE_FILTER_H + +#include "src/core/lib/channel/channel_stack.h" + +extern const grpc_channel_filter grpc_message_size_filter; + +#endif /* GRPC_CORE_LIB_CHANNEL_MESSAGE_SIZE_FILTER_H */ diff --git a/src/core/lib/http/format_request.c b/src/core/lib/http/format_request.c index 9240356fea..e818b70113 100644 --- a/src/core/lib/http/format_request.c +++ b/src/core/lib/http/format_request.c @@ -44,7 +44,7 @@ #include "src/core/lib/support/string.h" static void fill_common_header(const grpc_httpcli_request *request, - gpr_strvec *buf) { + gpr_strvec *buf, bool connection_close) { size_t i; gpr_strvec_add(buf, gpr_strdup(request->http.path)); gpr_strvec_add(buf, gpr_strdup(" HTTP/1.0\r\n")); @@ -52,7 +52,8 @@ static void fill_common_header(const grpc_httpcli_request *request, gpr_strvec_add(buf, gpr_strdup("Host: ")); gpr_strvec_add(buf, gpr_strdup(request->host)); gpr_strvec_add(buf, gpr_strdup("\r\n")); - gpr_strvec_add(buf, gpr_strdup("Connection: close\r\n")); + if (connection_close) + gpr_strvec_add(buf, gpr_strdup("Connection: close\r\n")); gpr_strvec_add(buf, gpr_strdup("User-Agent: " GRPC_HTTPCLI_USER_AGENT "\r\n")); /* user supplied headers */ @@ -71,7 +72,7 @@ gpr_slice grpc_httpcli_format_get_request(const grpc_httpcli_request *request) { gpr_strvec_init(&out); gpr_strvec_add(&out, gpr_strdup("GET ")); - fill_common_header(request, &out); + fill_common_header(request, &out, true); gpr_strvec_add(&out, gpr_strdup("\r\n")); flat = gpr_strvec_flatten(&out, &flat_len); @@ -91,7 +92,7 @@ gpr_slice grpc_httpcli_format_post_request(const grpc_httpcli_request *request, gpr_strvec_init(&out); gpr_strvec_add(&out, gpr_strdup("POST ")); - fill_common_header(request, &out); + fill_common_header(request, &out, true); if (body_bytes) { uint8_t has_content_type = 0; for (i = 0; i < request->http.hdr_count; i++) { @@ -118,3 +119,16 @@ gpr_slice grpc_httpcli_format_post_request(const grpc_httpcli_request *request, return gpr_slice_new(tmp, out_len, gpr_free); } + +gpr_slice grpc_httpcli_format_connect_request( + const grpc_httpcli_request *request) { + gpr_strvec out; + gpr_strvec_init(&out); + gpr_strvec_add(&out, gpr_strdup("CONNECT ")); + fill_common_header(request, &out, false); + gpr_strvec_add(&out, gpr_strdup("\r\n")); + size_t flat_len; + char *flat = gpr_strvec_flatten(&out, &flat_len); + gpr_strvec_destroy(&out); + return gpr_slice_new(flat, flat_len, gpr_free); +} diff --git a/src/core/lib/http/format_request.h b/src/core/lib/http/format_request.h index 1543efe4b0..7abd55f2f7 100644 --- a/src/core/lib/http/format_request.h +++ b/src/core/lib/http/format_request.h @@ -41,5 +41,7 @@ gpr_slice grpc_httpcli_format_get_request(const grpc_httpcli_request *request); gpr_slice grpc_httpcli_format_post_request(const grpc_httpcli_request *request, const char *body_bytes, size_t body_size); +gpr_slice grpc_httpcli_format_connect_request( + const grpc_httpcli_request *request); #endif /* GRPC_CORE_LIB_HTTP_FORMAT_REQUEST_H */ diff --git a/src/core/lib/http/httpcli.c b/src/core/lib/http/httpcli.c index 18135bcb58..7f3c2d120d 100644 --- a/src/core/lib/http/httpcli.c +++ b/src/core/lib/http/httpcli.c @@ -146,7 +146,7 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, if (GPR_SLICE_LENGTH(req->incoming.slices[i])) { req->have_read_byte = 1; grpc_error *err = - grpc_http_parser_parse(&req->parser, req->incoming.slices[i]); + grpc_http_parser_parse(&req->parser, req->incoming.slices[i], NULL); if (err != GRPC_ERROR_NONE) { finish(exec_ctx, req, err); return; diff --git a/src/core/lib/http/httpcli.h b/src/core/lib/http/httpcli.h index 662e176f4c..320c0f86c6 100644 --- a/src/core/lib/http/httpcli.h +++ b/src/core/lib/http/httpcli.h @@ -93,8 +93,7 @@ void grpc_httpcli_context_destroy(grpc_httpcli_context *context); 'request' contains request parameters - these are caller owned and can be destroyed once the call returns 'deadline' contains a deadline for the request (or gpr_inf_future) - 'on_response' is a callback to report results to (and 'user_data' is a user - supplied pointer to pass to said call) */ + 'on_response' is a callback to report results to */ void grpc_httpcli_get(grpc_exec_ctx *exec_ctx, grpc_httpcli_context *context, grpc_polling_entity *pollent, const grpc_httpcli_request *request, @@ -113,8 +112,7 @@ void grpc_httpcli_get(grpc_exec_ctx *exec_ctx, grpc_httpcli_context *context, 'deadline' contains a deadline for the request (or gpr_inf_future) 'em' points to a caller owned event manager that must be alive for the lifetime of the request - 'on_response' is a callback to report results to (and 'user_data' is a user - supplied pointer to pass to said call) + 'on_response' is a callback to report results to Does not support ?var1=val1&var2=val2 in the path. */ void grpc_httpcli_post(grpc_exec_ctx *exec_ctx, grpc_httpcli_context *context, grpc_polling_entity *pollent, diff --git a/src/core/lib/http/httpcli_security_connector.c b/src/core/lib/http/httpcli_security_connector.c index a57d93bb7b..0006e809a6 100644 --- a/src/core/lib/http/httpcli_security_connector.c +++ b/src/core/lib/http/httpcli_security_connector.c @@ -61,6 +61,7 @@ static void httpcli_ssl_destroy(grpc_security_connector *sc) { static void httpcli_ssl_do_handshake(grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc, grpc_endpoint *nonsecure_endpoint, + gpr_slice_buffer *read_buffer, gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data) { @@ -69,6 +70,7 @@ static void httpcli_ssl_do_handshake(grpc_exec_ctx *exec_ctx, tsi_result result = TSI_OK; tsi_handshaker *handshaker; if (c->handshaker_factory == NULL) { + gpr_free(read_buffer); cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL, NULL); return; } @@ -77,10 +79,12 @@ static void httpcli_ssl_do_handshake(grpc_exec_ctx *exec_ctx, if (result != TSI_OK) { gpr_log(GPR_ERROR, "Handshaker creation failed with error %s.", tsi_result_to_string(result)); + gpr_free(read_buffer); cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL, NULL); } else { grpc_do_security_handshake(exec_ctx, handshaker, &sc->base, true, - nonsecure_endpoint, deadline, cb, user_data); + nonsecure_endpoint, read_buffer, deadline, cb, + user_data); } } @@ -183,7 +187,7 @@ static void ssl_handshake(grpc_exec_ctx *exec_ctx, void *arg, pem_root_certs, pem_root_certs_size, host, &sc) == GRPC_SECURITY_OK); grpc_channel_security_connector_do_handshake( - exec_ctx, sc, tcp, deadline, on_secure_transport_setup_done, c); + exec_ctx, sc, tcp, NULL, deadline, on_secure_transport_setup_done, c); GRPC_SECURITY_CONNECTOR_UNREF(&sc->base, "httpcli"); } diff --git a/src/core/lib/http/parser.c b/src/core/lib/http/parser.c index 92ed08ae51..be9e9b6b63 100644 --- a/src/core/lib/http/parser.c +++ b/src/core/lib/http/parser.c @@ -33,6 +33,7 @@ #include "src/core/lib/http/parser.h" +#include <stdbool.h> #include <string.h> #include <grpc/support/alloc.h> @@ -200,7 +201,8 @@ done: return error; } -static grpc_error *finish_line(grpc_http_parser *parser) { +static grpc_error *finish_line(grpc_http_parser *parser, + bool *found_body_start) { grpc_error *err; switch (parser->state) { case GRPC_HTTP_FIRST_LINE: @@ -211,6 +213,7 @@ static grpc_error *finish_line(grpc_http_parser *parser) { case GRPC_HTTP_HEADERS: if (parser->cur_line_length == parser->cur_line_end_length) { parser->state = GRPC_HTTP_BODY; + *found_body_start = true; break; } err = add_header(parser); @@ -274,7 +277,8 @@ static bool check_line(grpc_http_parser *parser) { return false; } -static grpc_error *addbyte(grpc_http_parser *parser, uint8_t byte) { +static grpc_error *addbyte(grpc_http_parser *parser, uint8_t byte, + bool *found_body_start) { switch (parser->state) { case GRPC_HTTP_FIRST_LINE: case GRPC_HTTP_HEADERS: @@ -282,20 +286,18 @@ static grpc_error *addbyte(grpc_http_parser *parser, uint8_t byte) { if (grpc_http1_trace) gpr_log(GPR_ERROR, "HTTP client max line length (%d) exceeded", GRPC_HTTP_PARSER_MAX_HEADER_LENGTH); - return 0; + return GRPC_ERROR_NONE; } parser->cur_line[parser->cur_line_length] = byte; parser->cur_line_length++; if (check_line(parser)) { - return finish_line(parser); - } else { - return GRPC_ERROR_NONE; + return finish_line(parser, found_body_start); } - GPR_UNREACHABLE_CODE(return 0); + return GRPC_ERROR_NONE; case GRPC_HTTP_BODY: return addbyte_body(parser, byte); } - GPR_UNREACHABLE_CODE(return 0); + GPR_UNREACHABLE_CODE(return GRPC_ERROR_NONE); } void grpc_http_parser_init(grpc_http_parser *parser, grpc_http_type type, @@ -331,14 +333,15 @@ void grpc_http_response_destroy(grpc_http_response *response) { gpr_free(response->hdrs); } -grpc_error *grpc_http_parser_parse(grpc_http_parser *parser, gpr_slice slice) { - size_t i; - - for (i = 0; i < GPR_SLICE_LENGTH(slice); i++) { - grpc_error *err = addbyte(parser, GPR_SLICE_START_PTR(slice)[i]); +grpc_error *grpc_http_parser_parse(grpc_http_parser *parser, gpr_slice slice, + size_t *start_of_body) { + for (size_t i = 0; i < GPR_SLICE_LENGTH(slice); i++) { + bool found_body_start = false; + grpc_error *err = + addbyte(parser, GPR_SLICE_START_PTR(slice)[i], &found_body_start); if (err != GRPC_ERROR_NONE) return err; + if (found_body_start && start_of_body != NULL) *start_of_body = i + 1; } - return GRPC_ERROR_NONE; } diff --git a/src/core/lib/http/parser.h b/src/core/lib/http/parser.h index 6df3cc8b13..fab42979cd 100644 --- a/src/core/lib/http/parser.h +++ b/src/core/lib/http/parser.h @@ -113,7 +113,9 @@ void grpc_http_parser_init(grpc_http_parser *parser, grpc_http_type type, void *request_or_response); void grpc_http_parser_destroy(grpc_http_parser *parser); -grpc_error *grpc_http_parser_parse(grpc_http_parser *parser, gpr_slice slice); +/* Sets \a start_of_body to the offset in \a slice of the start of the body. */ +grpc_error *grpc_http_parser_parse(grpc_http_parser *parser, gpr_slice slice, + size_t *start_of_body); grpc_error *grpc_http_parser_eof(grpc_http_parser *parser); void grpc_http_request_destroy(grpc_http_request *request); diff --git a/src/core/lib/iomgr/closure.c b/src/core/lib/iomgr/closure.c index 0b6c3b2539..1ba0a5c141 100644 --- a/src/core/lib/iomgr/closure.c +++ b/src/core/lib/iomgr/closure.c @@ -41,6 +41,10 @@ void grpc_closure_init(grpc_closure *closure, grpc_iomgr_cb_func cb, closure->cb_arg = cb_arg; } +void grpc_closure_list_init(grpc_closure_list *closure_list) { + closure_list->head = closure_list->tail = NULL; +} + void grpc_closure_list_append(grpc_closure_list *closure_list, grpc_closure *closure, grpc_error *error) { if (closure == NULL) { diff --git a/src/core/lib/iomgr/closure.h b/src/core/lib/iomgr/closure.h index 08e59a168e..c1a22b6021 100644 --- a/src/core/lib/iomgr/closure.h +++ b/src/core/lib/iomgr/closure.h @@ -37,6 +37,7 @@ #include <grpc/support/port_platform.h> #include <stdbool.h> #include "src/core/lib/iomgr/error.h" +#include "src/core/lib/support/mpscq.h" struct grpc_closure; typedef struct grpc_closure grpc_closure; @@ -60,6 +61,14 @@ typedef void (*grpc_iomgr_cb_func)(grpc_exec_ctx *exec_ctx, void *arg, /** A closure over a grpc_iomgr_cb_func. */ struct grpc_closure { + /** Once queued, next indicates the next queued closure; before then, scratch + * space */ + union { + grpc_closure *next; + gpr_mpscq_node atm_next; + uintptr_t scratch; + } next_data; + /** Bound callback. */ grpc_iomgr_cb_func cb; @@ -68,13 +77,6 @@ struct grpc_closure { /** Once queued, the result of the closure. Before then: scratch space */ grpc_error *error; - - /** Once queued, next indicates the next queued closure; before then, scratch - * space */ - union { - grpc_closure *next; - uintptr_t scratch; - } next_data; }; /** Initializes \a closure with \a cb and \a cb_arg. */ @@ -87,6 +89,8 @@ grpc_closure *grpc_closure_create(grpc_iomgr_cb_func cb, void *cb_arg); #define GRPC_CLOSURE_LIST_INIT \ { NULL, NULL } +void grpc_closure_list_init(grpc_closure_list *list); + /** add \a closure to the end of \a list and set \a closure's result to \a error */ void grpc_closure_list_append(grpc_closure_list *list, grpc_closure *closure, diff --git a/src/core/lib/iomgr/combiner.c b/src/core/lib/iomgr/combiner.c new file mode 100644 index 0000000000..831bdb4aff --- /dev/null +++ b/src/core/lib/iomgr/combiner.c @@ -0,0 +1,293 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/iomgr/combiner.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> + +#include "src/core/lib/iomgr/workqueue.h" +#include "src/core/lib/profiling/timers.h" + +int grpc_combiner_trace = 0; + +#define GRPC_COMBINER_TRACE(fn) \ + do { \ + if (grpc_combiner_trace) { \ + fn; \ + } \ + } while (0) + +struct grpc_combiner { + grpc_workqueue *optional_workqueue; + gpr_mpscq queue; + // state is: + // lower bit - zero if orphaned + // other bits - number of items queued on the lock + gpr_atm state; + bool take_async_break_before_final_list; + grpc_closure_list final_list; + grpc_closure continue_finishing; +}; + +grpc_combiner *grpc_combiner_create(grpc_workqueue *optional_workqueue) { + grpc_combiner *lock = gpr_malloc(sizeof(*lock)); + lock->optional_workqueue = optional_workqueue; + gpr_atm_no_barrier_store(&lock->state, 1); + gpr_mpscq_init(&lock->queue); + lock->take_async_break_before_final_list = false; + grpc_closure_list_init(&lock->final_list); + GRPC_COMBINER_TRACE(gpr_log(GPR_DEBUG, "C:%p create", lock)); + return lock; +} + +static void really_destroy(grpc_exec_ctx *exec_ctx, grpc_combiner *lock) { + GRPC_COMBINER_TRACE(gpr_log(GPR_DEBUG, "C:%p really_destroy", lock)); + GPR_ASSERT(gpr_atm_no_barrier_load(&lock->state) == 0); + gpr_mpscq_destroy(&lock->queue); + GRPC_WORKQUEUE_UNREF(exec_ctx, lock->optional_workqueue, "combiner"); + gpr_free(lock); +} + +void grpc_combiner_destroy(grpc_exec_ctx *exec_ctx, grpc_combiner *lock) { + gpr_atm old_state = gpr_atm_full_fetch_add(&lock->state, -1); + GRPC_COMBINER_TRACE(gpr_log( + GPR_DEBUG, "C:%p really_destroy old_state=%" PRIdPTR, lock, old_state)); + if (old_state == 1) { + really_destroy(exec_ctx, lock); + } +} + +static bool maybe_finish_one(grpc_exec_ctx *exec_ctx, grpc_combiner *lock); +static void finish(grpc_exec_ctx *exec_ctx, grpc_combiner *lock); + +static void continue_finishing_mainline(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + GPR_TIMER_BEGIN("combiner.continue_executing_mainline", 0); + grpc_combiner *lock = arg; + GRPC_COMBINER_TRACE( + gpr_log(GPR_DEBUG, "C:%p continue_finishing_mainline", lock)); + GPR_ASSERT(exec_ctx->active_combiner == NULL); + exec_ctx->active_combiner = lock; + if (maybe_finish_one(exec_ctx, lock)) finish(exec_ctx, lock); + GPR_ASSERT(exec_ctx->active_combiner == lock); + exec_ctx->active_combiner = NULL; + GPR_TIMER_END("combiner.continue_executing_mainline", 0); +} + +static void execute_final(grpc_exec_ctx *exec_ctx, grpc_combiner *lock) { + GPR_TIMER_BEGIN("combiner.execute_final", 0); + grpc_closure *c = lock->final_list.head; + GPR_ASSERT(c != NULL); + grpc_closure_list_init(&lock->final_list); + lock->take_async_break_before_final_list = false; + int loops = 0; + while (c != NULL) { + GRPC_COMBINER_TRACE( + gpr_log(GPR_DEBUG, "C:%p execute_final[%d] c=%p", lock, loops, c)); + grpc_closure *next = c->next_data.next; + grpc_error *error = c->error; + c->cb(exec_ctx, c->cb_arg, error); + GRPC_ERROR_UNREF(error); + c = next; + loops++; + } + GPR_TIMER_END("combiner.execute_final", 0); +} + +static void continue_executing_final(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + GPR_TIMER_BEGIN("combiner.continue_executing_final", 0); + grpc_combiner *lock = arg; + GRPC_COMBINER_TRACE( + gpr_log(GPR_DEBUG, "C:%p continue_executing_final", lock)); + GPR_ASSERT(exec_ctx->active_combiner == NULL); + exec_ctx->active_combiner = lock; + // quick peek to see if new things have turned up on the queue: if so, go back + // to executing them before the final list + if ((gpr_atm_acq_load(&lock->state) >> 1) > 1) { + if (maybe_finish_one(exec_ctx, lock)) finish(exec_ctx, lock); + } else { + execute_final(exec_ctx, lock); + finish(exec_ctx, lock); + } + GPR_ASSERT(exec_ctx->active_combiner == lock); + exec_ctx->active_combiner = NULL; + GPR_TIMER_END("combiner.continue_executing_final", 0); +} + +static bool start_execute_final(grpc_exec_ctx *exec_ctx, grpc_combiner *lock) { + GPR_TIMER_BEGIN("combiner.start_execute_final", 0); + GPR_ASSERT(exec_ctx->active_combiner == lock); + GRPC_COMBINER_TRACE( + gpr_log(GPR_DEBUG, + "C:%p start_execute_final take_async_break_before_final_list=%d", + lock, lock->take_async_break_before_final_list)); + if (lock->take_async_break_before_final_list) { + grpc_closure_init(&lock->continue_finishing, continue_executing_final, + lock); + grpc_exec_ctx_sched(exec_ctx, &lock->continue_finishing, GRPC_ERROR_NONE, + GRPC_WORKQUEUE_REF(lock->optional_workqueue, "sched")); + GPR_TIMER_END("combiner.start_execute_final", 0); + return false; + } else { + execute_final(exec_ctx, lock); + GPR_TIMER_END("combiner.start_execute_final", 0); + return true; + } +} + +static bool maybe_finish_one(grpc_exec_ctx *exec_ctx, grpc_combiner *lock) { + GPR_TIMER_BEGIN("combiner.maybe_finish_one", 0); + gpr_mpscq_node *n = gpr_mpscq_pop(&lock->queue); + GRPC_COMBINER_TRACE( + gpr_log(GPR_DEBUG, "C:%p maybe_finish_one n=%p", lock, n)); + GPR_ASSERT(exec_ctx->active_combiner == lock); + if (n == NULL) { + // Queue is in an transiently inconsistent state: a new item is being queued + // but is not visible to this thread yet. + // Use this as a cue that we should go off and do something else for a while + // (and come back later) + grpc_closure_init(&lock->continue_finishing, continue_finishing_mainline, + lock); + grpc_exec_ctx_sched(exec_ctx, &lock->continue_finishing, GRPC_ERROR_NONE, + GRPC_WORKQUEUE_REF(lock->optional_workqueue, "sched")); + GPR_TIMER_END("combiner.maybe_finish_one", 0); + return false; + } + grpc_closure *cl = (grpc_closure *)n; + grpc_error *error = cl->error; + cl->cb(exec_ctx, cl->cb_arg, error); + GRPC_ERROR_UNREF(error); + GPR_TIMER_END("combiner.maybe_finish_one", 0); + return true; +} + +static void finish(grpc_exec_ctx *exec_ctx, grpc_combiner *lock) { + bool (*executor)(grpc_exec_ctx * exec_ctx, grpc_combiner * lock); + GPR_TIMER_BEGIN("combiner.finish", 0); + int loops = 0; + do { + executor = maybe_finish_one; + gpr_atm old_state = gpr_atm_full_fetch_add(&lock->state, -2); + GRPC_COMBINER_TRACE(gpr_log(GPR_DEBUG, + "C:%p finish[%d] old_state=%" PRIdPTR, lock, + loops, old_state)); + switch (old_state) { + default: + // we have multiple queued work items: just continue executing them + break; + case 5: // we're down to one queued item: if it's the final list we + case 4: // should do that + if (!grpc_closure_list_empty(lock->final_list)) { + executor = start_execute_final; + } + break; + case 3: // had one count, one unorphaned --> unlocked unorphaned + GPR_TIMER_END("combiner.finish", 0); + return; + case 2: // and one count, one orphaned --> unlocked and orphaned + really_destroy(exec_ctx, lock); + GPR_TIMER_END("combiner.finish", 0); + return; + case 1: + case 0: + // these values are illegal - representing an already unlocked or + // deleted lock + GPR_UNREACHABLE_CODE(return ); + } + loops++; + } while (executor(exec_ctx, lock)); + GPR_TIMER_END("combiner.finish", 0); +} + +void grpc_combiner_execute(grpc_exec_ctx *exec_ctx, grpc_combiner *lock, + grpc_closure *cl, grpc_error *error) { + GRPC_COMBINER_TRACE( + gpr_log(GPR_DEBUG, "C:%p grpc_combiner_execute c=%p", lock, cl)); + GPR_TIMER_BEGIN("combiner.execute", 0); + gpr_atm last = gpr_atm_full_fetch_add(&lock->state, 2); + GPR_ASSERT(last & 1); // ensure lock has not been destroyed + if (last == 1) { + exec_ctx->active_combiner = lock; + GPR_TIMER_BEGIN("combiner.execute_first_cb", 0); + cl->cb(exec_ctx, cl->cb_arg, error); + GPR_TIMER_END("combiner.execute_first_cb", 0); + GRPC_ERROR_UNREF(error); + finish(exec_ctx, lock); + GPR_ASSERT(exec_ctx->active_combiner == lock); + exec_ctx->active_combiner = NULL; + } else { + cl->error = error; + gpr_mpscq_push(&lock->queue, &cl->next_data.atm_next); + } + GPR_TIMER_END("combiner.execute", 0); +} + +static void enqueue_finally(grpc_exec_ctx *exec_ctx, void *closure, + grpc_error *error) { + grpc_combiner_execute_finally(exec_ctx, exec_ctx->active_combiner, closure, + GRPC_ERROR_REF(error), false); +} + +void grpc_combiner_execute_finally(grpc_exec_ctx *exec_ctx, grpc_combiner *lock, + grpc_closure *closure, grpc_error *error, + bool force_async_break) { + GRPC_COMBINER_TRACE(gpr_log( + GPR_DEBUG, + "C:%p grpc_combiner_execute_finally c=%p force_async_break=%d; ac=%p", + lock, closure, force_async_break, exec_ctx->active_combiner)); + GPR_TIMER_BEGIN("combiner.execute_finally", 0); + if (exec_ctx->active_combiner != lock) { + GPR_TIMER_MARK("slowpath", 0); + grpc_combiner_execute(exec_ctx, lock, + grpc_closure_create(enqueue_finally, closure), error); + GPR_TIMER_END("combiner.execute_finally", 0); + return; + } + + if (force_async_break) { + lock->take_async_break_before_final_list = true; + } + if (grpc_closure_list_empty(lock->final_list)) { + gpr_atm_full_fetch_add(&lock->state, 2); + } + grpc_closure_list_append(&lock->final_list, closure, error); + GPR_TIMER_END("combiner.execute_finally", 0); +} + +void grpc_combiner_force_async_finally(grpc_combiner *lock) { + lock->take_async_break_before_final_list = true; +} diff --git a/src/core/lib/iomgr/combiner.h b/src/core/lib/iomgr/combiner.h new file mode 100644 index 0000000000..1409db24b9 --- /dev/null +++ b/src/core/lib/iomgr/combiner.h @@ -0,0 +1,71 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_IOMGR_COMBINER_H +#define GRPC_CORE_LIB_IOMGR_COMBINER_H + +#include <stddef.h> + +#include <grpc/support/atm.h> +#include "src/core/lib/iomgr/exec_ctx.h" +#include "src/core/lib/support/mpscq.h" + +// Provides serialized access to some resource. +// Each action queued on a combiner is executed serially in a borrowed thread. +// The actual thread executing actions may change over time (but there will only +// every be one at a time). + +// Initialize the lock, with an optional workqueue to shift load to when +// necessary +grpc_combiner *grpc_combiner_create(grpc_workqueue *optional_workqueue); +// Destroy the lock +void grpc_combiner_destroy(grpc_exec_ctx *exec_ctx, grpc_combiner *lock); +// Execute \a action within the lock. +void grpc_combiner_execute(grpc_exec_ctx *exec_ctx, grpc_combiner *lock, + grpc_closure *closure, grpc_error *error); +// Execute \a action within the lock just prior to unlocking. +// if \a hint_async_break is true, the combiner tries to hand execution to +// another thread before finishing the primary queue of combined closures and +// executing the finally list. +// Deprecation warning: \a hint_async_break will be removed in a future version +// Takes a very slow and round-about path if not called from a +// grpc_combiner_execute closure. +void grpc_combiner_execute_finally(grpc_exec_ctx *exec_ctx, grpc_combiner *lock, + grpc_closure *closure, grpc_error *error, + bool hint_async_break); +// Deprecated: force the finally list execution onto another thread +void grpc_combiner_force_async_finally(grpc_combiner *lock); + +extern int grpc_combiner_trace; + +#endif /* GRPC_CORE_LIB_IOMGR_COMBINER_H */ diff --git a/src/core/lib/iomgr/endpoint.h b/src/core/lib/iomgr/endpoint.h index 894efc0b23..910a6f6532 100644 --- a/src/core/lib/iomgr/endpoint.h +++ b/src/core/lib/iomgr/endpoint.h @@ -64,7 +64,8 @@ struct grpc_endpoint_vtable { /* When data is available on the connection, calls the callback with slices. Callback success indicates that the endpoint can accept more reads, failure indicates the endpoint is closed. - Valid slices may be placed into \a slices even on callback success == 0. */ + Valid slices may be placed into \a slices even when the callback is + invoked with error != GRPC_ERROR_NONE. */ void grpc_endpoint_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, gpr_slice_buffer *slices, grpc_closure *cb); diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c index 149c55663c..31c80260f8 100644 --- a/src/core/lib/iomgr/error.c +++ b/src/core/lib/iomgr/error.c @@ -324,6 +324,64 @@ const char *grpc_error_get_str(grpc_error *err, grpc_error_strs which) { return gpr_avl_get(err->strs, (void *)(uintptr_t)which); } +typedef struct { + grpc_error *error; + grpc_status_code code; + const char *msg; +} special_error_status_map; +static special_error_status_map error_status_map[] = { + {GRPC_ERROR_NONE, GRPC_STATUS_OK, ""}, + {GRPC_ERROR_CANCELLED, GRPC_STATUS_CANCELLED, "RPC cancelled"}, + {GRPC_ERROR_OOM, GRPC_STATUS_RESOURCE_EXHAUSTED, "Out of memory"}, +}; + +static grpc_error *recursively_find_error_with_status(grpc_error *error, + intptr_t *status) { + // If the error itself has a status code, return it. + if (grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, status)) { + return error; + } + // Otherwise, search through its children. + intptr_t key = 0; + while (true) { + grpc_error *child_error = gpr_avl_get(error->errs, (void *)key++); + if (child_error == NULL) break; + grpc_error *result = + recursively_find_error_with_status(child_error, status); + if (result != NULL) return result; + } + return NULL; +} + +void grpc_error_get_status(grpc_error *error, grpc_status_code *code, + const char **msg) { + // Handle special errors via the static map. + for (size_t i = 0; i < GPR_ARRAY_SIZE(error_status_map); ++i) { + if (error == error_status_map[i].error) { + *code = error_status_map[i].code; + *msg = error_status_map[i].msg; + return; + } + } + // Populate code. + // Start with the parent error and recurse through the tree of children + // until we find the first one that has a status code. + intptr_t status = GRPC_STATUS_UNKNOWN; // Default in case we don't find one. + grpc_error *found_error = recursively_find_error_with_status(error, &status); + *code = (grpc_status_code)status; + // Now populate msg. + // If we found an error with a status code above, use that; otherwise, + // fall back to using the parent error. + if (found_error == NULL) found_error = error; + // If the error has a status message, use it. Otherwise, fall back to + // the error description. + *msg = grpc_error_get_str(found_error, GRPC_ERROR_STR_GRPC_MESSAGE); + if (*msg == NULL) { + *msg = grpc_error_get_str(found_error, GRPC_ERROR_STR_DESCRIPTION); + if (*msg == NULL) *msg = "uknown error"; // Just in case. + } +} + grpc_error *grpc_error_add_child(grpc_error *src, grpc_error *child) { GPR_TIMER_BEGIN("grpc_error_add_child", 0); grpc_error *new = copy_error_and_unref(src); @@ -332,7 +390,7 @@ grpc_error *grpc_error_add_child(grpc_error *src, grpc_error *child) { return new; } -static const char *no_error_string = "null"; +static const char *no_error_string = "\"No Error\""; static const char *oom_error_string = "\"Out of memory\""; static const char *cancelled_error_string = "\"Cancelled\""; diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h index 13f898e31a..7e2fd7a3bd 100644 --- a/src/core/lib/iomgr/error.h +++ b/src/core/lib/iomgr/error.h @@ -37,6 +37,7 @@ #include <stdbool.h> #include <stdint.h> +#include <grpc/status.h> #include <grpc/support/time.h> /// Opaque representation of an error. @@ -47,7 +48,8 @@ /// if a grpc_error is passed to a grpc_closure callback function (functions /// with the signature: /// void (*f)(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error)) -/// then those functions do not automatically own a ref to error +/// then those functions do not own a ref to error (but are free to manually +/// take a reference). /// if a grpc_error is passed to *ANY OTHER FUNCTION* then that function takes /// ownership of the error /// Errors have: @@ -171,7 +173,16 @@ grpc_error *grpc_error_set_time(grpc_error *src, grpc_error_times which, gpr_timespec value) GRPC_MUST_USE_RESULT; grpc_error *grpc_error_set_str(grpc_error *src, grpc_error_strs which, const char *value) GRPC_MUST_USE_RESULT; +/// Returns NULL if the specified string is not set. +/// Caller does NOT own return value. const char *grpc_error_get_str(grpc_error *error, grpc_error_strs which); + +/// A utility function to get the status code and message to be returned +/// to the application. If not set in the top-level message, looks +/// through child errors until it finds the first one with these attributes. +void grpc_error_get_status(grpc_error *error, grpc_status_code *code, + const char **msg); + /// Add a child error: an error that is believed to have contributed to this /// error occurring. Allows root causing high level errors from lower level /// errors that contributed to them. diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c index 6a63c4d1d1..ab77ebc78b 100644 --- a/src/core/lib/iomgr/ev_epoll_linux.c +++ b/src/core/lib/iomgr/ev_epoll_linux.c @@ -42,6 +42,7 @@ #include <assert.h> #include <errno.h> #include <poll.h> +#include <pthread.h> #include <signal.h> #include <string.h> #include <sys/epoll.h> @@ -927,7 +928,8 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, fd->polling_island = NULL; } - grpc_exec_ctx_sched(exec_ctx, fd->on_done_closure, error, NULL); + grpc_exec_ctx_sched(exec_ctx, fd->on_done_closure, GRPC_ERROR_REF(error), + NULL); gpr_mu_unlock(&fd->mu); UNREF_BY(fd, 2, reason); /* Drop the reference */ @@ -939,6 +941,7 @@ static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, PI_UNREF(exec_ctx, unref_pi, "fd_orphan"); } GRPC_LOG_IF_ERROR("fd_orphan", GRPC_ERROR_REF(error)); + GRPC_ERROR_UNREF(error); } static grpc_error *fd_shutdown_error(bool shutdown) { @@ -1352,8 +1355,10 @@ static void pollset_work_and_unlock(grpc_exec_ctx *exec_ctx, gpr_mu_unlock(&pollset->mu); do { + GRPC_SCHEDULING_START_BLOCKING_REGION; ep_rv = epoll_pwait(epoll_fd, ep_ev, GRPC_EPOLL_MAX_EVENTS, timeout_ms, sig_mask); + GRPC_SCHEDULING_END_BLOCKING_REGION; if (ep_rv < 0) { if (errno != EINTR) { gpr_asprintf(&err_msg, @@ -1526,6 +1531,8 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_fd *fd) { + GPR_TIMER_BEGIN("pollset_add_fd", 0); + grpc_error *error = GRPC_ERROR_NONE; gpr_mu_lock(&pollset->mu); @@ -1638,6 +1645,8 @@ retry: gpr_mu_unlock(&pollset->mu); GRPC_LOG_IF_ERROR("pollset_add_fd", error); + + GPR_TIMER_END("pollset_add_fd", 0); } /******************************************************************************* @@ -1888,7 +1897,7 @@ const grpc_event_engine_vtable *grpc_init_epoll_linux(void) { } if (!is_grpc_wakeup_signal_initialized) { - grpc_use_signal(SIGRTMIN + 2); + grpc_use_signal(SIGRTMIN + 6); } fd_global_init(); diff --git a/src/core/lib/iomgr/exec_ctx.h b/src/core/lib/iomgr/exec_ctx.h index 917f332f03..4d20ecf922 100644 --- a/src/core/lib/iomgr/exec_ctx.h +++ b/src/core/lib/iomgr/exec_ctx.h @@ -40,10 +40,9 @@ /** A workqueue represents a list of work to be executed asynchronously. Forward declared here to avoid a circular dependency with workqueue.h. */ -struct grpc_workqueue; typedef struct grpc_workqueue grpc_workqueue; +typedef struct grpc_combiner grpc_combiner; -#ifndef GRPC_EXECUTION_CONTEXT_SANITIZER /** Execution context. * A bag of data that collects information along a callstack. * Generally created at public API entry points, and passed down as @@ -58,21 +57,24 @@ typedef struct grpc_workqueue grpc_workqueue; * should actively try to finish up and get this thread back to its owner * * CONVENTIONS: - * Instance of this must ALWAYS be constructed on the stack, never - * heap allocated. Instances and pointers to them must always be called - * exec_ctx. Instances are always passed as the first argument - * to a function that takes it, and always as a pointer (grpc_exec_ctx - * is never copied). + * - Instance of this must ALWAYS be constructed on the stack, never + * heap allocated. + * - Instances and pointers to them must always be called exec_ctx. + * - Instances are always passed as the first argument to a function that + * takes it, and always as a pointer (grpc_exec_ctx is never copied). */ +#ifndef GRPC_EXECUTION_CONTEXT_SANITIZER struct grpc_exec_ctx { grpc_closure_list closure_list; + /** currently active combiner: updated only via combiner.c */ + grpc_combiner *active_combiner; bool cached_ready_to_finish; void *check_ready_to_finish_arg; bool (*check_ready_to_finish)(grpc_exec_ctx *exec_ctx, void *arg); }; #define GRPC_EXEC_CTX_INIT_WITH_FINISH_CHECK(finish_check, finish_check_arg) \ - { GRPC_CLOSURE_LIST_INIT, false, finish_check_arg, finish_check } + { GRPC_CLOSURE_LIST_INIT, NULL, false, finish_check_arg, finish_check } #else struct grpc_exec_ctx { bool cached_ready_to_finish; diff --git a/src/core/lib/iomgr/socket_windows.c b/src/core/lib/iomgr/socket_windows.c index d7d5f6f157..78ef46d042 100644 --- a/src/core/lib/iomgr/socket_windows.c +++ b/src/core/lib/iomgr/socket_windows.c @@ -84,7 +84,7 @@ void grpc_winsocket_shutdown(grpc_winsocket *winsocket) { DisconnectEx(winsocket->socket, NULL, 0, 0); } else { char *utf8_message = gpr_format_message(WSAGetLastError()); - gpr_log(GPR_ERROR, "Unable to retrieve DisconnectEx pointer : %s", + gpr_log(GPR_INFO, "Unable to retrieve DisconnectEx pointer : %s", utf8_message); gpr_free(utf8_message); } diff --git a/src/core/lib/iomgr/tcp_client_posix.c b/src/core/lib/iomgr/tcp_client_posix.c index 80c7a3f128..3496b6094f 100644 --- a/src/core/lib/iomgr/tcp_client_posix.c +++ b/src/core/lib/iomgr/tcp_client_posix.c @@ -146,61 +146,57 @@ static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) { grpc_timer_cancel(exec_ctx, &ac->alarm); gpr_mu_lock(&ac->mu); - if (error == GRPC_ERROR_NONE) { - do { - so_error_size = sizeof(so_error); - err = getsockopt(grpc_fd_wrapped_fd(fd), SOL_SOCKET, SO_ERROR, &so_error, - &so_error_size); - } while (err < 0 && errno == EINTR); - if (err < 0) { - error = GRPC_OS_ERROR(errno, "getsockopt"); - goto finish; - } else if (so_error != 0) { - if (so_error == ENOBUFS) { - /* We will get one of these errors if we have run out of - memory in the kernel for the data structures allocated - when you connect a socket. If this happens it is very - likely that if we wait a little bit then try again the - connection will work (since other programs or this - program will close their network connections and free up - memory). This does _not_ indicate that there is anything - wrong with the server we are connecting to, this is a - local problem. - - If you are looking at this code, then chances are that - your program or another program on the same computer - opened too many network connections. The "easy" fix: - don't do that! */ - gpr_log(GPR_ERROR, "kernel out of buffers"); - gpr_mu_unlock(&ac->mu); - grpc_fd_notify_on_write(exec_ctx, fd, &ac->write_closure); - return; - } else { - switch (so_error) { - case ECONNREFUSED: - error = grpc_error_set_int(error, GRPC_ERROR_INT_ERRNO, errno); - error = grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, - "Connection refused"); - break; - default: - error = GRPC_OS_ERROR(errno, "getsockopt(SO_ERROR)"); - break; - } - goto finish; - } - } else { - grpc_pollset_set_del_fd(exec_ctx, ac->interested_parties, fd); - *ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str); - fd = NULL; - goto finish; - } - } else { + if (error != GRPC_ERROR_NONE) { error = grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, "Timeout occurred"); goto finish; } - GPR_UNREACHABLE_CODE(return ); + do { + so_error_size = sizeof(so_error); + err = getsockopt(grpc_fd_wrapped_fd(fd), SOL_SOCKET, SO_ERROR, &so_error, + &so_error_size); + } while (err < 0 && errno == EINTR); + if (err < 0) { + error = GRPC_OS_ERROR(errno, "getsockopt"); + goto finish; + } + + switch (so_error) { + case 0: + grpc_pollset_set_del_fd(exec_ctx, ac->interested_parties, fd); + *ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str); + fd = NULL; + break; + case ENOBUFS: + /* We will get one of these errors if we have run out of + memory in the kernel for the data structures allocated + when you connect a socket. If this happens it is very + likely that if we wait a little bit then try again the + connection will work (since other programs or this + program will close their network connections and free up + memory). This does _not_ indicate that there is anything + wrong with the server we are connecting to, this is a + local problem. + + If you are looking at this code, then chances are that + your program or another program on the same computer + opened too many network connections. The "easy" fix: + don't do that! */ + gpr_log(GPR_ERROR, "kernel out of buffers"); + gpr_mu_unlock(&ac->mu); + grpc_fd_notify_on_write(exec_ctx, fd, &ac->write_closure); + return; + case ECONNREFUSED: + /* This error shouldn't happen for anything other than connect(). */ + error = GRPC_OS_ERROR(so_error, "connect"); + break; + default: + /* We don't really know which syscall triggered the problem here, + so punt by reporting getsockopt(). */ + error = GRPC_OS_ERROR(so_error, "getsockopt(SO_ERROR)"); + break; + } finish: if (fd != NULL) { diff --git a/src/core/lib/iomgr/tcp_posix.c b/src/core/lib/iomgr/tcp_posix.c index ec21e03944..92767721d5 100644 --- a/src/core/lib/iomgr/tcp_posix.c +++ b/src/core/lib/iomgr/tcp_posix.c @@ -284,7 +284,7 @@ static void tcp_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, } /* returns true if done, false if pending; if returning true, *error is set */ -#define MAX_WRITE_IOVEC 1024 +#define MAX_WRITE_IOVEC 1000 static bool tcp_flush(grpc_tcp *tcp, grpc_error **error) { struct msghdr msg; struct iovec iov[MAX_WRITE_IOVEC]; @@ -379,10 +379,19 @@ static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */, } if (!tcp_flush(tcp, &error)) { + if (grpc_tcp_trace) { + gpr_log(GPR_DEBUG, "write: delayed"); + } grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_closure); } else { cb = tcp->write_cb; tcp->write_cb = NULL; + if (grpc_tcp_trace) { + const char *str = grpc_error_string(error); + gpr_log(GPR_DEBUG, "write: %s", str); + grpc_error_free_string(str); + } + GPR_TIMER_BEGIN("tcp_handle_write.cb", 0); cb->cb(exec_ctx, cb->cb_arg, error); GPR_TIMER_END("tcp_handle_write.cb", 0); @@ -425,8 +434,16 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, if (!tcp_flush(tcp, &error)) { TCP_REF(tcp, "write"); tcp->write_cb = cb; + if (grpc_tcp_trace) { + gpr_log(GPR_DEBUG, "write: delayed"); + } grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_closure); } else { + if (grpc_tcp_trace) { + const char *str = grpc_error_string(error); + gpr_log(GPR_DEBUG, "write: %s", str); + grpc_error_free_string(str); + } grpc_exec_ctx_sched(exec_ctx, cb, error, NULL); } diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c index 38ebd2dbcb..2d3f6cf9a7 100644 --- a/src/core/lib/iomgr/tcp_server_posix.c +++ b/src/core/lib/iomgr/tcp_server_posix.c @@ -90,10 +90,12 @@ struct grpc_tcp_listener { grpc_closure read_closure; grpc_closure destroyed_closure; struct grpc_tcp_listener *next; - /* When we add a listener, more than one can be created, mainly because of - IPv6. A sibling will still be in the normal list, but will be flagged - as such. Any action, such as ref or unref, will affect all of the - siblings in the list. */ + /* sibling is a linked list of all listeners for a given port. add_port and + clone_port place all new listeners in the same sibling list. A member of + the 'sibling' list is also a member of the 'next' list. The head of each + sibling list has is_sibling==0, and subsequent members of sibling lists + have is_sibling==1. is_sibling allows separate sibling lists to be + identified while iterating through 'next'. */ struct grpc_tcp_listener *sibling; int is_sibling; }; @@ -138,15 +140,17 @@ struct grpc_tcp_server { }; static gpr_once check_init = GPR_ONCE_INIT; -static bool has_so_reuseport; +static bool has_so_reuseport = false; static void init(void) { +#ifndef GPR_MANYLINUX1 int s = socket(AF_INET, SOCK_STREAM, 0); if (s >= 0) { has_so_reuseport = GRPC_LOG_IF_ERROR("check for SO_REUSEPORT", grpc_set_socket_reuse_port(s, 1)); close(s); } +#endif } grpc_error *grpc_tcp_server_create(grpc_closure *shutdown_complete, @@ -306,7 +310,7 @@ static grpc_error *prepare_socket(int fd, const struct sockaddr *addr, GPR_ASSERT(fd >= 0); - if (so_reuseport) { + if (so_reuseport && !grpc_is_unix_socket(addr)) { err = grpc_set_socket_reuse_port(fd, 1); if (err != GRPC_ERROR_NONE) goto error; } @@ -480,6 +484,9 @@ static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd, return err; } +/* Insert count new listeners after listener. Every new listener will have the + same listen address as listener (SO_REUSEPORT must be enabled). Every new + listener is a sibling of listener. */ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) { grpc_tcp_listener *sp = NULL; char *addr_str; @@ -506,6 +513,11 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) { sp = gpr_malloc(sizeof(grpc_tcp_listener)); sp->next = listener->next; listener->next = sp; + /* sp (the new listener) is a sibling of 'listener' (the original + listener). */ + sp->is_sibling = 1; + sp->sibling = listener->sibling; + listener->sibling = sp; sp->server = listener->server; sp->fd = fd; sp->emfd = grpc_fd_create(fd, name); @@ -514,8 +526,6 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) { sp->port = port; sp->port_index = listener->port_index; sp->fd_index = listener->fd_index + count - i; - sp->is_sibling = 1; - sp->sibling = listener->is_sibling ? listener->sibling : listener; GPR_ASSERT(sp->emfd); while (listener->server->tail->next != NULL) { listener->server->tail = listener->server->tail->next; @@ -685,7 +695,8 @@ void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s, s->pollset_count = pollset_count; sp = s->head; while (sp != NULL) { - if (s->so_reuseport && pollset_count > 1) { + if (s->so_reuseport && !grpc_is_unix_socket(&sp->addr.sockaddr) && + pollset_count > 1) { GPR_ASSERT(GRPC_LOG_IF_ERROR( "clone_port", clone_port(sp, (unsigned)(pollset_count - 1)))); for (i = 0; i < pollset_count; i++) { diff --git a/src/core/lib/iomgr/tcp_windows.c b/src/core/lib/iomgr/tcp_windows.c index 35054c42b5..448a72671c 100644 --- a/src/core/lib/iomgr/tcp_windows.c +++ b/src/core/lib/iomgr/tcp_windows.c @@ -319,6 +319,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, ? GRPC_ERROR_NONE : GRPC_WSA_ERROR(info->wsa_error, "WSASend"); grpc_exec_ctx_sched(exec_ctx, cb, error, NULL); + if (allocated) gpr_free(allocated); return; } diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c index f68288169b..df68402981 100644 --- a/src/core/lib/iomgr/udp_server.c +++ b/src/core/lib/iomgr/udp_server.c @@ -170,6 +170,8 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) { sp->destroyed_closure.cb = destroyed_port; sp->destroyed_closure.cb_arg = s; + /* Call the orphan_cb to signal that the FD is about to be closed and + * should no longer be used. */ GPR_ASSERT(sp->orphan_cb); sp->orphan_cb(sp->emfd); @@ -196,6 +198,11 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, /* shutdown all fd's */ if (s->active_ports) { for (i = 0; i < s->nports; i++) { + /* Call the orphan_cb to signal that the FD is about to be closed and + * should no longer be used. */ + GPR_ASSERT(sp->orphan_cb); + sp->orphan_cb(sp->emfd); + grpc_fd_shutdown(exec_ctx, s->ports[i].emfd); } gpr_mu_unlock(&s->mu); diff --git a/src/core/lib/iomgr/workqueue.h b/src/core/lib/iomgr/workqueue.h index 7156e490d7..b2805dc66c 100644 --- a/src/core/lib/iomgr/workqueue.h +++ b/src/core/lib/iomgr/workqueue.h @@ -50,10 +50,6 @@ /* grpc_workqueue is forward declared in exec_ctx.h */ -/* Deprecated: do not use. - This has *already* been removed in a future commit. */ -void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue); - /* Reference counting functions. Use the macro's always (GRPC_WORKQUEUE_{REF,UNREF}). diff --git a/src/core/lib/iomgr/workqueue_posix.c b/src/core/lib/iomgr/workqueue_posix.c index e0d6dac230..ecfea68f56 100644 --- a/src/core/lib/iomgr/workqueue_posix.c +++ b/src/core/lib/iomgr/workqueue_posix.c @@ -44,6 +44,7 @@ #include <grpc/support/useful.h> #include "src/core/lib/iomgr/ev_posix.h" +#include "src/core/lib/profiling/timers.h" static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error); @@ -52,8 +53,7 @@ grpc_error *grpc_workqueue_create(grpc_exec_ctx *exec_ctx, char name[32]; *workqueue = gpr_malloc(sizeof(grpc_workqueue)); gpr_ref_init(&(*workqueue)->refs, 1); - gpr_mu_init(&(*workqueue)->mu); - (*workqueue)->closure_list.head = (*workqueue)->closure_list.tail = NULL; + gpr_atm_no_barrier_store(&(*workqueue)->state, 1); grpc_error *err = grpc_wakeup_fd_init(&(*workqueue)->wakeup_fd); if (err != GRPC_ERROR_NONE) { gpr_free(*workqueue); @@ -62,6 +62,7 @@ grpc_error *grpc_workqueue_create(grpc_exec_ctx *exec_ctx, sprintf(name, "workqueue:%p", (void *)(*workqueue)); (*workqueue)->wakeup_read_fd = grpc_fd_create( GRPC_WAKEUP_FD_GET_READ_FD(&(*workqueue)->wakeup_fd), name); + gpr_mpscq_init(&(*workqueue)->queue); grpc_closure_init(&(*workqueue)->read_closure, on_readable, *workqueue); grpc_fd_notify_on_read(exec_ctx, (*workqueue)->wakeup_read_fd, &(*workqueue)->read_closure); @@ -70,57 +71,79 @@ grpc_error *grpc_workqueue_create(grpc_exec_ctx *exec_ctx, static void workqueue_destroy(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) { - grpc_exec_ctx_enqueue_list(exec_ctx, &workqueue->closure_list, NULL); grpc_fd_shutdown(exec_ctx, workqueue->wakeup_read_fd); } +static void workqueue_orphan(grpc_exec_ctx *exec_ctx, + grpc_workqueue *workqueue) { + if (gpr_atm_full_fetch_add(&workqueue->state, -1) == 1) { + workqueue_destroy(exec_ctx, workqueue); + } +} + #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG void grpc_workqueue_ref(grpc_workqueue *workqueue, const char *file, int line, const char *reason) { + if (workqueue == NULL) return; gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "WORKQUEUE:%p ref %d -> %d %s", workqueue, (int)workqueue->refs.count, (int)workqueue->refs.count + 1, reason); + gpr_ref(&workqueue->refs); +} #else void grpc_workqueue_ref(grpc_workqueue *workqueue) { -#endif + if (workqueue == NULL) return; gpr_ref(&workqueue->refs); } +#endif #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG void grpc_workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue, const char *file, int line, const char *reason) { + if (workqueue == NULL) return; gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "WORKQUEUE:%p unref %d -> %d %s", workqueue, (int)workqueue->refs.count, (int)workqueue->refs.count - 1, reason); + if (gpr_unref(&workqueue->refs)) { + workqueue_orphan(exec_ctx, workqueue); + } +} #else void grpc_workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) { -#endif + if (workqueue == NULL) return; if (gpr_unref(&workqueue->refs)) { - workqueue_destroy(exec_ctx, workqueue); + workqueue_orphan(exec_ctx, workqueue); } } +#endif + +static void drain(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) { + abort(); +} -void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) { - gpr_mu_lock(&workqueue->mu); - grpc_exec_ctx_enqueue_list(exec_ctx, &workqueue->closure_list, NULL); - gpr_mu_unlock(&workqueue->mu); +static void wakeup(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) { + GPR_TIMER_MARK("workqueue.wakeup", 0); + grpc_error *err = grpc_wakeup_fd_wakeup(&workqueue->wakeup_fd); + if (!GRPC_LOG_IF_ERROR("wakeupfd_wakeup", err)) { + drain(exec_ctx, workqueue); + } } static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + GPR_TIMER_BEGIN("workqueue.on_readable", 0); + grpc_workqueue *workqueue = arg; if (error != GRPC_ERROR_NONE) { - gpr_mu_destroy(&workqueue->mu); /* HACK: let wakeup_fd code know that we stole the fd */ workqueue->wakeup_fd.read_fd = 0; grpc_wakeup_fd_destroy(&workqueue->wakeup_fd); grpc_fd_orphan(exec_ctx, workqueue->wakeup_read_fd, NULL, NULL, "destroy"); + GPR_ASSERT(gpr_atm_no_barrier_load(&workqueue->state) == 0); gpr_free(workqueue); } else { - gpr_mu_lock(&workqueue->mu); - grpc_exec_ctx_enqueue_list(exec_ctx, &workqueue->closure_list, NULL); error = grpc_wakeup_fd_consume_wakeup(&workqueue->wakeup_fd); - gpr_mu_unlock(&workqueue->mu); + gpr_mpscq_node *n = gpr_mpscq_pop(&workqueue->queue); if (error == GRPC_ERROR_NONE) { grpc_fd_notify_on_read(exec_ctx, workqueue->wakeup_read_fd, &workqueue->read_closure); @@ -128,24 +151,46 @@ static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { /* recurse to get error handling */ on_readable(exec_ctx, arg, error); } + if (n == NULL) { + /* try again - queue in an inconsistant state */ + wakeup(exec_ctx, workqueue); + } else { + switch (gpr_atm_full_fetch_add(&workqueue->state, -2)) { + case 3: // had one count, one unorphaned --> done, unorphaned + break; + case 2: // had one count, one orphaned --> done, orphaned + workqueue_destroy(exec_ctx, workqueue); + break; + case 1: + case 0: + // these values are illegal - representing an already done or + // deleted workqueue + GPR_UNREACHABLE_CODE(break); + default: + // schedule a wakeup since there's more to do + wakeup(exec_ctx, workqueue); + } + grpc_closure *cl = (grpc_closure *)n; + grpc_error *clerr = cl->error; + cl->cb(exec_ctx, cl->cb_arg, clerr); + GRPC_ERROR_UNREF(clerr); + } } + + GPR_TIMER_END("workqueue.on_readable", 0); } void grpc_workqueue_enqueue(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue, grpc_closure *closure, grpc_error *error) { - grpc_error *push_error = GRPC_ERROR_NONE; - gpr_mu_lock(&workqueue->mu); - if (grpc_closure_list_empty(workqueue->closure_list)) { - push_error = grpc_wakeup_fd_wakeup(&workqueue->wakeup_fd); - } - grpc_closure_list_append(&workqueue->closure_list, closure, error); - if (push_error != GRPC_ERROR_NONE) { - const char *msg = grpc_error_string(push_error); - gpr_log(GPR_ERROR, "Failed to push to workqueue: %s", msg); - grpc_error_free_string(msg); - grpc_exec_ctx_enqueue_list(exec_ctx, &workqueue->closure_list, NULL); + GPR_TIMER_BEGIN("workqueue.enqueue", 0); + gpr_atm last = gpr_atm_full_fetch_add(&workqueue->state, 2); + GPR_ASSERT(last & 1); + closure->error = error; + gpr_mpscq_push(&workqueue->queue, &closure->next_data.atm_next); + if (last == 1) { + wakeup(exec_ctx, workqueue); } - gpr_mu_unlock(&workqueue->mu); + GPR_TIMER_END("workqueue.enqueue", 0); } #endif /* GPR_POSIX_SOCKET */ diff --git a/src/core/lib/iomgr/workqueue_posix.h b/src/core/lib/iomgr/workqueue_posix.h index 0f26ba58e2..03ee21cef7 100644 --- a/src/core/lib/iomgr/workqueue_posix.h +++ b/src/core/lib/iomgr/workqueue_posix.h @@ -35,14 +35,17 @@ #define GRPC_CORE_LIB_IOMGR_WORKQUEUE_POSIX_H #include "src/core/lib/iomgr/wakeup_fd_posix.h" +#include "src/core/lib/support/mpscq.h" struct grpc_fd; struct grpc_workqueue { gpr_refcount refs; - - gpr_mu mu; - grpc_closure_list closure_list; + gpr_mpscq queue; + // state is: + // lower bit - zero if orphaned + // other bits - number of items enqueued + gpr_atm state; grpc_wakeup_fd wakeup_fd; struct grpc_fd *wakeup_read_fd; diff --git a/src/core/lib/iomgr/workqueue_windows.c b/src/core/lib/iomgr/workqueue_windows.c index 23e2dea185..ee81dc248e 100644 --- a/src/core/lib/iomgr/workqueue_windows.c +++ b/src/core/lib/iomgr/workqueue_windows.c @@ -42,8 +42,6 @@ // context, which is at least correct, if not performant or in the spirit of // workqueues. -void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) {} - #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG void grpc_workqueue_ref(grpc_workqueue *workqueue, const char *file, int line, const char *reason) {} diff --git a/src/core/lib/json/json_reader.c b/src/core/lib/json/json_reader.c index bc04bccc65..5b42ca53ff 100644 --- a/src/core/lib/json/json_reader.c +++ b/src/core/lib/json/json_reader.c @@ -1,6 +1,6 @@ /* * - * Copyright 2015, Google Inc. + * Copyright 2015-2016, Google Inc. * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -171,8 +171,9 @@ grpc_json_reader_status grpc_json_reader_run(grpc_json_reader *reader) { switch (reader->state) { case GRPC_JSON_STATE_OBJECT_KEY_STRING: case GRPC_JSON_STATE_VALUE_STRING: - if (reader->unicode_high_surrogate != 0) + if (reader->unicode_high_surrogate != 0) { return GRPC_JSON_PARSE_ERROR; + } json_reader_string_add_char(reader, c); break; @@ -289,8 +290,9 @@ grpc_json_reader_status grpc_json_reader_run(grpc_json_reader *reader) { break; case GRPC_JSON_STATE_OBJECT_KEY_STRING: - if (reader->unicode_high_surrogate != 0) + if (reader->unicode_high_surrogate != 0) { return GRPC_JSON_PARSE_ERROR; + } if (c == '"') { reader->state = GRPC_JSON_STATE_OBJECT_KEY_END; json_reader_set_key(reader); @@ -302,8 +304,9 @@ grpc_json_reader_status grpc_json_reader_run(grpc_json_reader *reader) { break; case GRPC_JSON_STATE_VALUE_STRING: - if (reader->unicode_high_surrogate != 0) + if (reader->unicode_high_surrogate != 0) { return GRPC_JSON_PARSE_ERROR; + } if (c == '"') { reader->state = GRPC_JSON_STATE_VALUE_END; json_reader_set_string(reader); @@ -383,8 +386,9 @@ grpc_json_reader_status grpc_json_reader_run(grpc_json_reader *reader) { } else { reader->state = GRPC_JSON_STATE_VALUE_STRING; } - if (reader->unicode_high_surrogate && c != 'u') + if (reader->unicode_high_surrogate && c != 'u') { return GRPC_JSON_PARSE_ERROR; + } switch (c) { case '"': case '/': diff --git a/src/core/lib/security/context/security_context.c b/src/core/lib/security/context/security_context.c index 127b13ee50..2204fadf54 100644 --- a/src/core/lib/security/context/security_context.c +++ b/src/core/lib/security/context/security_context.c @@ -99,6 +99,9 @@ void grpc_client_security_context_destroy(void *ctx) { grpc_client_security_context *c = (grpc_client_security_context *)ctx; grpc_call_credentials_unref(c->creds); GRPC_AUTH_CONTEXT_UNREF(c->auth_context, "client_security_context"); + if (c->extension.instance != NULL && c->extension.destroy != NULL) { + c->extension.destroy(c->extension.instance); + } gpr_free(ctx); } @@ -114,6 +117,9 @@ grpc_server_security_context *grpc_server_security_context_create(void) { void grpc_server_security_context_destroy(void *ctx) { grpc_server_security_context *c = (grpc_server_security_context *)ctx; GRPC_AUTH_CONTEXT_UNREF(c->auth_context, "server_security_context"); + if (c->extension.instance != NULL && c->extension.destroy != NULL) { + c->extension.destroy(c->extension.instance); + } gpr_free(ctx); } diff --git a/src/core/lib/security/context/security_context.h b/src/core/lib/security/context/security_context.h index ef0c06b1fb..1e131a0c23 100644 --- a/src/core/lib/security/context/security_context.h +++ b/src/core/lib/security/context/security_context.h @@ -37,6 +37,10 @@ #include "src/core/lib/iomgr/pollset.h" #include "src/core/lib/security/credentials/credentials.h" +#ifdef __cplusplus +extern "C" { +#endif + /* --- grpc_auth_context --- High level authentication context object. Can optionally be chained. */ @@ -80,6 +84,16 @@ void grpc_auth_context_unref(grpc_auth_context *policy); void grpc_auth_property_reset(grpc_auth_property *property); +/* --- grpc_security_context_extension --- + + Extension to the security context that may be set in a filter and accessed + later by a higher level method on a grpc_call object. */ + +typedef struct { + void *instance; + void (*destroy)(void *); +} grpc_security_context_extension; + /* --- grpc_client_security_context --- Internal client-side security context. */ @@ -87,6 +101,7 @@ void grpc_auth_property_reset(grpc_auth_property *property); typedef struct { grpc_call_credentials *creds; grpc_auth_context *auth_context; + grpc_security_context_extension extension; } grpc_client_security_context; grpc_client_security_context *grpc_client_security_context_create(void); @@ -98,6 +113,7 @@ void grpc_client_security_context_destroy(void *ctx); typedef struct { grpc_auth_context *auth_context; + grpc_security_context_extension extension; } grpc_server_security_context; grpc_server_security_context *grpc_server_security_context_create(void); @@ -111,4 +127,8 @@ grpc_auth_context *grpc_auth_context_from_arg(const grpc_arg *arg); grpc_auth_context *grpc_find_auth_context_in_args( const grpc_channel_args *args); +#ifdef __cplusplus +} +#endif + #endif /* GRPC_CORE_LIB_SECURITY_CONTEXT_SECURITY_CONTEXT_H */ diff --git a/src/core/lib/security/credentials/jwt/json_token.c b/src/core/lib/security/credentials/jwt/json_token.c index 354c13133e..192a5f47ed 100644 --- a/src/core/lib/security/credentials/jwt/json_token.c +++ b/src/core/lib/security/credentials/jwt/json_token.c @@ -38,6 +38,7 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/string_util.h> +#include <grpc/support/time.h> #include "src/core/lib/security/util/b64.h" #include "src/core/lib/security/util/json_util.h" diff --git a/src/core/lib/security/credentials/plugin/plugin_credentials.c b/src/core/lib/security/credentials/plugin/plugin_credentials.c index 824ff081dc..905de3723e 100644 --- a/src/core/lib/security/credentials/plugin/plugin_credentials.c +++ b/src/core/lib/security/credentials/plugin/plugin_credentials.c @@ -37,6 +37,7 @@ #include "src/core/lib/surface/api_trace.h" +#include <grpc/grpc.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/string_util.h> @@ -71,17 +72,33 @@ static void plugin_md_request_metadata_ready(void *request, error_details); } else { size_t i; + bool seen_illegal_header = false; grpc_credentials_md *md_array = NULL; - if (num_md > 0) { + for (i = 0; i < num_md; i++) { + if (!grpc_header_key_is_legal(md[i].key, strlen(md[i].key))) { + gpr_log(GPR_ERROR, "Plugin added invalid metadata key: %s", md[i].key); + seen_illegal_header = true; + break; + } else if (!grpc_is_binary_header(md[i].key, strlen(md[i].key)) && + !grpc_header_nonbin_value_is_legal(md[i].value, + md[i].value_length)) { + gpr_log(GPR_ERROR, "Plugin added invalid metadata value."); + seen_illegal_header = true; + break; + } + } + if (seen_illegal_header) { + r->cb(&exec_ctx, r->user_data, NULL, 0, GRPC_CREDENTIALS_ERROR, + "Illegal metadata"); + } else if (num_md > 0) { md_array = gpr_malloc(num_md * sizeof(grpc_credentials_md)); for (i = 0; i < num_md; i++) { md_array[i].key = gpr_slice_from_copied_string(md[i].key); md_array[i].value = gpr_slice_from_copied_buffer(md[i].value, md[i].value_length); } - } - r->cb(&exec_ctx, r->user_data, md_array, num_md, GRPC_CREDENTIALS_OK, NULL); - if (md_array != NULL) { + r->cb(&exec_ctx, r->user_data, md_array, num_md, GRPC_CREDENTIALS_OK, + NULL); for (i = 0; i < num_md; i++) { gpr_slice_unref(md_array[i].key); gpr_slice_unref(md_array[i].value); diff --git a/src/core/lib/security/transport/client_auth_filter.c b/src/core/lib/security/transport/client_auth_filter.c index ed7929aa27..b366d1410f 100644 --- a/src/core/lib/security/transport/client_auth_filter.c +++ b/src/core/lib/security/transport/client_auth_filter.c @@ -40,6 +40,7 @@ #include <grpc/support/string_util.h> #include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/profiling/timers.h" #include "src/core/lib/security/context/security_context.h" #include "src/core/lib/security/credentials/credentials.h" #include "src/core/lib/security/transport/security_connector.h" @@ -218,6 +219,8 @@ static void on_host_checked(grpc_exec_ctx *exec_ctx, void *user_data, static void auth_start_transport_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op *op) { + GPR_TIMER_BEGIN("auth_start_transport_op", 0); + /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; @@ -258,19 +261,23 @@ static void auth_start_transport_op(grpc_exec_ctx *exec_ctx, grpc_channel_security_connector_check_call_host( exec_ctx, chand->security_connector, call_host, chand->auth_context, on_host_checked, elem); + GPR_TIMER_END("auth_start_transport_op", 0); return; /* early exit */ } } /* pass control down the stack */ grpc_call_next_op(exec_ctx, elem, op); + GPR_TIMER_END("auth_start_transport_op", 0); } /* Constructor for call_data */ -static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { +static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { call_data *calld = elem->call_data; memset(calld, 0, sizeof(*calld)); + return GRPC_ERROR_NONE; } static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, @@ -282,7 +289,8 @@ static void set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, void *ignored) { + const grpc_call_final_info *final_info, + void *ignored) { call_data *calld = elem->call_data; grpc_call_credentials_unref(calld->creds); if (calld->host != NULL) { diff --git a/src/core/lib/security/transport/handshake.c b/src/core/lib/security/transport/handshake.c index 540a17283d..fbeec312b6 100644 --- a/src/core/lib/security/transport/handshake.c +++ b/src/core/lib/security/transport/handshake.c @@ -325,8 +325,9 @@ static void on_timeout(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { void grpc_do_security_handshake( grpc_exec_ctx *exec_ctx, tsi_handshaker *handshaker, grpc_security_connector *connector, bool is_client_side, - grpc_endpoint *nonsecure_endpoint, gpr_timespec deadline, - grpc_security_handshake_done_cb cb, void *user_data) { + grpc_endpoint *nonsecure_endpoint, gpr_slice_buffer *read_buffer, + gpr_timespec deadline, grpc_security_handshake_done_cb cb, + void *user_data) { grpc_security_connector_handshake_list *handshake_node; grpc_security_handshake *h = gpr_malloc(sizeof(grpc_security_handshake)); memset(h, 0, sizeof(grpc_security_handshake)); @@ -346,6 +347,10 @@ void grpc_do_security_handshake( gpr_slice_buffer_init(&h->left_overs); gpr_slice_buffer_init(&h->outgoing); gpr_slice_buffer_init(&h->incoming); + if (read_buffer != NULL) { + gpr_slice_buffer_move_into(read_buffer, &h->incoming); + gpr_free(read_buffer); + } if (!is_client_side) { grpc_server_security_connector *server_connector = (grpc_server_security_connector *)connector; diff --git a/src/core/lib/security/transport/handshake.h b/src/core/lib/security/transport/handshake.h index c0906dd6af..53092f5421 100644 --- a/src/core/lib/security/transport/handshake.h +++ b/src/core/lib/security/transport/handshake.h @@ -37,12 +37,13 @@ #include "src/core/lib/iomgr/endpoint.h" #include "src/core/lib/security/transport/security_connector.h" -/* Calls the callback upon completion. Takes owership of handshaker. */ +/* Calls the callback upon completion. Takes owership of handshaker and + * read_buffer. */ void grpc_do_security_handshake( grpc_exec_ctx *exec_ctx, tsi_handshaker *handshaker, grpc_security_connector *connector, bool is_client_side, - grpc_endpoint *nonsecure_endpoint, gpr_timespec deadline, - grpc_security_handshake_done_cb cb, void *user_data); + grpc_endpoint *nonsecure_endpoint, gpr_slice_buffer *read_buffer, + gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data); void grpc_security_handshake_shutdown(grpc_exec_ctx *exec_ctx, void *handshake); diff --git a/src/core/lib/security/transport/secure_endpoint.c b/src/core/lib/security/transport/secure_endpoint.c index bc50f9d1b0..acb0113ea8 100644 --- a/src/core/lib/security/transport/secure_endpoint.c +++ b/src/core/lib/security/transport/secure_endpoint.c @@ -38,6 +38,7 @@ #include <grpc/support/slice_buffer.h> #include <grpc/support/sync.h> #include "src/core/lib/debug/trace.h" +#include "src/core/lib/profiling/timers.h" #include "src/core/lib/security/transport/tsi_error.h" #include "src/core/lib/support/string.h" #include "src/core/lib/tsi/transport_security_interface.h" @@ -128,7 +129,7 @@ static void flush_read_staging_buffer(secure_endpoint *ep, uint8_t **cur, static void call_read_cb(grpc_exec_ctx *exec_ctx, secure_endpoint *ep, grpc_error *error) { - if (false && grpc_trace_secure_endpoint) { + if (grpc_trace_secure_endpoint) { size_t i; for (i = 0; i < ep->read_buffer->count; i++) { char *data = gpr_dump_slice(ep->read_buffer->slices[i], @@ -248,6 +249,8 @@ static void flush_write_staging_buffer(secure_endpoint *ep, uint8_t **cur, static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, gpr_slice_buffer *slices, grpc_closure *cb) { + GPR_TIMER_BEGIN("secure_endpoint.endpoint_write", 0); + unsigned i; tsi_result result = TSI_OK; secure_endpoint *ep = (secure_endpoint *)secure_ep; @@ -256,7 +259,7 @@ static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, gpr_slice_buffer_reset_and_unref(&ep->output_buffer); - if (false && grpc_trace_secure_endpoint) { + if (grpc_trace_secure_endpoint) { for (i = 0; i < slices->count; i++) { char *data = gpr_dump_slice(slices->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); @@ -323,10 +326,12 @@ static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep, exec_ctx, cb, grpc_set_tsi_error_result(GRPC_ERROR_CREATE("Wrap failed"), result), NULL); + GPR_TIMER_END("secure_endpoint.endpoint_write", 0); return; } grpc_endpoint_write(exec_ctx, ep->wrapped_ep, &ep->output_buffer, cb); + GPR_TIMER_END("secure_endpoint.endpoint_write", 0); } static void endpoint_shutdown(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/security/transport/security_connector.c b/src/core/lib/security/transport/security_connector.c index f0ee6770e5..0eca46eb52 100644 --- a/src/core/lib/security/transport/security_connector.c +++ b/src/core/lib/security/transport/security_connector.c @@ -127,25 +127,29 @@ void grpc_server_security_connector_shutdown( void grpc_channel_security_connector_do_handshake( grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc, - grpc_endpoint *nonsecure_endpoint, gpr_timespec deadline, - grpc_security_handshake_done_cb cb, void *user_data) { + grpc_endpoint *nonsecure_endpoint, gpr_slice_buffer *read_buffer, + gpr_timespec deadline, grpc_security_handshake_done_cb cb, + void *user_data) { if (sc == NULL || nonsecure_endpoint == NULL) { + gpr_free(read_buffer); cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL, NULL); } else { - sc->do_handshake(exec_ctx, sc, nonsecure_endpoint, deadline, cb, user_data); + sc->do_handshake(exec_ctx, sc, nonsecure_endpoint, read_buffer, deadline, + cb, user_data); } } void grpc_server_security_connector_do_handshake( grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc, grpc_tcp_server_acceptor *acceptor, grpc_endpoint *nonsecure_endpoint, - gpr_timespec deadline, grpc_security_handshake_done_cb cb, - void *user_data) { + gpr_slice_buffer *read_buffer, gpr_timespec deadline, + grpc_security_handshake_done_cb cb, void *user_data) { if (sc == NULL || nonsecure_endpoint == NULL) { + gpr_free(read_buffer); cb(exec_ctx, user_data, GRPC_SECURITY_ERROR, NULL, NULL); } else { - sc->do_handshake(exec_ctx, sc, acceptor, nonsecure_endpoint, deadline, cb, - user_data); + sc->do_handshake(exec_ctx, sc, acceptor, nonsecure_endpoint, read_buffer, + deadline, cb, user_data); } } @@ -312,23 +316,23 @@ static void fake_channel_check_call_host(grpc_exec_ctx *exec_ctx, static void fake_channel_do_handshake(grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc, grpc_endpoint *nonsecure_endpoint, + gpr_slice_buffer *read_buffer, gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data) { grpc_do_security_handshake(exec_ctx, tsi_create_fake_handshaker(1), &sc->base, - true, nonsecure_endpoint, deadline, cb, user_data); + true, nonsecure_endpoint, read_buffer, deadline, + cb, user_data); } -static void fake_server_do_handshake(grpc_exec_ctx *exec_ctx, - grpc_server_security_connector *sc, - grpc_tcp_server_acceptor *acceptor, - grpc_endpoint *nonsecure_endpoint, - gpr_timespec deadline, - grpc_security_handshake_done_cb cb, - void *user_data) { +static void fake_server_do_handshake( + grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc, + grpc_tcp_server_acceptor *acceptor, grpc_endpoint *nonsecure_endpoint, + gpr_slice_buffer *read_buffer, gpr_timespec deadline, + grpc_security_handshake_done_cb cb, void *user_data) { grpc_do_security_handshake(exec_ctx, tsi_create_fake_handshaker(0), &sc->base, - false, nonsecure_endpoint, deadline, cb, - user_data); + false, nonsecure_endpoint, read_buffer, deadline, + cb, user_data); } static grpc_security_connector_vtable fake_channel_vtable = { @@ -418,6 +422,7 @@ static grpc_security_status ssl_create_handshaker( static void ssl_channel_do_handshake(grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc, grpc_endpoint *nonsecure_endpoint, + gpr_slice_buffer *read_buffer, gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data) { @@ -430,30 +435,32 @@ static void ssl_channel_do_handshake(grpc_exec_ctx *exec_ctx, : c->target_name, &handshaker); if (status != GRPC_SECURITY_OK) { + gpr_free(read_buffer); cb(exec_ctx, user_data, status, NULL, NULL); } else { grpc_do_security_handshake(exec_ctx, handshaker, &sc->base, true, - nonsecure_endpoint, deadline, cb, user_data); + nonsecure_endpoint, read_buffer, deadline, cb, + user_data); } } -static void ssl_server_do_handshake(grpc_exec_ctx *exec_ctx, - grpc_server_security_connector *sc, - grpc_tcp_server_acceptor *acceptor, - grpc_endpoint *nonsecure_endpoint, - gpr_timespec deadline, - grpc_security_handshake_done_cb cb, - void *user_data) { +static void ssl_server_do_handshake( + grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc, + grpc_tcp_server_acceptor *acceptor, grpc_endpoint *nonsecure_endpoint, + gpr_slice_buffer *read_buffer, gpr_timespec deadline, + grpc_security_handshake_done_cb cb, void *user_data) { grpc_ssl_server_security_connector *c = (grpc_ssl_server_security_connector *)sc; tsi_handshaker *handshaker; grpc_security_status status = ssl_create_handshaker(c->handshaker_factory, false, NULL, &handshaker); if (status != GRPC_SECURITY_OK) { + gpr_free(read_buffer); cb(exec_ctx, user_data, status, NULL, NULL); } else { grpc_do_security_handshake(exec_ctx, handshaker, &sc->base, false, - nonsecure_endpoint, deadline, cb, user_data); + nonsecure_endpoint, read_buffer, deadline, cb, + user_data); } } diff --git a/src/core/lib/security/transport/security_connector.h b/src/core/lib/security/transport/security_connector.h index c2ddf5ee1e..0b5b44bf1a 100644 --- a/src/core/lib/security/transport/security_connector.h +++ b/src/core/lib/security/transport/security_connector.h @@ -143,7 +143,8 @@ struct grpc_channel_security_connector { grpc_security_call_host_check_cb cb, void *user_data); void (*do_handshake)(grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *sc, - grpc_endpoint *nonsecure_endpoint, gpr_timespec deadline, + grpc_endpoint *nonsecure_endpoint, + gpr_slice_buffer *read_buffer, gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data); }; @@ -156,8 +157,8 @@ void grpc_channel_security_connector_check_call_host( /* Handshake. */ void grpc_channel_security_connector_do_handshake( grpc_exec_ctx *exec_ctx, grpc_channel_security_connector *connector, - grpc_endpoint *nonsecure_endpoint, gpr_timespec deadline, - grpc_security_handshake_done_cb cb, void *user_data); + grpc_endpoint *nonsecure_endpoint, gpr_slice_buffer *read_buffer, + gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data); /* --- server_security_connector object. --- @@ -174,14 +175,16 @@ struct grpc_server_security_connector { void (*do_handshake)(grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc, grpc_tcp_server_acceptor *acceptor, - grpc_endpoint *nonsecure_endpoint, gpr_timespec deadline, + grpc_endpoint *nonsecure_endpoint, + gpr_slice_buffer *read_buffer, gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data); }; void grpc_server_security_connector_do_handshake( grpc_exec_ctx *exec_ctx, grpc_server_security_connector *sc, grpc_tcp_server_acceptor *acceptor, grpc_endpoint *nonsecure_endpoint, - gpr_timespec deadline, grpc_security_handshake_done_cb cb, void *user_data); + gpr_slice_buffer *read_buffer, gpr_timespec deadline, + grpc_security_handshake_done_cb cb, void *user_data); void grpc_server_security_connector_shutdown( grpc_exec_ctx *exec_ctx, grpc_server_security_connector *connector); diff --git a/src/core/lib/security/transport/server_auth_filter.c b/src/core/lib/security/transport/server_auth_filter.c index 12e789bde9..b2c6815af8 100644 --- a/src/core/lib/security/transport/server_auth_filter.c +++ b/src/core/lib/security/transport/server_auth_filter.c @@ -48,7 +48,7 @@ typedef struct call_data { up-call on transport_op, and remember to call our on_done_recv member after handling it. */ grpc_closure auth_on_recv; - grpc_transport_stream_op transport_op; + grpc_transport_stream_op *transport_op; grpc_metadata_array md; const grpc_metadata *consumed_md; size_t num_consumed_md; @@ -106,6 +106,10 @@ static grpc_mdelem *remove_consumed_md(void *user_data, grpc_mdelem *md) { return md; } +static void destroy_op(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + gpr_free(arg); +} + /* called from application code */ static void on_md_processing_done( void *user_data, const grpc_metadata *consumed_md, size_t num_consumed_md, @@ -131,21 +135,22 @@ static void on_md_processing_done( grpc_exec_ctx_sched(&exec_ctx, calld->on_done_recv, GRPC_ERROR_NONE, NULL); } else { gpr_slice message; - grpc_transport_stream_op close_op; - memset(&close_op, 0, sizeof(close_op)); + grpc_transport_stream_op *close_op = gpr_malloc(sizeof(*close_op)); + memset(close_op, 0, sizeof(*close_op)); grpc_metadata_array_destroy(&calld->md); error_details = error_details != NULL ? error_details : "Authentication metadata processing failed."; message = gpr_slice_from_copied_string(error_details); - calld->transport_op.send_initial_metadata = NULL; - if (calld->transport_op.send_message != NULL) { - grpc_byte_stream_destroy(&exec_ctx, calld->transport_op.send_message); - calld->transport_op.send_message = NULL; + calld->transport_op->send_initial_metadata = NULL; + if (calld->transport_op->send_message != NULL) { + grpc_byte_stream_destroy(&exec_ctx, calld->transport_op->send_message); + calld->transport_op->send_message = NULL; } - calld->transport_op.send_trailing_metadata = NULL; - grpc_transport_stream_op_add_close(&close_op, status, &message); - grpc_call_next_op(&exec_ctx, elem, &close_op); + calld->transport_op->send_trailing_metadata = NULL; + close_op->on_complete = grpc_closure_create(destroy_op, close_op); + grpc_transport_stream_op_add_close(close_op, status, &message); + grpc_call_next_op(&exec_ctx, elem, close_op); grpc_exec_ctx_sched(&exec_ctx, calld->on_done_recv, grpc_error_set_int(GRPC_ERROR_CREATE(error_details), GRPC_ERROR_INT_GRPC_STATUS, status), @@ -182,7 +187,7 @@ static void set_recv_ops_md_callbacks(grpc_call_element *elem, calld->recv_initial_metadata = op->recv_initial_metadata; calld->on_done_recv = op->recv_initial_metadata_ready; op->recv_initial_metadata_ready = &calld->auth_on_recv; - calld->transport_op = *op; + calld->transport_op = op; } } @@ -199,8 +204,9 @@ static void auth_start_transport_op(grpc_exec_ctx *exec_ctx, } /* Constructor for call_data */ -static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { +static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; @@ -222,11 +228,14 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, args->context[GRPC_CONTEXT_SECURITY].value = server_ctx; args->context[GRPC_CONTEXT_SECURITY].destroy = grpc_server_security_context_destroy; + + return GRPC_ERROR_NONE; } /* Destructor for call_data */ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, void *ignored) {} + const grpc_call_final_info *final_info, + void *ignored) {} /* Constructor for channel_data */ static void init_channel_elem(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/support/log_linux.c b/src/core/lib/support/log_linux.c index 508fae4eec..299b377373 100644 --- a/src/core/lib/support/log_linux.c +++ b/src/core/lib/support/log_linux.c @@ -47,7 +47,6 @@ #include <grpc/support/log.h> #include <grpc/support/string_util.h> #include <grpc/support/time.h> -#include <linux/unistd.h> #include <stdarg.h> #include <stdio.h> #include <string.h> diff --git a/src/core/lib/support/log_posix.c b/src/core/lib/support/log_posix.c index 6ae6320767..f972da0887 100644 --- a/src/core/lib/support/log_posix.c +++ b/src/core/lib/support/log_posix.c @@ -33,7 +33,7 @@ #include <grpc/support/port_platform.h> -#if defined(GPR_POSIX_LOG) +#ifdef GPR_POSIX_LOG #include <grpc/support/alloc.h> #include <grpc/support/log.h> diff --git a/src/core/lib/support/mpscq.c b/src/core/lib/support/mpscq.c new file mode 100644 index 0000000000..5b9323275a --- /dev/null +++ b/src/core/lib/support/mpscq.c @@ -0,0 +1,83 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/support/mpscq.h" + +#include <grpc/support/log.h> + +void gpr_mpscq_init(gpr_mpscq *q) { + gpr_atm_no_barrier_store(&q->head, (gpr_atm)&q->stub); + q->tail = &q->stub; + gpr_atm_no_barrier_store(&q->stub.next, (gpr_atm)NULL); +} + +void gpr_mpscq_destroy(gpr_mpscq *q) { + GPR_ASSERT(gpr_atm_no_barrier_load(&q->head) == (gpr_atm)&q->stub); + GPR_ASSERT(q->tail == &q->stub); +} + +void gpr_mpscq_push(gpr_mpscq *q, gpr_mpscq_node *n) { + gpr_atm_no_barrier_store(&n->next, (gpr_atm)NULL); + gpr_mpscq_node *prev = + (gpr_mpscq_node *)gpr_atm_full_xchg(&q->head, (gpr_atm)n); + gpr_atm_rel_store(&prev->next, (gpr_atm)n); +} + +gpr_mpscq_node *gpr_mpscq_pop(gpr_mpscq *q) { + gpr_mpscq_node *tail = q->tail; + gpr_mpscq_node *next = (gpr_mpscq_node *)gpr_atm_acq_load(&tail->next); + if (tail == &q->stub) { + // indicates the list is actually (ephemerally) empty + if (next == NULL) return NULL; + q->tail = next; + tail = next; + next = (gpr_mpscq_node *)gpr_atm_acq_load(&tail->next); + } + if (next != NULL) { + q->tail = next; + return tail; + } + gpr_mpscq_node *head = (gpr_mpscq_node *)gpr_atm_acq_load(&q->head); + if (tail != head) { + // indicates a retry is in order: we're still adding + return NULL; + } + gpr_mpscq_push(q, &q->stub); + next = (gpr_mpscq_node *)gpr_atm_acq_load(&tail->next); + if (next != NULL) { + q->tail = next; + return tail; + } + // indicates a retry is in order: we're still adding + return NULL; +} diff --git a/src/core/lib/support/mpscq.h b/src/core/lib/support/mpscq.h new file mode 100644 index 0000000000..977a117952 --- /dev/null +++ b/src/core/lib/support/mpscq.h @@ -0,0 +1,65 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_SUPPORT_MPSCQ_H +#define GRPC_CORE_LIB_SUPPORT_MPSCQ_H + +#include <grpc/support/atm.h> +#include <stddef.h> + +// Multiple-producer single-consumer lock free queue, based upon the +// implementation from Dmitry Vyukov here: +// http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue + +// List node (include this in a data structure at the top, and add application +// fields after it - to simulate inheritance) +typedef struct gpr_mpscq_node { gpr_atm next; } gpr_mpscq_node; + +// Actual queue type +typedef struct gpr_mpscq { + gpr_atm head; + // make sure head & tail don't share a cacheline + char padding[GPR_CACHELINE_SIZE]; + gpr_mpscq_node *tail; + gpr_mpscq_node stub; +} gpr_mpscq; + +void gpr_mpscq_init(gpr_mpscq *q); +void gpr_mpscq_destroy(gpr_mpscq *q); +// Push a node +void gpr_mpscq_push(gpr_mpscq *q, gpr_mpscq_node *n); +// Pop a node (returns NULL if no node is ready - which doesn't indicate that +// the queue is empty!!) +gpr_mpscq_node *gpr_mpscq_pop(gpr_mpscq *q); + +#endif /* GRPC_CORE_LIB_SUPPORT_MPSCQ_H */ diff --git a/src/core/lib/support/percent_encoding.c b/src/core/lib/support/percent_encoding.c new file mode 100644 index 0000000000..3c19f264f9 --- /dev/null +++ b/src/core/lib/support/percent_encoding.c @@ -0,0 +1,180 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/lib/support/percent_encoding.h" + +#include <grpc/support/log.h> + +const uint8_t gpr_url_percent_encoding_unreserved_bytes[256 / 8] = { + 0x00, 0x00, 0x00, 0x00, 0x00, 0x60, 0xff, 0x03, 0xfe, 0xff, 0xff, + 0x87, 0xfe, 0xff, 0xff, 0x47, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}; +const uint8_t gpr_compatible_percent_encoding_unreserved_bytes[256 / 8] = { + 0x00, 0x00, 0x00, 0x00, 0xdf, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0xff, 0xff, 0xff, 0xff, 0x7f, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}; + +static bool is_unreserved_character(uint8_t c, + const uint8_t *unreserved_bytes) { + return ((unreserved_bytes[c / 8] >> (c % 8)) & 1) != 0; +} + +gpr_slice gpr_percent_encode_slice(gpr_slice slice, + const uint8_t *unreserved_bytes) { + static const uint8_t hex[] = "0123456789ABCDEF"; + + // first pass: count the number of bytes needed to output this string + size_t output_length = 0; + const uint8_t *slice_start = GPR_SLICE_START_PTR(slice); + const uint8_t *slice_end = GPR_SLICE_END_PTR(slice); + const uint8_t *p; + bool any_reserved_bytes = false; + for (p = slice_start; p < slice_end; p++) { + bool unres = is_unreserved_character(*p, unreserved_bytes); + output_length += unres ? 1 : 3; + any_reserved_bytes |= !unres; + } + // no unreserved bytes: return the string unmodified + if (!any_reserved_bytes) { + return gpr_slice_ref(slice); + } + // second pass: actually encode + gpr_slice out = gpr_slice_malloc(output_length); + uint8_t *q = GPR_SLICE_START_PTR(out); + for (p = slice_start; p < slice_end; p++) { + if (is_unreserved_character(*p, unreserved_bytes)) { + *q++ = *p; + } else { + *q++ = '%'; + *q++ = hex[*p >> 4]; + *q++ = hex[*p & 15]; + } + } + GPR_ASSERT(q == GPR_SLICE_END_PTR(out)); + return out; +} + +static bool valid_hex(const uint8_t *p, const uint8_t *end) { + if (p >= end) return false; + return (*p >= '0' && *p <= '9') || (*p >= 'a' && *p <= 'f') || + (*p >= 'A' && *p <= 'F'); +} + +static uint8_t dehex(uint8_t c) { + if (c >= '0' && c <= '9') return (uint8_t)(c - '0'); + if (c >= 'A' && c <= 'F') return (uint8_t)(c - 'A' + 10); + if (c >= 'a' && c <= 'f') return (uint8_t)(c - 'a' + 10); + GPR_UNREACHABLE_CODE(return 255); +} + +bool gpr_strict_percent_decode_slice(gpr_slice slice_in, + const uint8_t *unreserved_bytes, + gpr_slice *slice_out) { + const uint8_t *p = GPR_SLICE_START_PTR(slice_in); + const uint8_t *in_end = GPR_SLICE_END_PTR(slice_in); + size_t out_length = 0; + bool any_percent_encoded_stuff = false; + while (p != in_end) { + if (*p == '%') { + if (!valid_hex(++p, in_end)) return false; + if (!valid_hex(++p, in_end)) return false; + p++; + out_length++; + any_percent_encoded_stuff = true; + } else if (is_unreserved_character(*p, unreserved_bytes)) { + p++; + out_length++; + } else { + return false; + } + } + if (!any_percent_encoded_stuff) { + *slice_out = gpr_slice_ref(slice_in); + return true; + } + p = GPR_SLICE_START_PTR(slice_in); + *slice_out = gpr_slice_malloc(out_length); + uint8_t *q = GPR_SLICE_START_PTR(*slice_out); + while (p != in_end) { + if (*p == '%') { + *q++ = (uint8_t)(dehex(p[1]) << 4) | (dehex(p[2])); + p += 3; + } else { + *q++ = *p++; + } + } + GPR_ASSERT(q == GPR_SLICE_END_PTR(*slice_out)); + return true; +} + +gpr_slice gpr_permissive_percent_decode_slice(gpr_slice slice_in) { + const uint8_t *p = GPR_SLICE_START_PTR(slice_in); + const uint8_t *in_end = GPR_SLICE_END_PTR(slice_in); + size_t out_length = 0; + bool any_percent_encoded_stuff = false; + while (p != in_end) { + if (*p == '%') { + if (!valid_hex(p + 1, in_end) || !valid_hex(p + 2, in_end)) { + p++; + out_length++; + } else { + p += 3; + out_length++; + any_percent_encoded_stuff = true; + } + } else { + p++; + out_length++; + } + } + if (!any_percent_encoded_stuff) { + return gpr_slice_ref(slice_in); + } + p = GPR_SLICE_START_PTR(slice_in); + gpr_slice out = gpr_slice_malloc(out_length); + uint8_t *q = GPR_SLICE_START_PTR(out); + while (p != in_end) { + if (*p == '%') { + if (!valid_hex(p + 1, in_end) || !valid_hex(p + 2, in_end)) { + *q++ = *p++; + } else { + *q++ = (uint8_t)(dehex(p[1]) << 4) | (dehex(p[2])); + p += 3; + } + } else { + *q++ = *p++; + } + } + GPR_ASSERT(q == GPR_SLICE_END_PTR(out)); + return out; +} diff --git a/src/core/lib/support/percent_encoding.h b/src/core/lib/support/percent_encoding.h new file mode 100644 index 0000000000..000bf14ede --- /dev/null +++ b/src/core/lib/support/percent_encoding.h @@ -0,0 +1,78 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_CORE_LIB_SUPPORT_PERCENT_ENCODING_H +#define GRPC_CORE_LIB_SUPPORT_PERCENT_ENCODING_H + +/* Percent encoding and decoding of slices. + Transforms arbitrary strings into safe-for-transmission strings by using + variants of percent encoding (RFC 3986). + Two major variants are supplied: one that strictly matches URL encoding, + and another which applies percent encoding only to non-http2 header + bytes (the 'compatible' variant) */ + +#include <stdbool.h> + +#include <grpc/support/slice.h> + +/* URL percent encoding spec bitfield (usabel as 'unreserved_bytes' in + gpr_percent_encode_slice, gpr_strict_percent_decode_slice). + Flags [A-Za-z0-9-_.~] as unreserved bytes for the percent encoding routines + */ +extern const uint8_t gpr_url_percent_encoding_unreserved_bytes[256 / 8]; +/* URL percent encoding spec bitfield (usabel as 'unreserved_bytes' in + gpr_percent_encode_slice, gpr_strict_percent_decode_slice). + Flags ascii7 non-control characters excluding '%' as unreserved bytes for the + percent encoding routines */ +extern const uint8_t gpr_compatible_percent_encoding_unreserved_bytes[256 / 8]; + +/* Percent-encode a slice, returning the new slice (this cannot fail): + unreserved_bytes is a bitfield indicating which bytes are considered + unreserved and thus do not need percent encoding */ +gpr_slice gpr_percent_encode_slice(gpr_slice slice, + const uint8_t *unreserved_bytes); +/* Percent-decode a slice, strictly. + If the input is legal (contains no unreserved bytes, and legal % encodings), + returns true and sets *slice_out to the decoded slice. + If the input is not legal, returns false and leaves *slice_out untouched. + unreserved_bytes is a bitfield indicating which bytes are considered + unreserved and thus do not need percent encoding */ +bool gpr_strict_percent_decode_slice(gpr_slice slice_in, + const uint8_t *unreserved_bytes, + gpr_slice *slice_out); +/* Percent-decode a slice, permissively. + If a % triplet can not be decoded, pass it through verbatim. + This cannot fail. */ +gpr_slice gpr_permissive_percent_decode_slice(gpr_slice slice_in); + +#endif /* GRPC_CORE_LIB_SUPPORT_PERCENT_ENCODING_H */ diff --git a/src/core/lib/support/time_posix.c b/src/core/lib/support/time_posix.c index 11542072fe..a69c501e9f 100644 --- a/src/core/lib/support/time_posix.c +++ b/src/core/lib/support/time_posix.c @@ -127,7 +127,7 @@ static gpr_timespec now_impl(gpr_clock_type clock) { now.tv_nsec = now_tv.tv_usec * 1000; break; case GPR_CLOCK_MONOTONIC: - now_dbl = (mach_absolute_time() - g_time_start) * g_time_scale; + now_dbl = ((double)(mach_absolute_time() - g_time_start)) * g_time_scale; now.tv_sec = (int64_t)(now_dbl * 1e-9); now.tv_nsec = (int32_t)(now_dbl - ((double)now.tv_sec) * 1e9); break; diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index fc9df76dc1..5690bcab1e 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -109,6 +109,10 @@ typedef struct batch_control { uint8_t recv_message; uint8_t recv_final_op; uint8_t is_notify_tag_closure; + + /* TODO(ctiller): now that this is inlined, figure out how much of the above + state can be eliminated */ + grpc_transport_stream_op op; } batch_control; struct grpc_call { @@ -122,8 +126,6 @@ struct grpc_call { /* client or server call */ bool is_client; - /* is the alarm set */ - bool have_alarm; /** has grpc_call_destroy been called */ bool destroy_called; /** flag indicating that cancellation is inherited */ @@ -154,8 +156,9 @@ struct grpc_call { /* Received call statuses from various sources */ received_status status[STATUS_SOURCE_COUNT]; - /* Call stats: only valid after trailing metadata received */ - grpc_call_stats stats; + /* Call data useful used for reporting. Only valid after the call has + * completed */ + grpc_call_final_info final_info; /* Compression algorithm for *incoming* data */ grpc_compression_algorithm incoming_compression_algorithm; @@ -165,9 +168,6 @@ struct grpc_call { /* Contexts for various subsystems (security, tracing, ...). */ grpc_call_context_element context[GRPC_CONTEXT_COUNT]; - /* Deadline alarm - if have_alarm is non-zero */ - grpc_timer alarm; - /* for the client, extra metadata is initial metadata; for the server, it's trailing metadata */ grpc_linked_mdelem send_extra_metadata[MAX_SEND_EXTRA_METADATA_COUNT]; @@ -210,8 +210,6 @@ struct grpc_call { #define CALL_FROM_TOP_ELEM(top_elem) \ CALL_FROM_CALL_STACK(grpc_call_stack_from_top_element(top_elem)) -static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call, - gpr_timespec deadline); static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call, grpc_transport_stream_op *op); static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c, @@ -259,29 +257,8 @@ grpc_call *grpc_call_create( call->metadata_batch[i][j].deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); } } - call->send_deadline = - gpr_convert_clock_type(send_deadline, GPR_CLOCK_MONOTONIC); - GRPC_CHANNEL_INTERNAL_REF(channel, "call"); - /* initial refcount dropped by grpc_call_destroy */ - grpc_call_stack_init(&exec_ctx, channel_stack, 1, destroy_call, call, - call->context, server_transport_data, - CALL_STACK_FROM_CALL(call)); - if (cq != NULL) { - GPR_ASSERT( - pollset_set_alternative == NULL && - "Only one of 'cq' and 'pollset_set_alternative' should be non-NULL."); - GRPC_CQ_INTERNAL_REF(cq, "bind"); - call->pollent = - grpc_polling_entity_create_from_pollset(grpc_cq_pollset(cq)); - } - if (pollset_set_alternative != NULL) { - call->pollent = - grpc_polling_entity_create_from_pollset_set(pollset_set_alternative); - } - if (!grpc_polling_entity_is_empty(&call->pollent)) { - grpc_call_stack_set_pollset_or_pollset_set( - &exec_ctx, CALL_STACK_FROM_CALL(call), &call->pollent); - } + send_deadline = gpr_convert_clock_type(send_deadline, GPR_CLOCK_MONOTONIC); + if (parent_call != NULL) { GRPC_CALL_INTERNAL_REF(parent_call, "child"); GPR_ASSERT(call->is_client); @@ -323,10 +300,38 @@ grpc_call *grpc_call_create( gpr_mu_unlock(&parent_call->mu); } - if (gpr_time_cmp(send_deadline, gpr_inf_future(send_deadline.clock_type)) != - 0) { - set_deadline_alarm(&exec_ctx, call, send_deadline); + + call->send_deadline = send_deadline; + + GRPC_CHANNEL_INTERNAL_REF(channel, "call"); + /* initial refcount dropped by grpc_call_destroy */ + grpc_error *error = grpc_call_stack_init( + &exec_ctx, channel_stack, 1, destroy_call, call, call->context, + server_transport_data, send_deadline, CALL_STACK_FROM_CALL(call)); + if (error != GRPC_ERROR_NONE) { + grpc_status_code status; + const char *error_str; + grpc_error_get_status(error, &status, &error_str); + close_with_status(&exec_ctx, call, status, error_str); + GRPC_ERROR_UNREF(error); + } + if (cq != NULL) { + GPR_ASSERT( + pollset_set_alternative == NULL && + "Only one of 'cq' and 'pollset_set_alternative' should be non-NULL."); + GRPC_CQ_INTERNAL_REF(cq, "bind"); + call->pollent = + grpc_polling_entity_create_from_pollset(grpc_cq_pollset(cq)); + } + if (pollset_set_alternative != NULL) { + call->pollent = + grpc_polling_entity_create_from_pollset_set(pollset_set_alternative); + } + if (!grpc_polling_entity_is_empty(&call->pollent)) { + grpc_call_stack_set_pollset_or_pollset_set( + &exec_ctx, CALL_STACK_FROM_CALL(call), &call->pollent); } + grpc_exec_ctx_finish(&exec_ctx); GPR_TIMER_END("grpc_call_create", 0); return call; @@ -361,6 +366,25 @@ 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 get_final_status(grpc_call *call, + void (*set_value)(grpc_status_code code, + void *user_data), + void *set_value_user_data) { + int i; + for (i = 0; i < STATUS_SOURCE_COUNT; i++) { + if (call->status[i].is_set) { + set_value(call->status[i].code, set_value_user_data); + return; + } + } + if (call->is_client) { + set_value(GRPC_STATUS_UNKNOWN, set_value_user_data); + } else { + set_value(GRPC_STATUS_OK, set_value_user_data); + } +} + +static void set_status_value_directly(grpc_status_code status, void *dest); static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, grpc_error *error) { size_t i; @@ -392,7 +416,11 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, GRPC_CQ_INTERNAL_UNREF(c->cq, "bind"); } grpc_channel *channel = c->channel; - grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c), &c->stats, c); + + get_final_status(call, set_status_value_directly, + &c->final_info.final_status); + + grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c), &c->final_info, c); GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel, "call"); GPR_TIMER_END("destroy_call", 0); } @@ -414,40 +442,13 @@ static void set_status_details(grpc_call *call, status_source source, } } -static void get_final_status(grpc_call *call, - void (*set_value)(grpc_status_code code, - void *user_data), - void *set_value_user_data) { - int i; - for (i = 0; i < STATUS_SOURCE_COUNT; i++) { - if (call->status[i].is_set) { - set_value(call->status[i].code, set_value_user_data); - return; - } - } - if (call->is_client) { - set_value(GRPC_STATUS_UNKNOWN, set_value_user_data); - } else { - set_value(GRPC_STATUS_OK, set_value_user_data); - } -} - static void set_status_from_error(grpc_call *call, status_source source, grpc_error *error) { - intptr_t status; - if (grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status)) { - set_status_code(call, source, (uint32_t)status); - } else { - set_status_code(call, source, GRPC_STATUS_INTERNAL); - } - const char *msg = grpc_error_get_str(error, GRPC_ERROR_STR_GRPC_MESSAGE); - bool free_msg = false; - if (msg == NULL) { - free_msg = true; - msg = grpc_error_string(error); - } + grpc_status_code status; + const char *msg; + grpc_error_get_status(error, &status, &msg); + set_status_code(call, source, (uint32_t)status); set_status_details(call, source, grpc_mdstr_from_string(msg)); - if (free_msg) grpc_error_free_string(msg); } static void set_incoming_compression_algorithm( @@ -720,9 +721,6 @@ void grpc_call_destroy(grpc_call *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->received_final_op; gpr_mu_unlock(&c->mu); if (cancel) grpc_call_cancel(c, NULL); @@ -760,8 +758,8 @@ typedef struct termination_closure { grpc_closure closure; grpc_call *call; grpc_error *error; - grpc_closure *op_closure; enum { TC_CANCEL, TC_CLOSE } type; + grpc_transport_stream_op op; } termination_closure; static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp, @@ -776,31 +774,27 @@ static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp, break; } GRPC_ERROR_UNREF(tc->error); - grpc_exec_ctx_sched(exec_ctx, tc->op_closure, GRPC_ERROR_NONE, NULL); gpr_free(tc); } static void send_cancel(grpc_exec_ctx *exec_ctx, void *tcp, grpc_error *error) { - grpc_transport_stream_op op; termination_closure *tc = tcp; - memset(&op, 0, sizeof(op)); - op.cancel_error = tc->error; + memset(&tc->op, 0, sizeof(tc->op)); + tc->op.cancel_error = tc->error; /* reuse closure to catch completion */ grpc_closure_init(&tc->closure, done_termination, tc); - op.on_complete = &tc->closure; - execute_op(exec_ctx, tc->call, &op); + tc->op.on_complete = &tc->closure; + execute_op(exec_ctx, tc->call, &tc->op); } static void send_close(grpc_exec_ctx *exec_ctx, void *tcp, grpc_error *error) { - grpc_transport_stream_op op; termination_closure *tc = tcp; - memset(&op, 0, sizeof(op)); - op.close_error = tc->error; + memset(&tc->op, 0, sizeof(tc->op)); + tc->op.close_error = tc->error; /* reuse closure to catch completion */ grpc_closure_init(&tc->closure, done_termination, tc); - tc->op_closure = op.on_complete; - op.on_complete = &tc->closure; - execute_op(exec_ctx, tc->call, &op); + tc->op.on_complete = &tc->closure; + execute_op(exec_ctx, tc->call, &tc->op); } static grpc_call_error terminate_with_status(grpc_exec_ctx *exec_ctx, @@ -881,32 +875,6 @@ grpc_call *grpc_call_from_top_element(grpc_call_element *elem) { return CALL_FROM_TOP_ELEM(elem); } -static void call_alarm(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_call *call = arg; - gpr_mu_lock(&call->mu); - call->have_alarm = 0; - if (error != GRPC_ERROR_CANCELLED) { - cancel_with_status(exec_ctx, call, GRPC_STATUS_DEADLINE_EXCEEDED, - "Deadline Exceeded"); - } - gpr_mu_unlock(&call->mu); - GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "alarm"); -} - -static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call, - gpr_timespec deadline) { - if (call->have_alarm) { - gpr_log(GPR_ERROR, "Attempt to set deadline alarm twice"); - assert(0); - return; - } - GRPC_CALL_INTERNAL_REF(call, "alarm"); - call->have_alarm = 1; - call->send_deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); - grpc_timer_init(exec_ctx, &call->alarm, call->send_deadline, call_alarm, call, - gpr_now(GPR_CLOCK_MONOTONIC)); -} - /* we offset status by a small amount when storing it into transport metadata as metadata cannot store a 0 value (which is used as OK for grpc_status_codes */ @@ -1144,17 +1112,6 @@ static void process_data_after_md(grpc_exec_ctx *exec_ctx, batch_control *bctl, if (gpr_unref(&bctl->steps_to_complete)) { post_batch_completion(exec_ctx, bctl); } - } else if (call->receiving_stream->length > - grpc_channel_get_max_message_length(call->channel)) { - cancel_with_status(exec_ctx, call, GRPC_STATUS_INTERNAL, - "Max message size exceeded"); - grpc_byte_stream_destroy(exec_ctx, call->receiving_stream); - call->receiving_stream = NULL; - *call->receiving_buffer = NULL; - call->receiving_message = 0; - if (gpr_unref(&bctl->steps_to_complete)) { - post_batch_completion(exec_ctx, bctl); - } } else { call->test_only_last_message_flags = call->receiving_stream->flags; if ((call->receiving_stream->flags & GRPC_WRITE_INTERNAL_COMPRESS) && @@ -1259,9 +1216,8 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx, 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); + call->send_deadline = + gpr_convert_clock_type(md->deadline, GPR_CLOCK_MONOTONIC); } } @@ -1290,9 +1246,17 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, GRPC_ERROR_REF(error); gpr_mu_lock(&call->mu); + + // If the error has an associated status code, set the call's status. + intptr_t status; + if (error != GRPC_ERROR_NONE && + grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status)) { + set_status_from_error(call, STATUS_FROM_CORE, error); + } + if (bctl->send_initial_metadata) { if (error != GRPC_ERROR_NONE) { - set_status_code(call, STATUS_FROM_CORE, GRPC_STATUS_UNAVAILABLE); + set_status_from_error(call, STATUS_FROM_CORE, error); } grpc_metadata_batch_destroy( &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]); @@ -1310,9 +1274,6 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, grpc_metadata_batch_filter(md, recv_trailing_filter, call); call->received_final_op = true; - 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) { @@ -1354,19 +1315,19 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, grpc_call *call, const grpc_op *ops, size_t nops, void *notify_tag, int is_notify_tag_closure) { - grpc_transport_stream_op stream_op; size_t i; const grpc_op *op; batch_control *bctl; int num_completion_callbacks_needed = 1; grpc_call_error error = GRPC_CALL_OK; + // sent_initial_metadata guards against variable reuse. + grpc_metadata compression_md; + GPR_TIMER_BEGIN("grpc_call_start_batch", 0); GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, notify_tag); - memset(&stream_op, 0, sizeof(stream_op)); - /* TODO(ctiller): this feels like it could be made lock-free */ gpr_mu_lock(&call->mu); bctl = allocate_batch_control(call); @@ -1375,6 +1336,9 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, bctl->notify_tag = notify_tag; bctl->is_notify_tag_closure = (uint8_t)(is_notify_tag_closure != 0); + grpc_transport_stream_op *stream_op = &bctl->op; + memset(stream_op, 0, sizeof(*stream_op)); + if (nops == 0) { GRPC_CALL_INTERNAL_REF(call, "completion"); bctl->error = GRPC_ERROR_NONE; @@ -1406,8 +1370,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, goto done_with_error; } /* process compression level */ - grpc_metadata compression_md; - memset(&compression_md, 0, sizeof(grpc_metadata)); + memset(&compression_md, 0, sizeof(compression_md)); size_t additional_metadata_count = 0; grpc_compression_level effective_compression_level; bool level_set = false; @@ -1453,9 +1416,9 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, } /* TODO(ctiller): just make these the same variable? */ call->metadata_batch[0][0].deadline = call->send_deadline; - stream_op.send_initial_metadata = + stream_op->send_initial_metadata = &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]; - stream_op.send_initial_metadata_flags = op->flags; + stream_op->send_initial_metadata_flags = op->flags; break; case GRPC_OP_SEND_MESSAGE: if (!are_write_flags_valid(op->flags)) { @@ -1475,7 +1438,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, 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; + stream_op->send_message = &call->sending_stream.base; break; case GRPC_OP_SEND_CLOSE_FROM_CLIENT: /* Flag validation: currently allow no flags */ @@ -1493,7 +1456,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, } bctl->send_final_op = 1; call->sent_final_op = 1; - stream_op.send_trailing_metadata = + stream_op->send_trailing_metadata = &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]; break; case GRPC_OP_SEND_STATUS_FROM_SERVER: @@ -1540,7 +1503,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, error = GRPC_CALL_ERROR_INVALID_METADATA; goto done_with_error; } - stream_op.send_trailing_metadata = + stream_op->send_trailing_metadata = &call->metadata_batch[0 /* is_receiving */][1 /* is_trailing */]; break; case GRPC_OP_RECV_INITIAL_METADATA: @@ -1558,9 +1521,9 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, grpc_closure_init(&call->receiving_initial_metadata_ready, receiving_initial_metadata_ready, bctl); bctl->recv_initial_metadata = 1; - stream_op.recv_initial_metadata = + stream_op->recv_initial_metadata = &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */]; - stream_op.recv_initial_metadata_ready = + stream_op->recv_initial_metadata_ready = &call->receiving_initial_metadata_ready; num_completion_callbacks_needed++; break; @@ -1577,10 +1540,10 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, call->receiving_message = 1; bctl->recv_message = 1; call->receiving_buffer = op->data.recv_message; - stream_op.recv_message = &call->receiving_stream; + 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; + stream_op->recv_message_ready = &call->receiving_stream_ready; num_completion_callbacks_needed++; break; case GRPC_OP_RECV_STATUS_ON_CLIENT: @@ -1606,9 +1569,10 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, call->final_op.client.status_details_capacity = op->data.recv_status_on_client.status_details_capacity; bctl->recv_final_op = 1; - stream_op.recv_trailing_metadata = + stream_op->recv_trailing_metadata = &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; - stream_op.collect_stats = &call->stats.transport_stream_stats; + stream_op->collect_stats = + &call->final_info.stats.transport_stream_stats; break; case GRPC_OP_RECV_CLOSE_ON_SERVER: /* Flag validation: currently allow no flags */ @@ -1628,9 +1592,10 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, call->final_op.server.cancelled = op->data.recv_close_on_server.cancelled; bctl->recv_final_op = 1; - stream_op.recv_trailing_metadata = + stream_op->recv_trailing_metadata = &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */]; - stream_op.collect_stats = &call->stats.transport_stream_stats; + stream_op->collect_stats = + &call->final_info.stats.transport_stream_stats; break; } } @@ -1641,12 +1606,12 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, } gpr_ref_init(&bctl->steps_to_complete, num_completion_callbacks_needed); - stream_op.context = call->context; + stream_op->context = call->context; grpc_closure_init(&bctl->finish_batch, finish_batch, bctl); - stream_op.on_complete = &bctl->finish_batch; + stream_op->on_complete = &bctl->finish_batch; gpr_mu_unlock(&call->mu); - execute_op(exec_ctx, call, &stream_op); + execute_op(exec_ctx, call, stream_op); done: GPR_TIMER_END("grpc_call_start_batch", 0); diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c index 6d2b1c4935..6adb70a987 100644 --- a/src/core/lib/surface/channel.c +++ b/src/core/lib/surface/channel.c @@ -64,7 +64,6 @@ typedef struct registered_call { struct grpc_channel { int is_client; - uint32_t max_message_length; grpc_compression_options compression_options; grpc_mdelem *default_authority; @@ -80,9 +79,6 @@ struct grpc_channel { #define CHANNEL_FROM_TOP_ELEM(top_elem) \ CHANNEL_FROM_CHANNEL_STACK(grpc_channel_stack_from_top_element(top_elem)) -/* the protobuf library will (by default) start warning at 100megs */ -#define DEFAULT_MAX_MESSAGE_LENGTH (4 * 1024 * 1024) - static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error); @@ -114,21 +110,10 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, gpr_mu_init(&channel->registered_call_mu); channel->registered_calls = NULL; - channel->max_message_length = DEFAULT_MAX_MESSAGE_LENGTH; grpc_compression_options_init(&channel->compression_options); if (args) { for (size_t i = 0; i < args->num_args; i++) { - if (0 == strcmp(args->args[i].key, GRPC_ARG_MAX_MESSAGE_LENGTH)) { - if (args->args[i].type != GRPC_ARG_INTEGER) { - gpr_log(GPR_ERROR, "%s ignored: it must be an integer", - GRPC_ARG_MAX_MESSAGE_LENGTH); - } else if (args->args[i].value.integer < 0) { - gpr_log(GPR_ERROR, "%s ignored: it must be >= 0", - GRPC_ARG_MAX_MESSAGE_LENGTH); - } else { - channel->max_message_length = (uint32_t)args->args[i].value.integer; - } - } else if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) { if (args->args[i].type != GRPC_ARG_STRING) { gpr_log(GPR_ERROR, "%s ignored: it must be a string", GRPC_ARG_DEFAULT_AUTHORITY); @@ -334,14 +319,13 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, } void grpc_channel_destroy(grpc_channel *channel) { - grpc_transport_op op; + grpc_transport_op *op = grpc_make_transport_op(NULL); grpc_channel_element *elem; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; GRPC_API_TRACE("grpc_channel_destroy(channel=%p)", 1, (channel)); - memset(&op, 0, sizeof(op)); - op.disconnect_with_error = GRPC_ERROR_CREATE("Channel Destroyed"); + op->disconnect_with_error = GRPC_ERROR_CREATE("Channel Destroyed"); elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CHANNEL(channel), 0); - elem->filter->start_transport_op(&exec_ctx, elem, &op); + elem->filter->start_transport_op(&exec_ctx, elem, op); GRPC_CHANNEL_INTERNAL_UNREF(&exec_ctx, channel, "channel"); @@ -371,7 +355,3 @@ grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int i) { return grpc_mdelem_from_metadata_strings(GRPC_MDSTR_GRPC_STATUS, grpc_mdstr_from_string(tmp)); } - -uint32_t grpc_channel_get_max_message_length(grpc_channel *channel) { - return channel->max_message_length; -} diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h index 7eff7b8883..23cc8656ca 100644 --- a/src/core/lib/surface/channel.h +++ b/src/core/lib/surface/channel.h @@ -42,6 +42,14 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target, grpc_channel_stack_type channel_stack_type, grpc_transport *optional_transport); +/** Create a call given a grpc_channel, in order to call \a method. + Progress is tied to activity on \a pollset_set. The returned call object is + meant to be used with \a grpc_call_start_batch_and_execute, which relies on + callbacks to signal completions. \a method and \a host need + only live through the invocation of this function. If \a parent_call is + non-NULL, it must be a server-side call. It will be used to propagate + properties from the server call to this new client call, depending on the + value of \a propagation_mask (see propagation_bits.h for possible values) */ grpc_call *grpc_channel_create_pollset_set_call( grpc_channel *channel, grpc_call *parent_call, uint32_t propagation_mask, grpc_pollset_set *pollset_set, const char *method, const char *host, @@ -56,7 +64,6 @@ grpc_channel_stack *grpc_channel_get_channel_stack(grpc_channel *channel); The returned elem is owned by the caller. */ grpc_mdelem *grpc_channel_get_reffed_status_elem(grpc_channel *channel, int status_code); -uint32_t grpc_channel_get_max_message_length(grpc_channel *channel); #ifdef GRPC_STREAM_REFCOUNT_DEBUG void grpc_channel_internal_ref(grpc_channel *channel, const char *reason); diff --git a/src/core/lib/surface/channel_init.h b/src/core/lib/surface/channel_init.h index 3a18a61ddb..b53f2aefb9 100644 --- a/src/core/lib/surface/channel_init.h +++ b/src/core/lib/surface/channel_init.h @@ -40,6 +40,10 @@ #define GRPC_CHANNEL_INIT_BUILTIN_PRIORITY 10000 +#ifdef __cplusplus +extern "C" { +#endif + /// This module provides a way for plugins (and the grpc core library itself) /// to register mutators for channel stacks. /// It also provides a universal entry path to run those mutators to build @@ -84,4 +88,8 @@ bool grpc_channel_init_create_stack(grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder, grpc_channel_stack_type type); +#ifdef __cplusplus +} +#endif + #endif /* GRPC_CORE_LIB_SURFACE_CHANNEL_INIT_H */ diff --git a/src/core/lib/surface/channel_ping.c b/src/core/lib/surface/channel_ping.c index 9818f9d2f2..0d2f01a649 100644 --- a/src/core/lib/surface/channel_ping.c +++ b/src/core/lib/surface/channel_ping.c @@ -61,19 +61,20 @@ static void ping_done(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { void grpc_channel_ping(grpc_channel *channel, grpc_completion_queue *cq, void *tag, void *reserved) { - grpc_transport_op op; + GRPC_API_TRACE("grpc_channel_ping(channel=%p, cq=%p, tag=%p, reserved=%p)", 4, + (channel, cq, tag, reserved)); + grpc_transport_op *op = grpc_make_transport_op(NULL); 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); + 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); + top_elem->filter->start_transport_op(&exec_ctx, top_elem, op); grpc_exec_ctx_finish(&exec_ctx); } diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index 3049284f68..4dbf3aae63 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -57,6 +57,8 @@ typedef struct grpc_cq_completion { uintptr_t next; } grpc_cq_completion; +//#define GRPC_CQ_REF_COUNT_DEBUG + #ifdef GRPC_CQ_REF_COUNT_DEBUG void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason, const char *file, int line); diff --git a/src/core/lib/surface/init.c b/src/core/lib/surface/init.c index 5397913a21..289f4ce8e8 100644 --- a/src/core/lib/surface/init.c +++ b/src/core/lib/surface/init.c @@ -43,10 +43,13 @@ #include "src/core/lib/channel/channel_stack.h" #include "src/core/lib/channel/compress_filter.h" #include "src/core/lib/channel/connected_channel.h" +#include "src/core/lib/channel/deadline_filter.h" #include "src/core/lib/channel/http_client_filter.h" #include "src/core/lib/channel/http_server_filter.h" +#include "src/core/lib/channel/message_size_filter.h" #include "src/core/lib/debug/trace.h" #include "src/core/lib/http/parser.h" +#include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/profiling/timers.h" @@ -98,6 +101,21 @@ static bool maybe_add_http_filter(grpc_channel_stack_builder *builder, static void register_builtin_channel_init() { grpc_channel_init_register_stage( + GRPC_CLIENT_DIRECT_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, + prepend_filter, (void *)&grpc_client_deadline_filter); + grpc_channel_init_register_stage( + GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, prepend_filter, + (void *)&grpc_server_deadline_filter); + grpc_channel_init_register_stage( + GRPC_CLIENT_SUBCHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, + prepend_filter, (void *)&grpc_message_size_filter); + grpc_channel_init_register_stage( + GRPC_CLIENT_DIRECT_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, + prepend_filter, (void *)&grpc_message_size_filter); + grpc_channel_init_register_stage( + GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, prepend_filter, + (void *)&grpc_message_size_filter); + grpc_channel_init_register_stage( GRPC_CLIENT_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, prepend_filter, (void *)&grpc_compress_filter); grpc_channel_init_register_stage( @@ -165,6 +183,7 @@ void grpc_init(void) { grpc_register_tracer("http1", &grpc_http1_trace); grpc_register_tracer("compression", &grpc_compression_trace); grpc_register_tracer("queue_pluck", &grpc_cq_pluck_trace); + grpc_register_tracer("combiner", &grpc_combiner_trace); // Default pluck trace to 1 grpc_cq_pluck_trace = 1; grpc_register_tracer("queue_timeout", &grpc_cq_event_timeout_trace); diff --git a/src/core/lib/surface/lame_client.c b/src/core/lib/surface/lame_client.c index 5ea4cba5d1..d32c884e8e 100644 --- a/src/core/lib/surface/lame_client.c +++ b/src/core/lib/surface/lame_client.c @@ -97,21 +97,24 @@ static void lame_start_transport_op(grpc_exec_ctx *exec_ctx, grpc_exec_ctx_sched(exec_ctx, op->on_connectivity_state_change, GRPC_ERROR_NONE, NULL); } - if (op->on_consumed != NULL) { - grpc_exec_ctx_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE, NULL); - } if (op->send_ping != NULL) { grpc_exec_ctx_sched(exec_ctx, op->send_ping, GRPC_ERROR_CREATE("lame client channel"), NULL); } GRPC_ERROR_UNREF(op->disconnect_with_error); + if (op->on_consumed != NULL) { + grpc_exec_ctx_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE, NULL); + } } -static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) {} +static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { + return GRPC_ERROR_NONE; +} static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, + const grpc_call_final_info *final_info, void *and_free_memory) { gpr_free(and_free_memory); } diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c index 2f108af48a..7300d79b9f 100644 --- a/src/core/lib/surface/server.c +++ b/src/core/lib/surface/server.c @@ -149,6 +149,7 @@ struct call_data { grpc_metadata_batch *recv_initial_metadata; bool recv_idempotent_request; + bool recv_cacheable_request; grpc_metadata_array initial_metadata; request_matcher *request_matcher; @@ -273,27 +274,24 @@ static void shutdown_cleanup(grpc_exec_ctx *exec_ctx, void *arg, static void send_shutdown(grpc_exec_ctx *exec_ctx, grpc_channel *channel, int send_goaway, grpc_error *send_disconnect) { - grpc_transport_op op; - struct shutdown_cleanup_args *sc; + struct shutdown_cleanup_args *sc = gpr_malloc(sizeof(*sc)); + grpc_closure_init(&sc->closure, shutdown_cleanup, sc); + grpc_transport_op *op = grpc_make_transport_op(&sc->closure); grpc_channel_element *elem; - memset(&op, 0, sizeof(op)); - op.send_goaway = send_goaway; - sc = gpr_malloc(sizeof(*sc)); + op->send_goaway = send_goaway; sc->slice = gpr_slice_from_copied_string("Server shutdown"); - op.goaway_message = &sc->slice; - op.goaway_status = GRPC_STATUS_OK; - op.disconnect_with_error = send_disconnect; - grpc_closure_init(&sc->closure, shutdown_cleanup, sc); - op.on_consumed = &sc->closure; + op->goaway_message = &sc->slice; + op->goaway_status = GRPC_STATUS_OK; + op->disconnect_with_error = send_disconnect; elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0); - elem->filter->start_transport_op(exec_ctx, elem, &op); + elem->filter->start_transport_op(exec_ctx, elem, op); } static void channel_broadcaster_shutdown(grpc_exec_ctx *exec_ctx, channel_broadcaster *cb, - int send_goaway, + bool send_goaway, grpc_error *force_disconnect) { size_t i; @@ -431,7 +429,8 @@ static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd, server_unref(exec_ctx, server); } -static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) { +static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand, + grpc_error *error) { if (is_channel_orphaned(chand)) return; GPR_ASSERT(chand->server != NULL); orphan_channel(chand); @@ -440,14 +439,20 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) { chand->finish_destroy_channel_closure.cb = finish_destroy_channel; chand->finish_destroy_channel_closure.cb_arg = chand; - grpc_transport_op op; - memset(&op, 0, sizeof(op)); - op.set_accept_stream = true; - op.on_consumed = &chand->finish_destroy_channel_closure; + grpc_transport_op *op = + grpc_make_transport_op(&chand->finish_destroy_channel_closure); + op->set_accept_stream = true; grpc_channel_next_op(exec_ctx, grpc_channel_stack_element( grpc_channel_get_channel_stack(chand->channel), 0), - &op); + op); + + if (error != GRPC_ERROR_NONE) { + const char *msg = grpc_error_string(error); + gpr_log(GPR_INFO, "Disconnected client: %s", msg); + grpc_error_free_string(msg); + } + GRPC_ERROR_UNREF(error); } static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) { @@ -497,9 +502,12 @@ static void publish_call(grpc_exec_ctx *exec_ctx, grpc_server *server, &rc->data.batch.details->method_capacity, calld->path); rc->data.batch.details->deadline = calld->deadline; rc->data.batch.details->flags = - 0 | (calld->recv_idempotent_request - ? GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST - : 0); + (calld->recv_idempotent_request + ? GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST + : 0) | + (calld->recv_cacheable_request + ? GRPC_INITIAL_METADATA_CACHEABLE_REQUEST + : 0); break; case REGISTERED_CALL: *rc->data.registered.deadline = calld->deadline; @@ -779,6 +787,7 @@ static void server_mutate_op(grpc_call_element *elem, calld->on_done_recv_initial_metadata = op->recv_initial_metadata_ready; op->recv_initial_metadata_ready = &calld->server_on_recv_initial_metadata; op->recv_idempotent_request = &calld->recv_idempotent_request; + op->recv_cacheable_request = &calld->recv_cacheable_request; } } @@ -840,24 +849,24 @@ static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd, channel_data *chand = cd; grpc_server *server = chand->server; if (chand->connectivity_state != GRPC_CHANNEL_SHUTDOWN) { - grpc_transport_op op; - memset(&op, 0, sizeof(op)); - op.on_connectivity_state_change = &chand->channel_connectivity_changed, - op.connectivity_state = &chand->connectivity_state; + grpc_transport_op *op = grpc_make_transport_op(NULL); + op->on_connectivity_state_change = &chand->channel_connectivity_changed, + op->connectivity_state = &chand->connectivity_state; grpc_channel_next_op(exec_ctx, grpc_channel_stack_element( grpc_channel_get_channel_stack(chand->channel), 0), - &op); + op); } else { gpr_mu_lock(&server->mu_global); - destroy_channel(exec_ctx, chand); + destroy_channel(exec_ctx, chand, GRPC_ERROR_REF(error)); gpr_mu_unlock(&server->mu_global); GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "connectivity"); } } -static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - grpc_call_element_args *args) { +static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; memset(calld, 0, sizeof(call_data)); @@ -869,10 +878,12 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, server_on_recv_initial_metadata, elem); server_ref(chand->server); + return GRPC_ERROR_NONE; } static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_stats *stats, void *ignored) { + const grpc_call_final_info *final_info, + void *ignored) { channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; @@ -1095,6 +1106,12 @@ void grpc_server_start(grpc_server *server) { grpc_exec_ctx_finish(&exec_ctx); } +void grpc_server_get_pollsets(grpc_server *server, grpc_pollset ***pollsets, + size_t *pollset_count) { + *pollset_count = server->cq_count; + *pollsets = server->pollsets; +} + void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, grpc_transport *transport, grpc_pollset *accepting_pollset, @@ -1111,7 +1128,7 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, size_t slots; uint32_t probes; uint32_t max_probes = 0; - grpc_transport_op op; + grpc_transport_op *op = NULL; channel = grpc_channel_create(exec_ctx, NULL, args, GRPC_SERVER_CHANNEL, transport); @@ -1171,16 +1188,16 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s, gpr_mu_unlock(&s->mu_global); GRPC_CHANNEL_INTERNAL_REF(channel, "connectivity"); - memset(&op, 0, sizeof(op)); - op.set_accept_stream = true; - op.set_accept_stream_fn = accept_stream; - op.set_accept_stream_user_data = chand; - op.on_connectivity_state_change = &chand->channel_connectivity_changed; - op.connectivity_state = &chand->connectivity_state; + op = grpc_make_transport_op(NULL); + op->set_accept_stream = true; + op->set_accept_stream_fn = accept_stream; + op->set_accept_stream_user_data = chand; + op->on_connectivity_state_change = &chand->channel_connectivity_changed; + op->connectivity_state = &chand->connectivity_state; if (gpr_atm_acq_load(&s->shutdown_flag) != 0) { - op.disconnect_with_error = GRPC_ERROR_CREATE("Server shutdown"); + op->disconnect_with_error = GRPC_ERROR_CREATE("Server shutdown"); } - grpc_transport_perform_op(exec_ctx, transport, &op); + grpc_transport_perform_op(exec_ctx, transport, op); } void done_published_shutdown(grpc_exec_ctx *exec_ctx, void *done_arg, @@ -1249,7 +1266,8 @@ void grpc_server_shutdown_and_notify(grpc_server *server, l->destroy(&exec_ctx, server, l->arg, &l->destroy_done); } - channel_broadcaster_shutdown(&exec_ctx, &broadcaster, 1, 0); + channel_broadcaster_shutdown(&exec_ctx, &broadcaster, true /* send_goaway */, + GRPC_ERROR_NONE); done: grpc_exec_ctx_finish(&exec_ctx); @@ -1265,7 +1283,7 @@ void grpc_server_cancel_all_calls(grpc_server *server) { channel_broadcaster_init(server, &broadcaster); gpr_mu_unlock(&server->mu_global); - channel_broadcaster_shutdown(&exec_ctx, &broadcaster, 0, + channel_broadcaster_shutdown(&exec_ctx, &broadcaster, false /* send_goaway */, GRPC_ERROR_CREATE("Cancelling all calls")); grpc_exec_ctx_finish(&exec_ctx); } diff --git a/src/core/lib/surface/server.h b/src/core/lib/surface/server.h index fb6e4d60c5..551a40a4ff 100644 --- a/src/core/lib/surface/server.h +++ b/src/core/lib/surface/server.h @@ -60,4 +60,9 @@ const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server); int grpc_server_has_open_connections(grpc_server *server); +/* Do not call this before grpc_server_start. Returns the pollsets and the + * number of pollsets via 'pollsets' and 'pollset_count'. */ +void grpc_server_get_pollsets(grpc_server *server, grpc_pollset ***pollsets, + size_t *pollset_count); + #endif /* GRPC_CORE_LIB_SURFACE_SERVER_H */ diff --git a/src/core/lib/surface/version.c b/src/core/lib/surface/version.c index 1942075054..41242684da 100644 --- a/src/core/lib/surface/version.c +++ b/src/core/lib/surface/version.c @@ -37,3 +37,5 @@ #include <grpc/grpc.h> const char *grpc_version_string(void) { return "1.1.0-dev"; } + +const char *grpc_g_stands_for(void) { return "good"; } diff --git a/src/core/lib/transport/byte_stream.h b/src/core/lib/transport/byte_stream.h index 95519a9eaf..e64dce6283 100644 --- a/src/core/lib/transport/byte_stream.h +++ b/src/core/lib/transport/byte_stream.h @@ -59,13 +59,9 @@ struct grpc_byte_stream { * on_complete will not be called), 0 if the bytes will be available * asynchronously. * - * on entry, *remaining can be set as a hint as to the maximum number + * max_size_hint can be set as a hint as to the maximum number * of bytes that would be acceptable to read. * - * fills *buffer, *length, *remaining with the bytes, length of bytes - * and length of data remaining to be read before either returning 1 - * or calling on_complete. - * * once a slice is returned into *slice, it is owned by the caller. */ int grpc_byte_stream_next(grpc_exec_ctx *exec_ctx, diff --git a/src/core/lib/transport/metadata.c b/src/core/lib/transport/metadata.c index 0677f29766..4b40c275ad 100644 --- a/src/core/lib/transport/metadata.c +++ b/src/core/lib/transport/metadata.c @@ -278,7 +278,7 @@ static void ref_md_locked(mdtab_shard *shard, internal_metadata *md DEBUG_ARGS) { #ifdef GRPC_METADATA_REFCOUNT_DEBUG gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "ELM REF:%p:%d->%d: '%s' = '%s'", md, + "ELM REF:%p:%zu->%zu: '%s' = '%s'", (void *)md, gpr_atm_no_barrier_load(&md->refcnt), gpr_atm_no_barrier_load(&md->refcnt) + 1, grpc_mdstr_as_c_string((grpc_mdstr *)md->key), @@ -566,7 +566,7 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdstr *mkey, shard->elems[idx] = md; gpr_mu_init(&md->mu_user_data); #ifdef GRPC_METADATA_REFCOUNT_DEBUG - gpr_log(GPR_DEBUG, "ELM NEW:%p:%d: '%s' = '%s'", md, + gpr_log(GPR_DEBUG, "ELM NEW:%p:%zu: '%s' = '%s'", (void *)md, gpr_atm_no_barrier_load(&md->refcnt), grpc_mdstr_as_c_string((grpc_mdstr *)md->key), grpc_mdstr_as_c_string((grpc_mdstr *)md->value)); @@ -639,7 +639,7 @@ grpc_mdelem *grpc_mdelem_ref(grpc_mdelem *gmd DEBUG_ARGS) { if (is_mdelem_static(gmd)) return gmd; #ifdef GRPC_METADATA_REFCOUNT_DEBUG gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "ELM REF:%p:%d->%d: '%s' = '%s'", md, + "ELM REF:%p:%zu->%zu: '%s' = '%s'", (void *)md, gpr_atm_no_barrier_load(&md->refcnt), gpr_atm_no_barrier_load(&md->refcnt) + 1, grpc_mdstr_as_c_string((grpc_mdstr *)md->key), @@ -649,7 +649,7 @@ grpc_mdelem *grpc_mdelem_ref(grpc_mdelem *gmd DEBUG_ARGS) { this function - meaning that no adjustment to mdtab_free is necessary, simplifying the logic here to be just an atomic increment */ /* use C assert to have this removed in opt builds */ - assert(gpr_atm_no_barrier_load(&md->refcnt) >= 1); + GPR_ASSERT(gpr_atm_no_barrier_load(&md->refcnt) >= 1); gpr_atm_no_barrier_fetch_add(&md->refcnt, 1); return gmd; } @@ -660,14 +660,16 @@ void grpc_mdelem_unref(grpc_mdelem *gmd DEBUG_ARGS) { if (is_mdelem_static(gmd)) return; #ifdef GRPC_METADATA_REFCOUNT_DEBUG gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "ELM UNREF:%p:%d->%d: '%s' = '%s'", md, + "ELM UNREF:%p:%zu->%zu: '%s' = '%s'", (void *)md, gpr_atm_no_barrier_load(&md->refcnt), gpr_atm_no_barrier_load(&md->refcnt) - 1, grpc_mdstr_as_c_string((grpc_mdstr *)md->key), grpc_mdstr_as_c_string((grpc_mdstr *)md->value)); #endif uint32_t hash = GRPC_MDSTR_KV_HASH(md->key->hash, md->value->hash); - if (1 == gpr_atm_full_fetch_add(&md->refcnt, -1)) { + const gpr_atm prev_refcount = gpr_atm_full_fetch_add(&md->refcnt, -1); + GPR_ASSERT(prev_refcount >= 1); + if (1 == prev_refcount) { /* once the refcount hits zero, some other thread can come along and free md at any time: it's unsafe from this point on to access it */ mdtab_shard *shard = @@ -676,10 +678,12 @@ void grpc_mdelem_unref(grpc_mdelem *gmd DEBUG_ARGS) { } } -const char *grpc_mdstr_as_c_string(grpc_mdstr *s) { +const char *grpc_mdstr_as_c_string(const grpc_mdstr *s) { return (const char *)GPR_SLICE_START_PTR(s->slice); } +size_t grpc_mdstr_length(const grpc_mdstr *s) { return GRPC_MDSTR_LENGTH(s); } + grpc_mdstr *grpc_mdstr_ref(grpc_mdstr *gs DEBUG_ARGS) { internal_string *s = (internal_string *)gs; if (is_mdstr_static(gs)) return gs; diff --git a/src/core/lib/transport/metadata.h b/src/core/lib/transport/metadata.h index 6d82f4d681..71eff0acf2 100644 --- a/src/core/lib/transport/metadata.h +++ b/src/core/lib/transport/metadata.h @@ -37,6 +37,10 @@ #include <grpc/support/slice.h> #include <grpc/support/useful.h> +#ifdef __cplusplus +extern "C" { +#endif + /* This file provides a mechanism for tracking metadata through the grpc stack. It's not intended for consumption outside of the library. @@ -143,7 +147,7 @@ void grpc_mdelem_unref(grpc_mdelem *md); /* Recover a char* from a grpc_mdstr. The returned string is null terminated. Does not promise that the returned string has no embedded nulls however. */ -const char *grpc_mdstr_as_c_string(grpc_mdstr *s); +const char *grpc_mdstr_as_c_string(const grpc_mdstr *s); #define GRPC_MDSTR_LENGTH(s) (GPR_SLICE_LENGTH(s->slice)) @@ -164,4 +168,8 @@ void grpc_mdctx_global_shutdown(void); extern gpr_slice (*grpc_chttp2_base64_encode_and_huffman_compress)( gpr_slice input); +#ifdef __cplusplus +} +#endif + #endif /* GRPC_CORE_LIB_TRANSPORT_METADATA_H */ diff --git a/src/core/lib/transport/metadata_batch.c b/src/core/lib/transport/metadata_batch.c index e4398abeb7..84b5a74d51 100644 --- a/src/core/lib/transport/metadata_batch.c +++ b/src/core/lib/transport/metadata_batch.c @@ -33,6 +33,7 @@ #include "src/core/lib/transport/metadata_batch.h" +#include <stdbool.h> #include <string.h> #include <grpc/support/alloc.h> @@ -187,7 +188,7 @@ void grpc_metadata_batch_clear(grpc_metadata_batch *batch) { grpc_metadata_batch_filter(batch, no_metadata_for_you, NULL); } -int grpc_metadata_batch_is_empty(grpc_metadata_batch *batch) { +bool grpc_metadata_batch_is_empty(grpc_metadata_batch *batch) { return batch->list.head == NULL && gpr_time_cmp(gpr_inf_future(batch->deadline.clock_type), batch->deadline) == 0; diff --git a/src/core/lib/transport/metadata_batch.h b/src/core/lib/transport/metadata_batch.h index 7af823f7ca..0424b4db98 100644 --- a/src/core/lib/transport/metadata_batch.h +++ b/src/core/lib/transport/metadata_batch.h @@ -34,12 +34,18 @@ #ifndef GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H #define GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H +#include <stdbool.h> + #include <grpc/grpc.h> #include <grpc/support/port_platform.h> #include <grpc/support/slice.h> #include <grpc/support/time.h> #include "src/core/lib/transport/metadata.h" +#ifdef __cplusplus +extern "C" { +#endif + typedef struct grpc_linked_mdelem { grpc_mdelem *md; struct grpc_linked_mdelem *next; @@ -64,7 +70,7 @@ typedef struct grpc_metadata_batch { void grpc_metadata_batch_init(grpc_metadata_batch *batch); void grpc_metadata_batch_destroy(grpc_metadata_batch *batch); void grpc_metadata_batch_clear(grpc_metadata_batch *batch); -int grpc_metadata_batch_is_empty(grpc_metadata_batch *batch); +bool grpc_metadata_batch_is_empty(grpc_metadata_batch *batch); /* Returns the transport size of the batch. */ size_t grpc_metadata_batch_size(grpc_metadata_batch *batch); @@ -125,4 +131,8 @@ void grpc_metadata_batch_assert_ok(grpc_metadata_batch *comd); } while (0) #endif +#ifdef __cplusplus +} +#endif + #endif /* GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H */ diff --git a/src/core/lib/transport/static_metadata.c b/src/core/lib/transport/static_metadata.c index c5f16e530d..5e0352a467 100644 --- a/src/core/lib/transport/static_metadata.c +++ b/src/core/lib/transport/static_metadata.c @@ -45,21 +45,21 @@ grpc_mdstr grpc_static_mdstr_table[GRPC_STATIC_MDSTR_COUNT]; grpc_mdelem grpc_static_mdelem_table[GRPC_STATIC_MDELEM_COUNT]; uintptr_t grpc_static_mdelem_user_data[GRPC_STATIC_MDELEM_COUNT] = { - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 4, 8, 6, 2, 4, 8, 6, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 4, 8, 6, 2, 4, 8, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; const uint8_t grpc_static_metadata_elem_indices[GRPC_STATIC_MDELEM_COUNT * 2] = - {11, 35, 10, 35, 12, 35, 12, 49, 13, 35, 14, 35, 15, 35, 16, 35, 17, 35, - 19, 35, 20, 35, 21, 35, 24, 35, 25, 35, 26, 35, 27, 35, 28, 35, 29, 35, - 30, 18, 30, 35, 31, 35, 32, 35, 36, 35, 37, 35, 38, 35, 39, 35, 42, 33, - 42, 34, 42, 48, 42, 53, 42, 54, 42, 55, 42, 56, 43, 33, 43, 48, 43, 53, - 46, 0, 46, 1, 46, 2, 50, 35, 57, 35, 58, 35, 59, 35, 60, 35, 61, 35, - 62, 35, 63, 35, 64, 35, 65, 35, 66, 35, 67, 40, 67, 69, 67, 72, 68, 80, - 68, 81, 70, 35, 71, 35, 73, 35, 74, 35, 75, 35, 76, 35, 77, 41, 77, 51, - 77, 52, 78, 35, 79, 35, 82, 3, 82, 4, 82, 5, 82, 6, 82, 7, 82, 8, - 82, 9, 83, 35, 84, 85, 86, 35, 87, 35, 88, 35, 89, 35, 90, 35}; + {11, 33, 10, 33, 12, 33, 12, 50, 13, 33, 14, 33, 15, 33, 16, 33, 17, 33, + 19, 33, 20, 33, 21, 33, 22, 33, 23, 33, 24, 33, 25, 33, 26, 33, 27, 33, + 28, 18, 28, 33, 29, 33, 30, 33, 34, 33, 35, 33, 36, 33, 37, 33, 40, 31, + 40, 32, 40, 49, 40, 54, 40, 55, 40, 56, 40, 57, 41, 31, 41, 49, 41, 54, + 46, 0, 46, 1, 46, 2, 51, 33, 58, 33, 59, 33, 60, 33, 61, 33, 62, 33, + 63, 33, 64, 33, 65, 33, 66, 33, 67, 33, 68, 33, 69, 38, 69, 71, 69, 74, + 70, 82, 70, 83, 72, 33, 73, 33, 75, 33, 76, 33, 77, 33, 78, 33, 79, 39, + 79, 52, 79, 53, 80, 33, 81, 33, 84, 3, 84, 4, 84, 5, 84, 6, 84, 7, + 84, 8, 84, 9, 85, 33, 86, 87, 88, 33, 89, 33, 90, 33, 91, 33, 92, 33}; const char *const grpc_static_metadata_strings[GRPC_STATIC_MDSTR_COUNT] = { "0", @@ -84,8 +84,6 @@ const char *const grpc_static_metadata_strings[GRPC_STATIC_MDSTR_COUNT] = { ":authority", "authorization", "cache-control", - "census-bin", - "census-binary-bin", "content-disposition", "content-encoding", "content-language", @@ -108,8 +106,11 @@ const char *const grpc_static_metadata_strings[GRPC_STATIC_MDSTR_COUNT] = { "grpc-encoding", "grpc-internal-encoding-request", "grpc-message", + "grpc-payload-bin", + "grpc-stats-bin", "grpc-status", "grpc-timeout", + "grpc-tracing-bin", "gzip", "gzip, deflate", "host", @@ -126,7 +127,8 @@ const char *const grpc_static_metadata_strings[GRPC_STATIC_MDSTR_COUNT] = { "if-unmodified-since", "last-modified", "link", - "load-reporting", + "load-reporting-initial", + "load-reporting-trailing", "location", "max-forwards", ":method", diff --git a/src/core/lib/transport/static_metadata.h b/src/core/lib/transport/static_metadata.h index 5ff0d2f3bc..5b9ee1a60a 100644 --- a/src/core/lib/transport/static_metadata.h +++ b/src/core/lib/transport/static_metadata.h @@ -44,7 +44,7 @@ #include "src/core/lib/transport/metadata.h" -#define GRPC_STATIC_MDSTR_COUNT 91 +#define GRPC_STATIC_MDSTR_COUNT 93 extern grpc_mdstr grpc_static_mdstr_table[GRPC_STATIC_MDSTR_COUNT]; /* "0" */ #define GRPC_MDSTR_0 (&grpc_static_mdstr_table[0]) @@ -90,147 +90,151 @@ extern grpc_mdstr grpc_static_mdstr_table[GRPC_STATIC_MDSTR_COUNT]; #define GRPC_MDSTR_AUTHORIZATION (&grpc_static_mdstr_table[20]) /* "cache-control" */ #define GRPC_MDSTR_CACHE_CONTROL (&grpc_static_mdstr_table[21]) -/* "census-bin" */ -#define GRPC_MDSTR_CENSUS_BIN (&grpc_static_mdstr_table[22]) -/* "census-binary-bin" */ -#define GRPC_MDSTR_CENSUS_BINARY_BIN (&grpc_static_mdstr_table[23]) /* "content-disposition" */ -#define GRPC_MDSTR_CONTENT_DISPOSITION (&grpc_static_mdstr_table[24]) +#define GRPC_MDSTR_CONTENT_DISPOSITION (&grpc_static_mdstr_table[22]) /* "content-encoding" */ -#define GRPC_MDSTR_CONTENT_ENCODING (&grpc_static_mdstr_table[25]) +#define GRPC_MDSTR_CONTENT_ENCODING (&grpc_static_mdstr_table[23]) /* "content-language" */ -#define GRPC_MDSTR_CONTENT_LANGUAGE (&grpc_static_mdstr_table[26]) +#define GRPC_MDSTR_CONTENT_LANGUAGE (&grpc_static_mdstr_table[24]) /* "content-length" */ -#define GRPC_MDSTR_CONTENT_LENGTH (&grpc_static_mdstr_table[27]) +#define GRPC_MDSTR_CONTENT_LENGTH (&grpc_static_mdstr_table[25]) /* "content-location" */ -#define GRPC_MDSTR_CONTENT_LOCATION (&grpc_static_mdstr_table[28]) +#define GRPC_MDSTR_CONTENT_LOCATION (&grpc_static_mdstr_table[26]) /* "content-range" */ -#define GRPC_MDSTR_CONTENT_RANGE (&grpc_static_mdstr_table[29]) +#define GRPC_MDSTR_CONTENT_RANGE (&grpc_static_mdstr_table[27]) /* "content-type" */ -#define GRPC_MDSTR_CONTENT_TYPE (&grpc_static_mdstr_table[30]) +#define GRPC_MDSTR_CONTENT_TYPE (&grpc_static_mdstr_table[28]) /* "cookie" */ -#define GRPC_MDSTR_COOKIE (&grpc_static_mdstr_table[31]) +#define GRPC_MDSTR_COOKIE (&grpc_static_mdstr_table[29]) /* "date" */ -#define GRPC_MDSTR_DATE (&grpc_static_mdstr_table[32]) +#define GRPC_MDSTR_DATE (&grpc_static_mdstr_table[30]) /* "deflate" */ -#define GRPC_MDSTR_DEFLATE (&grpc_static_mdstr_table[33]) +#define GRPC_MDSTR_DEFLATE (&grpc_static_mdstr_table[31]) /* "deflate,gzip" */ -#define GRPC_MDSTR_DEFLATE_COMMA_GZIP (&grpc_static_mdstr_table[34]) +#define GRPC_MDSTR_DEFLATE_COMMA_GZIP (&grpc_static_mdstr_table[32]) /* "" */ -#define GRPC_MDSTR_EMPTY (&grpc_static_mdstr_table[35]) +#define GRPC_MDSTR_EMPTY (&grpc_static_mdstr_table[33]) /* "etag" */ -#define GRPC_MDSTR_ETAG (&grpc_static_mdstr_table[36]) +#define GRPC_MDSTR_ETAG (&grpc_static_mdstr_table[34]) /* "expect" */ -#define GRPC_MDSTR_EXPECT (&grpc_static_mdstr_table[37]) +#define GRPC_MDSTR_EXPECT (&grpc_static_mdstr_table[35]) /* "expires" */ -#define GRPC_MDSTR_EXPIRES (&grpc_static_mdstr_table[38]) +#define GRPC_MDSTR_EXPIRES (&grpc_static_mdstr_table[36]) /* "from" */ -#define GRPC_MDSTR_FROM (&grpc_static_mdstr_table[39]) +#define GRPC_MDSTR_FROM (&grpc_static_mdstr_table[37]) /* "GET" */ -#define GRPC_MDSTR_GET (&grpc_static_mdstr_table[40]) +#define GRPC_MDSTR_GET (&grpc_static_mdstr_table[38]) /* "grpc" */ -#define GRPC_MDSTR_GRPC (&grpc_static_mdstr_table[41]) +#define GRPC_MDSTR_GRPC (&grpc_static_mdstr_table[39]) /* "grpc-accept-encoding" */ -#define GRPC_MDSTR_GRPC_ACCEPT_ENCODING (&grpc_static_mdstr_table[42]) +#define GRPC_MDSTR_GRPC_ACCEPT_ENCODING (&grpc_static_mdstr_table[40]) /* "grpc-encoding" */ -#define GRPC_MDSTR_GRPC_ENCODING (&grpc_static_mdstr_table[43]) +#define GRPC_MDSTR_GRPC_ENCODING (&grpc_static_mdstr_table[41]) /* "grpc-internal-encoding-request" */ -#define GRPC_MDSTR_GRPC_INTERNAL_ENCODING_REQUEST (&grpc_static_mdstr_table[44]) +#define GRPC_MDSTR_GRPC_INTERNAL_ENCODING_REQUEST (&grpc_static_mdstr_table[42]) /* "grpc-message" */ -#define GRPC_MDSTR_GRPC_MESSAGE (&grpc_static_mdstr_table[45]) +#define GRPC_MDSTR_GRPC_MESSAGE (&grpc_static_mdstr_table[43]) +/* "grpc-payload-bin" */ +#define GRPC_MDSTR_GRPC_PAYLOAD_BIN (&grpc_static_mdstr_table[44]) +/* "grpc-stats-bin" */ +#define GRPC_MDSTR_GRPC_STATS_BIN (&grpc_static_mdstr_table[45]) /* "grpc-status" */ #define GRPC_MDSTR_GRPC_STATUS (&grpc_static_mdstr_table[46]) /* "grpc-timeout" */ #define GRPC_MDSTR_GRPC_TIMEOUT (&grpc_static_mdstr_table[47]) +/* "grpc-tracing-bin" */ +#define GRPC_MDSTR_GRPC_TRACING_BIN (&grpc_static_mdstr_table[48]) /* "gzip" */ -#define GRPC_MDSTR_GZIP (&grpc_static_mdstr_table[48]) +#define GRPC_MDSTR_GZIP (&grpc_static_mdstr_table[49]) /* "gzip, deflate" */ -#define GRPC_MDSTR_GZIP_COMMA_DEFLATE (&grpc_static_mdstr_table[49]) +#define GRPC_MDSTR_GZIP_COMMA_DEFLATE (&grpc_static_mdstr_table[50]) /* "host" */ -#define GRPC_MDSTR_HOST (&grpc_static_mdstr_table[50]) +#define GRPC_MDSTR_HOST (&grpc_static_mdstr_table[51]) /* "http" */ -#define GRPC_MDSTR_HTTP (&grpc_static_mdstr_table[51]) +#define GRPC_MDSTR_HTTP (&grpc_static_mdstr_table[52]) /* "https" */ -#define GRPC_MDSTR_HTTPS (&grpc_static_mdstr_table[52]) +#define GRPC_MDSTR_HTTPS (&grpc_static_mdstr_table[53]) /* "identity" */ -#define GRPC_MDSTR_IDENTITY (&grpc_static_mdstr_table[53]) +#define GRPC_MDSTR_IDENTITY (&grpc_static_mdstr_table[54]) /* "identity,deflate" */ -#define GRPC_MDSTR_IDENTITY_COMMA_DEFLATE (&grpc_static_mdstr_table[54]) +#define GRPC_MDSTR_IDENTITY_COMMA_DEFLATE (&grpc_static_mdstr_table[55]) /* "identity,deflate,gzip" */ #define GRPC_MDSTR_IDENTITY_COMMA_DEFLATE_COMMA_GZIP \ - (&grpc_static_mdstr_table[55]) + (&grpc_static_mdstr_table[56]) /* "identity,gzip" */ -#define GRPC_MDSTR_IDENTITY_COMMA_GZIP (&grpc_static_mdstr_table[56]) +#define GRPC_MDSTR_IDENTITY_COMMA_GZIP (&grpc_static_mdstr_table[57]) /* "if-match" */ -#define GRPC_MDSTR_IF_MATCH (&grpc_static_mdstr_table[57]) +#define GRPC_MDSTR_IF_MATCH (&grpc_static_mdstr_table[58]) /* "if-modified-since" */ -#define GRPC_MDSTR_IF_MODIFIED_SINCE (&grpc_static_mdstr_table[58]) +#define GRPC_MDSTR_IF_MODIFIED_SINCE (&grpc_static_mdstr_table[59]) /* "if-none-match" */ -#define GRPC_MDSTR_IF_NONE_MATCH (&grpc_static_mdstr_table[59]) +#define GRPC_MDSTR_IF_NONE_MATCH (&grpc_static_mdstr_table[60]) /* "if-range" */ -#define GRPC_MDSTR_IF_RANGE (&grpc_static_mdstr_table[60]) +#define GRPC_MDSTR_IF_RANGE (&grpc_static_mdstr_table[61]) /* "if-unmodified-since" */ -#define GRPC_MDSTR_IF_UNMODIFIED_SINCE (&grpc_static_mdstr_table[61]) +#define GRPC_MDSTR_IF_UNMODIFIED_SINCE (&grpc_static_mdstr_table[62]) /* "last-modified" */ -#define GRPC_MDSTR_LAST_MODIFIED (&grpc_static_mdstr_table[62]) +#define GRPC_MDSTR_LAST_MODIFIED (&grpc_static_mdstr_table[63]) /* "link" */ -#define GRPC_MDSTR_LINK (&grpc_static_mdstr_table[63]) -/* "load-reporting" */ -#define GRPC_MDSTR_LOAD_REPORTING (&grpc_static_mdstr_table[64]) +#define GRPC_MDSTR_LINK (&grpc_static_mdstr_table[64]) +/* "load-reporting-initial" */ +#define GRPC_MDSTR_LOAD_REPORTING_INITIAL (&grpc_static_mdstr_table[65]) +/* "load-reporting-trailing" */ +#define GRPC_MDSTR_LOAD_REPORTING_TRAILING (&grpc_static_mdstr_table[66]) /* "location" */ -#define GRPC_MDSTR_LOCATION (&grpc_static_mdstr_table[65]) +#define GRPC_MDSTR_LOCATION (&grpc_static_mdstr_table[67]) /* "max-forwards" */ -#define GRPC_MDSTR_MAX_FORWARDS (&grpc_static_mdstr_table[66]) +#define GRPC_MDSTR_MAX_FORWARDS (&grpc_static_mdstr_table[68]) /* ":method" */ -#define GRPC_MDSTR_METHOD (&grpc_static_mdstr_table[67]) +#define GRPC_MDSTR_METHOD (&grpc_static_mdstr_table[69]) /* ":path" */ -#define GRPC_MDSTR_PATH (&grpc_static_mdstr_table[68]) +#define GRPC_MDSTR_PATH (&grpc_static_mdstr_table[70]) /* "POST" */ -#define GRPC_MDSTR_POST (&grpc_static_mdstr_table[69]) +#define GRPC_MDSTR_POST (&grpc_static_mdstr_table[71]) /* "proxy-authenticate" */ -#define GRPC_MDSTR_PROXY_AUTHENTICATE (&grpc_static_mdstr_table[70]) +#define GRPC_MDSTR_PROXY_AUTHENTICATE (&grpc_static_mdstr_table[72]) /* "proxy-authorization" */ -#define GRPC_MDSTR_PROXY_AUTHORIZATION (&grpc_static_mdstr_table[71]) +#define GRPC_MDSTR_PROXY_AUTHORIZATION (&grpc_static_mdstr_table[73]) /* "PUT" */ -#define GRPC_MDSTR_PUT (&grpc_static_mdstr_table[72]) +#define GRPC_MDSTR_PUT (&grpc_static_mdstr_table[74]) /* "range" */ -#define GRPC_MDSTR_RANGE (&grpc_static_mdstr_table[73]) +#define GRPC_MDSTR_RANGE (&grpc_static_mdstr_table[75]) /* "referer" */ -#define GRPC_MDSTR_REFERER (&grpc_static_mdstr_table[74]) +#define GRPC_MDSTR_REFERER (&grpc_static_mdstr_table[76]) /* "refresh" */ -#define GRPC_MDSTR_REFRESH (&grpc_static_mdstr_table[75]) +#define GRPC_MDSTR_REFRESH (&grpc_static_mdstr_table[77]) /* "retry-after" */ -#define GRPC_MDSTR_RETRY_AFTER (&grpc_static_mdstr_table[76]) +#define GRPC_MDSTR_RETRY_AFTER (&grpc_static_mdstr_table[78]) /* ":scheme" */ -#define GRPC_MDSTR_SCHEME (&grpc_static_mdstr_table[77]) +#define GRPC_MDSTR_SCHEME (&grpc_static_mdstr_table[79]) /* "server" */ -#define GRPC_MDSTR_SERVER (&grpc_static_mdstr_table[78]) +#define GRPC_MDSTR_SERVER (&grpc_static_mdstr_table[80]) /* "set-cookie" */ -#define GRPC_MDSTR_SET_COOKIE (&grpc_static_mdstr_table[79]) +#define GRPC_MDSTR_SET_COOKIE (&grpc_static_mdstr_table[81]) /* "/" */ -#define GRPC_MDSTR_SLASH (&grpc_static_mdstr_table[80]) +#define GRPC_MDSTR_SLASH (&grpc_static_mdstr_table[82]) /* "/index.html" */ -#define GRPC_MDSTR_SLASH_INDEX_DOT_HTML (&grpc_static_mdstr_table[81]) +#define GRPC_MDSTR_SLASH_INDEX_DOT_HTML (&grpc_static_mdstr_table[83]) /* ":status" */ -#define GRPC_MDSTR_STATUS (&grpc_static_mdstr_table[82]) +#define GRPC_MDSTR_STATUS (&grpc_static_mdstr_table[84]) /* "strict-transport-security" */ -#define GRPC_MDSTR_STRICT_TRANSPORT_SECURITY (&grpc_static_mdstr_table[83]) +#define GRPC_MDSTR_STRICT_TRANSPORT_SECURITY (&grpc_static_mdstr_table[85]) /* "te" */ -#define GRPC_MDSTR_TE (&grpc_static_mdstr_table[84]) +#define GRPC_MDSTR_TE (&grpc_static_mdstr_table[86]) /* "trailers" */ -#define GRPC_MDSTR_TRAILERS (&grpc_static_mdstr_table[85]) +#define GRPC_MDSTR_TRAILERS (&grpc_static_mdstr_table[87]) /* "transfer-encoding" */ -#define GRPC_MDSTR_TRANSFER_ENCODING (&grpc_static_mdstr_table[86]) +#define GRPC_MDSTR_TRANSFER_ENCODING (&grpc_static_mdstr_table[88]) /* "user-agent" */ -#define GRPC_MDSTR_USER_AGENT (&grpc_static_mdstr_table[87]) +#define GRPC_MDSTR_USER_AGENT (&grpc_static_mdstr_table[89]) /* "vary" */ -#define GRPC_MDSTR_VARY (&grpc_static_mdstr_table[88]) +#define GRPC_MDSTR_VARY (&grpc_static_mdstr_table[90]) /* "via" */ -#define GRPC_MDSTR_VIA (&grpc_static_mdstr_table[89]) +#define GRPC_MDSTR_VIA (&grpc_static_mdstr_table[91]) /* "www-authenticate" */ -#define GRPC_MDSTR_WWW_AUTHENTICATE (&grpc_static_mdstr_table[90]) +#define GRPC_MDSTR_WWW_AUTHENTICATE (&grpc_static_mdstr_table[92]) -#define GRPC_STATIC_MDELEM_COUNT 80 +#define GRPC_STATIC_MDELEM_COUNT 81 extern grpc_mdelem grpc_static_mdelem_table[GRPC_STATIC_MDELEM_COUNT]; extern uintptr_t grpc_static_mdelem_user_data[GRPC_STATIC_MDELEM_COUNT]; /* "accept-charset": "" */ @@ -335,73 +339,76 @@ extern uintptr_t grpc_static_mdelem_user_data[GRPC_STATIC_MDELEM_COUNT]; #define GRPC_MDELEM_LAST_MODIFIED_EMPTY (&grpc_static_mdelem_table[45]) /* "link": "" */ #define GRPC_MDELEM_LINK_EMPTY (&grpc_static_mdelem_table[46]) -/* "load-reporting": "" */ -#define GRPC_MDELEM_LOAD_REPORTING_EMPTY (&grpc_static_mdelem_table[47]) +/* "load-reporting-initial": "" */ +#define GRPC_MDELEM_LOAD_REPORTING_INITIAL_EMPTY (&grpc_static_mdelem_table[47]) +/* "load-reporting-trailing": "" */ +#define GRPC_MDELEM_LOAD_REPORTING_TRAILING_EMPTY \ + (&grpc_static_mdelem_table[48]) /* "location": "" */ -#define GRPC_MDELEM_LOCATION_EMPTY (&grpc_static_mdelem_table[48]) +#define GRPC_MDELEM_LOCATION_EMPTY (&grpc_static_mdelem_table[49]) /* "max-forwards": "" */ -#define GRPC_MDELEM_MAX_FORWARDS_EMPTY (&grpc_static_mdelem_table[49]) +#define GRPC_MDELEM_MAX_FORWARDS_EMPTY (&grpc_static_mdelem_table[50]) /* ":method": "GET" */ -#define GRPC_MDELEM_METHOD_GET (&grpc_static_mdelem_table[50]) +#define GRPC_MDELEM_METHOD_GET (&grpc_static_mdelem_table[51]) /* ":method": "POST" */ -#define GRPC_MDELEM_METHOD_POST (&grpc_static_mdelem_table[51]) +#define GRPC_MDELEM_METHOD_POST (&grpc_static_mdelem_table[52]) /* ":method": "PUT" */ -#define GRPC_MDELEM_METHOD_PUT (&grpc_static_mdelem_table[52]) +#define GRPC_MDELEM_METHOD_PUT (&grpc_static_mdelem_table[53]) /* ":path": "/" */ -#define GRPC_MDELEM_PATH_SLASH (&grpc_static_mdelem_table[53]) +#define GRPC_MDELEM_PATH_SLASH (&grpc_static_mdelem_table[54]) /* ":path": "/index.html" */ -#define GRPC_MDELEM_PATH_SLASH_INDEX_DOT_HTML (&grpc_static_mdelem_table[54]) +#define GRPC_MDELEM_PATH_SLASH_INDEX_DOT_HTML (&grpc_static_mdelem_table[55]) /* "proxy-authenticate": "" */ -#define GRPC_MDELEM_PROXY_AUTHENTICATE_EMPTY (&grpc_static_mdelem_table[55]) +#define GRPC_MDELEM_PROXY_AUTHENTICATE_EMPTY (&grpc_static_mdelem_table[56]) /* "proxy-authorization": "" */ -#define GRPC_MDELEM_PROXY_AUTHORIZATION_EMPTY (&grpc_static_mdelem_table[56]) +#define GRPC_MDELEM_PROXY_AUTHORIZATION_EMPTY (&grpc_static_mdelem_table[57]) /* "range": "" */ -#define GRPC_MDELEM_RANGE_EMPTY (&grpc_static_mdelem_table[57]) +#define GRPC_MDELEM_RANGE_EMPTY (&grpc_static_mdelem_table[58]) /* "referer": "" */ -#define GRPC_MDELEM_REFERER_EMPTY (&grpc_static_mdelem_table[58]) +#define GRPC_MDELEM_REFERER_EMPTY (&grpc_static_mdelem_table[59]) /* "refresh": "" */ -#define GRPC_MDELEM_REFRESH_EMPTY (&grpc_static_mdelem_table[59]) +#define GRPC_MDELEM_REFRESH_EMPTY (&grpc_static_mdelem_table[60]) /* "retry-after": "" */ -#define GRPC_MDELEM_RETRY_AFTER_EMPTY (&grpc_static_mdelem_table[60]) +#define GRPC_MDELEM_RETRY_AFTER_EMPTY (&grpc_static_mdelem_table[61]) /* ":scheme": "grpc" */ -#define GRPC_MDELEM_SCHEME_GRPC (&grpc_static_mdelem_table[61]) +#define GRPC_MDELEM_SCHEME_GRPC (&grpc_static_mdelem_table[62]) /* ":scheme": "http" */ -#define GRPC_MDELEM_SCHEME_HTTP (&grpc_static_mdelem_table[62]) +#define GRPC_MDELEM_SCHEME_HTTP (&grpc_static_mdelem_table[63]) /* ":scheme": "https" */ -#define GRPC_MDELEM_SCHEME_HTTPS (&grpc_static_mdelem_table[63]) +#define GRPC_MDELEM_SCHEME_HTTPS (&grpc_static_mdelem_table[64]) /* "server": "" */ -#define GRPC_MDELEM_SERVER_EMPTY (&grpc_static_mdelem_table[64]) +#define GRPC_MDELEM_SERVER_EMPTY (&grpc_static_mdelem_table[65]) /* "set-cookie": "" */ -#define GRPC_MDELEM_SET_COOKIE_EMPTY (&grpc_static_mdelem_table[65]) +#define GRPC_MDELEM_SET_COOKIE_EMPTY (&grpc_static_mdelem_table[66]) /* ":status": "200" */ -#define GRPC_MDELEM_STATUS_200 (&grpc_static_mdelem_table[66]) +#define GRPC_MDELEM_STATUS_200 (&grpc_static_mdelem_table[67]) /* ":status": "204" */ -#define GRPC_MDELEM_STATUS_204 (&grpc_static_mdelem_table[67]) +#define GRPC_MDELEM_STATUS_204 (&grpc_static_mdelem_table[68]) /* ":status": "206" */ -#define GRPC_MDELEM_STATUS_206 (&grpc_static_mdelem_table[68]) +#define GRPC_MDELEM_STATUS_206 (&grpc_static_mdelem_table[69]) /* ":status": "304" */ -#define GRPC_MDELEM_STATUS_304 (&grpc_static_mdelem_table[69]) +#define GRPC_MDELEM_STATUS_304 (&grpc_static_mdelem_table[70]) /* ":status": "400" */ -#define GRPC_MDELEM_STATUS_400 (&grpc_static_mdelem_table[70]) +#define GRPC_MDELEM_STATUS_400 (&grpc_static_mdelem_table[71]) /* ":status": "404" */ -#define GRPC_MDELEM_STATUS_404 (&grpc_static_mdelem_table[71]) +#define GRPC_MDELEM_STATUS_404 (&grpc_static_mdelem_table[72]) /* ":status": "500" */ -#define GRPC_MDELEM_STATUS_500 (&grpc_static_mdelem_table[72]) +#define GRPC_MDELEM_STATUS_500 (&grpc_static_mdelem_table[73]) /* "strict-transport-security": "" */ #define GRPC_MDELEM_STRICT_TRANSPORT_SECURITY_EMPTY \ - (&grpc_static_mdelem_table[73]) + (&grpc_static_mdelem_table[74]) /* "te": "trailers" */ -#define GRPC_MDELEM_TE_TRAILERS (&grpc_static_mdelem_table[74]) +#define GRPC_MDELEM_TE_TRAILERS (&grpc_static_mdelem_table[75]) /* "transfer-encoding": "" */ -#define GRPC_MDELEM_TRANSFER_ENCODING_EMPTY (&grpc_static_mdelem_table[75]) +#define GRPC_MDELEM_TRANSFER_ENCODING_EMPTY (&grpc_static_mdelem_table[76]) /* "user-agent": "" */ -#define GRPC_MDELEM_USER_AGENT_EMPTY (&grpc_static_mdelem_table[76]) +#define GRPC_MDELEM_USER_AGENT_EMPTY (&grpc_static_mdelem_table[77]) /* "vary": "" */ -#define GRPC_MDELEM_VARY_EMPTY (&grpc_static_mdelem_table[77]) +#define GRPC_MDELEM_VARY_EMPTY (&grpc_static_mdelem_table[78]) /* "via": "" */ -#define GRPC_MDELEM_VIA_EMPTY (&grpc_static_mdelem_table[78]) +#define GRPC_MDELEM_VIA_EMPTY (&grpc_static_mdelem_table[79]) /* "www-authenticate": "" */ -#define GRPC_MDELEM_WWW_AUTHENTICATE_EMPTY (&grpc_static_mdelem_table[79]) +#define GRPC_MDELEM_WWW_AUTHENTICATE_EMPTY (&grpc_static_mdelem_table[80]) extern const uint8_t grpc_static_metadata_elem_indices[GRPC_STATIC_MDELEM_COUNT * 2]; diff --git a/src/core/lib/transport/timeout_encoding.c b/src/core/lib/transport/timeout_encoding.c new file mode 100644 index 0000000000..b58ebbd0a8 --- /dev/null +++ b/src/core/lib/transport/timeout_encoding.c @@ -0,0 +1,188 @@ +/* + * + * 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/lib/transport/timeout_encoding.h" + +#include <stdio.h> +#include <string.h> + +#include <grpc/support/port_platform.h> +#include "src/core/lib/support/string.h" + +static int64_t round_up(int64_t x, int64_t divisor) { + return (x / divisor + (x % divisor != 0)) * divisor; +} + +/* round an integer up to the next value with three significant figures */ +static int64_t round_up_to_three_sig_figs(int64_t x) { + if (x < 1000) return x; + if (x < 10000) return round_up(x, 10); + if (x < 100000) return round_up(x, 100); + if (x < 1000000) return round_up(x, 1000); + if (x < 10000000) return round_up(x, 10000); + if (x < 100000000) return round_up(x, 100000); + if (x < 1000000000) return round_up(x, 1000000); + return round_up(x, 10000000); +} + +/* encode our minimum viable timeout value */ +static void enc_tiny(char *buffer) { memcpy(buffer, "1n", 3); } + +static void enc_ext(char *buffer, int64_t value, char ext) { + int n = int64_ttoa(value, buffer); + buffer[n] = ext; + buffer[n + 1] = 0; +} + +static void enc_seconds(char *buffer, int64_t sec) { + if (sec % 3600 == 0) { + enc_ext(buffer, sec / 3600, 'H'); + } else if (sec % 60 == 0) { + enc_ext(buffer, sec / 60, 'M'); + } else { + enc_ext(buffer, sec, 'S'); + } +} + +static void enc_nanos(char *buffer, int64_t x) { + x = round_up_to_three_sig_figs(x); + if (x < 100000) { + if (x % 1000 == 0) { + enc_ext(buffer, x / 1000, 'u'); + } else { + enc_ext(buffer, x, 'n'); + } + } else if (x < 100000000) { + if (x % 1000000 == 0) { + enc_ext(buffer, x / 1000000, 'm'); + } else { + enc_ext(buffer, x / 1000, 'u'); + } + } else if (x < 1000000000) { + enc_ext(buffer, x / 1000000, 'm'); + } else { + /* note that this is only ever called with times of less than one second, + so if we reach here the time must have been rounded up to a whole second + (and no more) */ + memcpy(buffer, "1S", 3); + } +} + +static void enc_micros(char *buffer, int64_t x) { + x = round_up_to_three_sig_figs(x); + if (x < 100000) { + if (x % 1000 == 0) { + enc_ext(buffer, x / 1000, 'm'); + } else { + enc_ext(buffer, x, 'u'); + } + } else if (x < 100000000) { + if (x % 1000000 == 0) { + enc_ext(buffer, x / 1000000, 'S'); + } else { + enc_ext(buffer, x / 1000, 'm'); + } + } else { + enc_ext(buffer, x / 1000000, 'S'); + } +} + +void grpc_http2_encode_timeout(gpr_timespec timeout, char *buffer) { + if (timeout.tv_sec < 0) { + enc_tiny(buffer); + } else if (timeout.tv_sec == 0) { + enc_nanos(buffer, timeout.tv_nsec); + } else if (timeout.tv_sec < 1000 && timeout.tv_nsec != 0) { + enc_micros(buffer, + (int64_t)(timeout.tv_sec * 1000000) + + (timeout.tv_nsec / 1000 + (timeout.tv_nsec % 1000 != 0))); + } else { + enc_seconds(buffer, timeout.tv_sec + (timeout.tv_nsec != 0)); + } +} + +static int is_all_whitespace(const char *p) { + while (*p == ' ') p++; + return *p == 0; +} + +int grpc_http2_decode_timeout(const char *buffer, gpr_timespec *timeout) { + int32_t x = 0; + const uint8_t *p = (const uint8_t *)buffer; + int have_digit = 0; + /* skip whitespace */ + for (; *p == ' '; p++) + ; + /* decode numeric part */ + for (; *p >= '0' && *p <= '9'; p++) { + int32_t digit = (int32_t)(*p - (uint8_t)'0'); + have_digit = 1; + /* spec allows max. 8 digits, but we allow values up to 1,000,000,000 */ + if (x >= (100 * 1000 * 1000)) { + if (x != (100 * 1000 * 1000) || digit != 0) { + *timeout = gpr_inf_future(GPR_TIMESPAN); + return 1; + } + } + x = x * 10 + digit; + } + if (!have_digit) return 0; + /* skip whitespace */ + for (; *p == ' '; p++) + ; + /* decode unit specifier */ + switch (*p) { + case 'n': + *timeout = gpr_time_from_nanos(x, GPR_TIMESPAN); + break; + case 'u': + *timeout = gpr_time_from_micros(x, GPR_TIMESPAN); + break; + case 'm': + *timeout = gpr_time_from_millis(x, GPR_TIMESPAN); + break; + case 'S': + *timeout = gpr_time_from_seconds(x, GPR_TIMESPAN); + break; + case 'M': + *timeout = gpr_time_from_minutes(x, GPR_TIMESPAN); + break; + case 'H': + *timeout = gpr_time_from_hours(x, GPR_TIMESPAN); + break; + default: + return 0; + } + p++; + return is_all_whitespace((const char *)p); +} diff --git a/src/core/lib/transport/timeout_encoding.h b/src/core/lib/transport/timeout_encoding.h new file mode 100644 index 0000000000..92f02f6ecd --- /dev/null +++ b/src/core/lib/transport/timeout_encoding.h @@ -0,0 +1,47 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_LIB_TRANSPORT_TIMEOUT_ENCODING_H +#define GRPC_CORE_LIB_TRANSPORT_TIMEOUT_ENCODING_H + +#include <grpc/support/time.h> +#include "src/core/lib/support/string.h" + +#define GRPC_HTTP2_TIMEOUT_ENCODE_MIN_BUFSIZE (GPR_LTOA_MIN_BUFSIZE + 1) + +/* Encode/decode timeouts to the GRPC over HTTP/2 format; + encoding may round up arbitrarily */ +void grpc_http2_encode_timeout(gpr_timespec timeout, char *buffer); +int grpc_http2_decode_timeout(const char *buffer, gpr_timespec *timeout); + +#endif /* GRPC_CORE_LIB_TRANSPORT_TIMEOUT_ENCODING_H */ diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c index 857c3909d2..82fc605218 100644 --- a/src/core/lib/transport/transport.c +++ b/src/core/lib/transport/transport.c @@ -32,10 +32,14 @@ */ #include "src/core/lib/transport/transport.h" + +#include <string.h> + #include <grpc/support/alloc.h> #include <grpc/support/atm.h> #include <grpc/support/log.h> #include <grpc/support/sync.h> + #include "src/core/lib/support/string.h" #include "src/core/lib/transport/transport_impl.h" @@ -43,7 +47,7 @@ void grpc_stream_ref(grpc_stream_refcount *refcount, const char *reason) { gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count); gpr_log(GPR_DEBUG, "%s %p:%p REF %d->%d %s", refcount->object_type, - refcount, refcount->destroy.cb_arg, val, val + 1, reason); + refcount, refcount->destroy.cb_arg, (int)val, (int)val + 1, reason); #else void grpc_stream_ref(grpc_stream_refcount *refcount) { #endif @@ -55,7 +59,7 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount, const char *reason) { gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count); gpr_log(GPR_DEBUG, "%s %p:%p UNREF %d->%d %s", refcount->object_type, - refcount, refcount->destroy.cb_arg, val, val - 1, reason); + refcount, refcount->destroy.cb_arg, (int)val, (int)val - 1, reason); #else void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount) { @@ -220,7 +224,7 @@ void grpc_transport_stream_op_add_cancellation_with_message( error = GRPC_ERROR_CREATE("Call cancelled"); } error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, status); - add_error(op, &op->close_error, error); + add_error(op, &op->cancel_error, error); } void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op, @@ -247,3 +251,26 @@ void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op, error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, status); add_error(op, &op->close_error, error); } + +typedef struct { + grpc_closure outer_on_complete; + grpc_closure *inner_on_complete; + grpc_transport_op op; +} made_transport_op; + +static void destroy_made_transport_op(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + made_transport_op *op = arg; + grpc_exec_ctx_sched(exec_ctx, op->inner_on_complete, GRPC_ERROR_REF(error), + NULL); + gpr_free(op); +} + +grpc_transport_op *grpc_make_transport_op(grpc_closure *on_complete) { + made_transport_op *op = gpr_malloc(sizeof(*op)); + grpc_closure_init(&op->outer_on_complete, destroy_made_transport_op, op); + op->inner_on_complete = on_complete; + memset(&op->op, 0, sizeof(op->op)); + op->op.on_consumed = &op->outer_on_complete; + return &op->op; +} diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h index e33fc5c761..8dc393fd61 100644 --- a/src/core/lib/transport/transport.h +++ b/src/core/lib/transport/transport.h @@ -43,6 +43,10 @@ #include "src/core/lib/transport/byte_stream.h" #include "src/core/lib/transport/metadata_batch.h" +#ifdef __cplusplus +extern "C" { +#endif + /* forward declarations */ typedef struct grpc_transport grpc_transport; @@ -96,6 +100,11 @@ void grpc_transport_move_one_way_stats(grpc_transport_one_way_stats *from, void grpc_transport_move_stats(grpc_transport_stream_stats *from, grpc_transport_stream_stats *to); +typedef struct { + grpc_closure closure; + void *args[2]; +} grpc_transport_private_op_data; + /* Transport stream op: a set of operations to perform on a transport against a single stream */ typedef struct grpc_transport_stream_op { @@ -120,6 +129,7 @@ typedef struct grpc_transport_stream_op { /** Receive initial metadata from the stream, into provided metadata batch. */ grpc_metadata_batch *recv_initial_metadata; bool *recv_idempotent_request; + bool *recv_cacheable_request; /** Should be enqueued when initial metadata is ready to be processed. */ grpc_closure *recv_initial_metadata_ready; @@ -144,6 +154,12 @@ typedef struct grpc_transport_stream_op { /* Indexes correspond to grpc_context_index enum values */ grpc_call_context_element *context; + + /*************************************************************************** + * remaining fields are initialized and used at the discretion of the + * transport implementation */ + + grpc_transport_private_op_data transport_private; } grpc_transport_stream_op; /** Transport op: a set of operations to perform on a transport as a whole */ @@ -158,7 +174,7 @@ typedef struct grpc_transport_op { /** should we send a goaway? after a goaway is sent, once there are no more active calls on the transport, the transport should disconnect */ - int send_goaway; + bool send_goaway; /** what should the goaway contain? */ grpc_status_code goaway_status; gpr_slice *goaway_message; @@ -177,6 +193,12 @@ typedef struct grpc_transport_op { grpc_pollset_set *bind_pollset_set; /** send a ping, call this back if not NULL */ grpc_closure *send_ping; + + /*************************************************************************** + * remaining fields are initialized and used at the discretion of the + * transport implementation */ + + grpc_transport_private_op_data transport_private; } grpc_transport_op; /* Returns the amount of memory required to store a grpc_stream for this @@ -268,4 +290,12 @@ void grpc_transport_destroy(grpc_exec_ctx *exec_ctx, grpc_transport *transport); char *grpc_transport_get_peer(grpc_exec_ctx *exec_ctx, grpc_transport *transport); +/* Allocate a grpc_transport_op, and preconfigure the on_consumed closure to + \a on_consumed and then delete the returned transport op */ +grpc_transport_op *grpc_make_transport_op(grpc_closure *on_consumed); + +#ifdef __cplusplus +} +#endif + #endif /* GRPC_CORE_LIB_TRANSPORT_TRANSPORT_H */ |