diff --git a/BUILD b/BUILD index b36ce254e9e..2b8c1180d2a 100644 --- a/BUILD +++ b/BUILD @@ -165,6 +165,7 @@ cc_library( "src/core/lib/channel/compress_filter.h", "src/core/lib/channel/connected_channel.h", "src/core/lib/channel/context.h", + "src/core/lib/channel/deadline_filter.h", "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", @@ -320,6 +321,7 @@ cc_library( "src/core/lib/channel/channel_stack_builder.c", "src/core/lib/channel/compress_filter.c", "src/core/lib/channel/connected_channel.c", + "src/core/lib/channel/deadline_filter.c", "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", @@ -560,6 +562,7 @@ cc_library( "src/core/lib/channel/compress_filter.h", "src/core/lib/channel/connected_channel.h", "src/core/lib/channel/context.h", + "src/core/lib/channel/deadline_filter.h", "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", @@ -702,6 +705,7 @@ cc_library( "src/core/lib/channel/channel_stack_builder.c", "src/core/lib/channel/compress_filter.c", "src/core/lib/channel/connected_channel.c", + "src/core/lib/channel/deadline_filter.c", "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", @@ -914,6 +918,7 @@ cc_library( "src/core/lib/channel/compress_filter.h", "src/core/lib/channel/connected_channel.h", "src/core/lib/channel/context.h", + "src/core/lib/channel/deadline_filter.h", "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", @@ -1046,6 +1051,7 @@ cc_library( "src/core/lib/channel/channel_stack_builder.c", "src/core/lib/channel/compress_filter.c", "src/core/lib/channel/connected_channel.c", + "src/core/lib/channel/deadline_filter.c", "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", @@ -1261,6 +1267,7 @@ cc_library( "src/core/lib/channel/compress_filter.h", "src/core/lib/channel/connected_channel.h", "src/core/lib/channel/context.h", + "src/core/lib/channel/deadline_filter.h", "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", @@ -1373,6 +1380,7 @@ cc_library( "src/core/lib/channel/channel_stack_builder.c", "src/core/lib/channel/compress_filter.c", "src/core/lib/channel/connected_channel.c", + "src/core/lib/channel/deadline_filter.c", "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", @@ -1671,6 +1679,7 @@ cc_library( "src/core/lib/channel/compress_filter.h", "src/core/lib/channel/connected_channel.h", "src/core/lib/channel/context.h", + "src/core/lib/channel/deadline_filter.h", "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", @@ -1778,6 +1787,7 @@ cc_library( "src/core/lib/channel/channel_stack_builder.c", "src/core/lib/channel/compress_filter.c", "src/core/lib/channel/connected_channel.c", + "src/core/lib/channel/deadline_filter.c", "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", @@ -2166,6 +2176,7 @@ objc_library( "src/core/lib/channel/channel_stack_builder.c", "src/core/lib/channel/compress_filter.c", "src/core/lib/channel/connected_channel.c", + "src/core/lib/channel/deadline_filter.c", "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", @@ -2385,6 +2396,7 @@ objc_library( "src/core/lib/channel/compress_filter.h", "src/core/lib/channel/connected_channel.h", "src/core/lib/channel/context.h", + "src/core/lib/channel/deadline_filter.h", "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", diff --git a/CMakeLists.txt b/CMakeLists.txt index fbcc6bbeaec..6eda5fe4f71 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -296,6 +296,7 @@ add_library(grpc src/core/lib/channel/channel_stack_builder.c src/core/lib/channel/compress_filter.c src/core/lib/channel/connected_channel.c + src/core/lib/channel/deadline_filter.c src/core/lib/channel/handshaker.c src/core/lib/channel/http_client_filter.c src/core/lib/channel/http_server_filter.c @@ -554,6 +555,7 @@ add_library(grpc_cronet src/core/lib/channel/channel_stack_builder.c src/core/lib/channel/compress_filter.c src/core/lib/channel/connected_channel.c + src/core/lib/channel/deadline_filter.c src/core/lib/channel/handshaker.c src/core/lib/channel/http_client_filter.c src/core/lib/channel/http_server_filter.c @@ -786,6 +788,7 @@ add_library(grpc_unsecure src/core/lib/channel/channel_stack_builder.c src/core/lib/channel/compress_filter.c src/core/lib/channel/connected_channel.c + src/core/lib/channel/deadline_filter.c src/core/lib/channel/handshaker.c src/core/lib/channel/http_client_filter.c src/core/lib/channel/http_server_filter.c @@ -1043,6 +1046,7 @@ add_library(grpc++ src/core/lib/channel/channel_stack_builder.c src/core/lib/channel/compress_filter.c src/core/lib/channel/connected_channel.c + src/core/lib/channel/deadline_filter.c src/core/lib/channel/handshaker.c src/core/lib/channel/http_client_filter.c src/core/lib/channel/http_server_filter.c @@ -1400,6 +1404,7 @@ add_library(grpc++_unsecure src/core/lib/channel/channel_stack_builder.c src/core/lib/channel/compress_filter.c src/core/lib/channel/connected_channel.c + src/core/lib/channel/deadline_filter.c src/core/lib/channel/handshaker.c src/core/lib/channel/http_client_filter.c src/core/lib/channel/http_server_filter.c diff --git a/Makefile b/Makefile index c7622c0b9a3..2a25ccb47f2 100644 --- a/Makefile +++ b/Makefile @@ -2512,6 +2512,7 @@ LIBGRPC_SRC = \ src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ + src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ @@ -2788,6 +2789,7 @@ LIBGRPC_CRONET_SRC = \ src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ + src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ @@ -3054,6 +3056,7 @@ LIBGRPC_TEST_UTIL_SRC = \ src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ + src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ @@ -3247,6 +3250,7 @@ LIBGRPC_UNSECURE_SRC = \ src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ + src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ @@ -3587,6 +3591,7 @@ LIBGRPC++_SRC = \ src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ + src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ @@ -4222,6 +4227,7 @@ LIBGRPC++_UNSECURE_SRC = \ src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ + src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ diff --git a/binding.gyp b/binding.gyp index 17dbfc0d381..1c29798bd52 100644 --- a/binding.gyp +++ b/binding.gyp @@ -568,6 +568,7 @@ 'src/core/lib/channel/channel_stack_builder.c', 'src/core/lib/channel/compress_filter.c', 'src/core/lib/channel/connected_channel.c', + 'src/core/lib/channel/deadline_filter.c', 'src/core/lib/channel/handshaker.c', 'src/core/lib/channel/http_client_filter.c', 'src/core/lib/channel/http_server_filter.c', diff --git a/build.yaml b/build.yaml index 506a02920c1..7f8a7d032b6 100644 --- a/build.yaml +++ b/build.yaml @@ -161,6 +161,7 @@ filegroups: - src/core/lib/channel/compress_filter.h - src/core/lib/channel/connected_channel.h - src/core/lib/channel/context.h + - src/core/lib/channel/deadline_filter.h - src/core/lib/channel/handshaker.h - src/core/lib/channel/http_client_filter.h - src/core/lib/channel/http_server_filter.h @@ -241,6 +242,7 @@ filegroups: - src/core/lib/channel/channel_stack_builder.c - src/core/lib/channel/compress_filter.c - src/core/lib/channel/connected_channel.c + - src/core/lib/channel/deadline_filter.c - src/core/lib/channel/handshaker.c - src/core/lib/channel/http_client_filter.c - src/core/lib/channel/http_server_filter.c diff --git a/config.m4 b/config.m4 index b37658dc617..2fff96f14a5 100644 --- a/config.m4 +++ b/config.m4 @@ -87,6 +87,7 @@ if test "$PHP_GRPC" != "no"; then src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ + src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 86b811a3ec6..5eee9c486ea 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -258,6 +258,7 @@ Pod::Spec.new do |s| 'src/core/lib/channel/compress_filter.h', 'src/core/lib/channel/connected_channel.h', 'src/core/lib/channel/context.h', + 'src/core/lib/channel/deadline_filter.h', 'src/core/lib/channel/handshaker.h', 'src/core/lib/channel/http_client_filter.h', 'src/core/lib/channel/http_server_filter.h', @@ -417,6 +418,7 @@ Pod::Spec.new do |s| 'src/core/lib/channel/channel_stack_builder.c', 'src/core/lib/channel/compress_filter.c', 'src/core/lib/channel/connected_channel.c', + 'src/core/lib/channel/deadline_filter.c', 'src/core/lib/channel/handshaker.c', 'src/core/lib/channel/http_client_filter.c', 'src/core/lib/channel/http_server_filter.c', @@ -619,6 +621,7 @@ Pod::Spec.new do |s| 'src/core/lib/channel/compress_filter.h', 'src/core/lib/channel/connected_channel.h', 'src/core/lib/channel/context.h', + 'src/core/lib/channel/deadline_filter.h', 'src/core/lib/channel/handshaker.h', 'src/core/lib/channel/http_client_filter.h', 'src/core/lib/channel/http_server_filter.h', diff --git a/grpc.gemspec b/grpc.gemspec index 8d74e36e7b6..4b79e6cda09 100755 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -177,6 +177,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/channel/compress_filter.h ) s.files += %w( src/core/lib/channel/connected_channel.h ) s.files += %w( src/core/lib/channel/context.h ) + s.files += %w( src/core/lib/channel/deadline_filter.h ) s.files += %w( src/core/lib/channel/handshaker.h ) s.files += %w( src/core/lib/channel/http_client_filter.h ) s.files += %w( src/core/lib/channel/http_server_filter.h ) @@ -336,6 +337,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/lib/channel/channel_stack_builder.c ) s.files += %w( src/core/lib/channel/compress_filter.c ) s.files += %w( src/core/lib/channel/connected_channel.c ) + s.files += %w( src/core/lib/channel/deadline_filter.c ) s.files += %w( src/core/lib/channel/handshaker.c ) s.files += %w( src/core/lib/channel/http_client_filter.c ) s.files += %w( src/core/lib/channel/http_server_filter.c ) diff --git a/package.xml b/package.xml index 387afc34982..5f774ef5c5b 100644 --- a/package.xml +++ b/package.xml @@ -185,6 +185,7 @@ + @@ -344,6 +345,7 @@ + diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c new file mode 100644 index 00000000000..032dea02213 --- /dev/null +++ b/src/core/lib/channel/deadline_filter.c @@ -0,0 +1,209 @@ +// +// 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 +#include + +#include +#include + +#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 { + grpc_call_stack* call_stack; + bool timer_pending; + grpc_timer timer; +} 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; + 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_log(GPR_INFO, "==> destroy_call_elem()"); +// FIXME: this is not working -- timer holds a ref, so we won't get +// called until after timer pops + if (calld->timer_pending) +{ +gpr_log(GPR_INFO, "CANCELLING TIMER"); + grpc_timer_cancel(exec_ctx, &calld->timer); +} + +} + +// 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; + calld->timer_pending = false; + if (error != GRPC_ERROR_CANCELLED) { +gpr_log(GPR_INFO, "DEADLINE EXCEEDED"); + gpr_slice message = gpr_slice_from_static_string("Deadline Exceeded"); + grpc_call_element_send_cancel_with_message( + exec_ctx, elem, GRPC_STATUS_DEADLINE_EXCEEDED, &message); + } +else gpr_log(GPR_INFO, "TIMER CANCELLED"); +gpr_log(GPR_INFO, "UNREF"); + GRPC_CALL_STACK_UNREF(exec_ctx, calld->call_stack, "deadline"); +} + +// 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. +gpr_log(GPR_INFO, "REF"); + GRPC_CALL_STACK_REF(calld->call_stack, "deadline"); + grpc_timer_init(exec_ctx, &calld->timer, deadline, timer_callback, elem, + gpr_now(GPR_CLOCK_MONOTONIC)); + calld->timer_pending = true; + } +} + +// 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) { + // 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); + } + // 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 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; + } + // 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 00000000000..323cb4e10cf --- /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 772681109a0..c05ed67c432 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/init.c b/src/core/lib/surface/init.c index 5397913a21f..4a3c03b915b 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" @@ -97,6 +98,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); diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index 7ae76f52c1f..bc8e8ea4447 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -81,6 +81,7 @@ CORE_SOURCE_FILES = [ 'src/core/lib/channel/channel_stack_builder.c', 'src/core/lib/channel/compress_filter.c', 'src/core/lib/channel/connected_channel.c', + 'src/core/lib/channel/deadline_filter.c', 'src/core/lib/channel/handshaker.c', 'src/core/lib/channel/http_client_filter.c', 'src/core/lib/channel/http_server_filter.c', diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index 12eb6513848..ebaca73ed24 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -879,6 +879,7 @@ src/core/lib/channel/channel_stack_builder.h \ src/core/lib/channel/compress_filter.h \ src/core/lib/channel/connected_channel.h \ src/core/lib/channel/context.h \ +src/core/lib/channel/deadline_filter.h \ src/core/lib/channel/handshaker.h \ src/core/lib/channel/http_client_filter.h \ src/core/lib/channel/http_server_filter.h \ @@ -991,6 +992,7 @@ src/core/lib/channel/channel_stack.c \ src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ +src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 426c6d91e36..71ac3496163 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -796,6 +796,7 @@ src/core/lib/channel/channel_stack_builder.h \ src/core/lib/channel/compress_filter.h \ src/core/lib/channel/connected_channel.h \ src/core/lib/channel/context.h \ +src/core/lib/channel/deadline_filter.h \ src/core/lib/channel/handshaker.h \ src/core/lib/channel/http_client_filter.h \ src/core/lib/channel/http_server_filter.h \ @@ -955,6 +956,7 @@ src/core/lib/channel/channel_stack.c \ src/core/lib/channel/channel_stack_builder.c \ src/core/lib/channel/compress_filter.c \ src/core/lib/channel/connected_channel.c \ +src/core/lib/channel/deadline_filter.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ diff --git a/tools/run_tests/sources_and_headers.json b/tools/run_tests/sources_and_headers.json index 11daf24e0a2..adc37dff83e 100644 --- a/tools/run_tests/sources_and_headers.json +++ b/tools/run_tests/sources_and_headers.json @@ -5823,6 +5823,7 @@ "src/core/lib/channel/compress_filter.h", "src/core/lib/channel/connected_channel.h", "src/core/lib/channel/context.h", + "src/core/lib/channel/deadline_filter.h", "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", @@ -5919,6 +5920,8 @@ "src/core/lib/channel/connected_channel.c", "src/core/lib/channel/connected_channel.h", "src/core/lib/channel/context.h", + "src/core/lib/channel/deadline_filter.c", + "src/core/lib/channel/deadline_filter.h", "src/core/lib/channel/handshaker.c", "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.c", diff --git a/vsprojects/vcxproj/grpc++/grpc++.vcxproj b/vsprojects/vcxproj/grpc++/grpc++.vcxproj index 321a403c497..baee3a309e2 100644 --- a/vsprojects/vcxproj/grpc++/grpc++.vcxproj +++ b/vsprojects/vcxproj/grpc++/grpc++.vcxproj @@ -379,6 +379,7 @@ + @@ -531,6 +532,8 @@ + + diff --git a/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters b/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters index b34ca03a535..fec96cda4c2 100644 --- a/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters +++ b/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters @@ -115,6 +115,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel @@ -731,6 +734,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel diff --git a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj index a7bb3ef23d1..a1a42d82b9d 100644 --- a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj @@ -375,6 +375,7 @@ + @@ -517,6 +518,8 @@ + + diff --git a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters index 4ad0ae31d93..78c0a924ed4 100644 --- a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters @@ -100,6 +100,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel @@ -704,6 +707,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj b/vsprojects/vcxproj/grpc/grpc.vcxproj index 252704756d1..e892f0d57d2 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj @@ -305,6 +305,7 @@ + @@ -472,6 +473,8 @@ + + diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters index 1fdf0f5ecad..d4a6d8834a7 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters @@ -19,6 +19,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel @@ -683,6 +686,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel diff --git a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj index 80dd6b2dcb2..1a74cf4b39e 100644 --- a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj +++ b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj @@ -196,6 +196,7 @@ + @@ -316,6 +317,8 @@ + + diff --git a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters index 8dc28d1cb99..ce5a1326dc3 100644 --- a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters @@ -67,6 +67,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel @@ -464,6 +467,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj index fffb409781d..96b6569e3e9 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj @@ -295,6 +295,7 @@ + @@ -440,6 +441,8 @@ + + diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters index 8c8afb6b861..0e2c27779b9 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters @@ -22,6 +22,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel @@ -593,6 +596,9 @@ src\core\lib\channel + + src\core\lib\channel + src\core\lib\channel