diff options
Diffstat (limited to 'src/core/lib')
| -rw-r--r-- | src/core/lib/channel/deadline_filter.c | 261 | ||||
| -rw-r--r-- | src/core/lib/channel/deadline_filter.h | 40 | ||||
| -rw-r--r-- | src/core/lib/surface/call.c | 54 | ||||
| -rw-r--r-- | src/core/lib/surface/completion_queue.h | 2 | ||||
| -rw-r--r-- | src/core/lib/surface/init.c | 10 | ||||
| -rw-r--r-- | src/core/lib/transport/transport.c | 8 |
6 files changed, 321 insertions, 54 deletions
diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c new file mode 100644 index 0000000000..9dff1c4d63 --- /dev/null +++ b/src/core/lib/channel/deadline_filter.c @@ -0,0 +1,261 @@ +// +// 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/log.h> +#include <grpc/support/sync.h> +#include <grpc/support/time.h> + +#include "src/core/lib/iomgr/timer.h" + +// Used for both client and server filters. +typedef struct channel_data { +} channel_data; + +// Call data used for both client and server filter. +typedef struct base_call_data { + // 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; +} 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 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) { +} + +// 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) { + base_call_data* calld = elem->call_data; + // Note: size of call data is different between client and server. + memset(calld, 0, elem->filter->sizeof_call_data); + calld->call_stack = args->call_stack; + gpr_mu_init(&calld->timer_mu); + 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) { + base_call_data* calld = elem->call_data; + gpr_mu_destroy(&calld->timer_mu); +} + +// Timer callback. +static void timer_callback(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_call_element* elem = arg; + base_call_data* calld = elem->call_data; + gpr_mu_lock(&calld->timer_mu); + calld->timer_pending = false; + gpr_mu_unlock(&calld->timer_mu); + if (error != GRPC_ERROR_CANCELLED) { + grpc_call_element_send_cancel(exec_ctx, elem); + } + GRPC_CALL_STACK_UNREF(exec_ctx, calld->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) { + base_call_data* calld = 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(calld->call_stack, "deadline_timer"); + gpr_mu_lock(&calld->timer_mu); + calld->timer_pending = true; + grpc_timer_init(exec_ctx, &calld->timer, deadline, timer_callback, elem, + gpr_now(GPR_CLOCK_MONOTONIC)); + gpr_mu_unlock(&calld->timer_mu); + } +} + +// Cancels the deadline timer. +static void cancel_timer_if_needed(grpc_exec_ctx* exec_ctx, + base_call_data* calld) { + gpr_mu_lock(&calld->timer_mu); + if (calld->timer_pending) { + grpc_timer_cancel(exec_ctx, &calld->timer); + calld->timer_pending = false; + } + gpr_mu_unlock(&calld->timer_mu); +} + +// Callback run when the call is complete. +static void on_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + base_call_data* calld = arg; + cancel_timer_if_needed(exec_ctx, calld); + // Invoke the next callback. + calld->next_on_complete->cb(exec_ctx, calld->next_on_complete->cb_arg, error); +} + +// 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) { + base_call_data* calld = elem->call_data; + // If the call is cancelled or closed, cancel the timer. + if (op->cancel_error != GRPC_ERROR_NONE || + op->close_error != GRPC_ERROR_NONE) { + cancel_timer_if_needed(exec_ctx, calld); + } else { + // If we're sending initial metadata, get the deadline from the metadata + // and start the timer if needed. + if (op->send_initial_metadata != NULL) { + start_timer_if_needed(exec_ctx, elem, + op->send_initial_metadata->deadline); + } + // Make sure we know when the call is complete, so that we can cancel + // the timer. + if (op->recv_trailing_metadata != NULL) { + calld->next_on_complete = op->on_complete; + grpc_closure_init(&calld->on_complete, on_complete, calld); + op->on_complete = &calld->on_complete; + } + } + // 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 the call is cancelled or closed, cancel the timer. + if (op->cancel_error != GRPC_ERROR_NONE || + op->close_error != GRPC_ERROR_NONE) { + cancel_timer_if_needed(exec_ctx, &calld->base); + } 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) { + calld->base.next_on_complete = op->on_complete; + grpc_closure_init(&calld->base.on_complete, on_complete, calld); + op->on_complete = &calld->base.on_complete; + } + } + // 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, + 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, + 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..323cb4e10c --- /dev/null +++ b/src/core/lib/channel/deadline_filter.h @@ -0,0 +1,40 @@ +// +// 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" + +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/surface/call.c b/src/core/lib/surface/call.c index 772681109a..c05ed67c43 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -122,8 +122,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 */ @@ -166,9 +164,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]; @@ -211,8 +206,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, @@ -260,7 +253,7 @@ grpc_call *grpc_call_create( call->metadata_batch[i][j].deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); } } - call->send_deadline = + send_deadline = gpr_convert_clock_type(send_deadline, GPR_CLOCK_MONOTONIC); GRPC_CHANNEL_INTERNAL_REF(channel, "call"); /* initial refcount dropped by grpc_call_destroy */ @@ -334,10 +327,7 @@ 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_exec_ctx_finish(&exec_ctx); GPR_TIMER_END("grpc_call_create", 0); return call; @@ -736,9 +726,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); @@ -897,32 +884,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 */ @@ -1271,14 +1232,6 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx, GPR_TIMER_BEGIN("validate_filtered_metadata", 0); validate_filtered_metadata(exec_ctx, bctl); GPR_TIMER_END("validate_filtered_metadata", 0); - - if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) != - 0 && - !call->is_client) { - GPR_TIMER_BEGIN("set_deadline_alarm", 0); - set_deadline_alarm(exec_ctx, call, md->deadline); - GPR_TIMER_END("set_deadline_alarm", 0); - } } call->has_initial_md_been_received = true; @@ -1326,9 +1279,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) { 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..4a3c03b915 100644 --- a/src/core/lib/surface/init.c +++ b/src/core/lib/surface/init.c @@ -43,6 +43,7 @@ #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/debug/trace.h" @@ -98,6 +99,15 @@ static bool maybe_add_http_filter(grpc_channel_stack_builder *builder, static void register_builtin_channel_init() { grpc_channel_init_register_stage( + GRPC_CLIENT_SUBCHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, + prepend_filter, (void *)&grpc_client_deadline_filter); + 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_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, prepend_filter, (void *)&grpc_compress_filter); grpc_channel_init_register_stage( diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c index 857c3909d2..d4e197fa5c 100644 --- a/src/core/lib/transport/transport.c +++ b/src/core/lib/transport/transport.c @@ -43,7 +43,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 +55,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,6 +220,10 @@ 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); + // TODO(ctiller): We are intentionally setting close_error instead of + // cancel_error here. This is an ugly hack and should be replaced + // by a more general-purpose mechanism that allows us to control + // cancel/close behavior. add_error(op, &op->close_error, error); } |