From af00d8bfb24f6debdc2a0ac78d2dc5bf67fc01e4 Mon Sep 17 00:00:00 2001 From: "Mark D. Roth" <roth@google.com> Date: Tue, 23 Aug 2016 12:48:16 -0700 Subject: [PATCH] Add channel arg for max send message size and add message size filter. --- BUILD | 8 + CMakeLists.txt | 3 + Makefile | 4 + binding.gyp | 1 + build.yaml | 2 + config.m4 | 1 + gRPC-Core.podspec | 3 + grpc.gemspec | 2 + include/grpc/impl/codegen/grpc_types.h | 6 +- package.xml | 2 + src/core/lib/channel/message_size_filter.c | 175 ++++++++++++++++++ src/core/lib/channel/message_size_filter.h | 40 ++++ src/core/lib/surface/call.c | 11 -- src/core/lib/surface/channel.c | 21 +-- src/core/lib/surface/channel.h | 1 - src/core/lib/surface/init.c | 10 + src/python/grpcio/grpc_core_dependencies.py | 1 + test/core/end2end/tests/max_message_length.c | 45 +++-- tools/doxygen/Doxyfile.core.internal | 2 + tools/run_tests/sources_and_headers.json | 3 + vsprojects/vcxproj/grpc/grpc.vcxproj | 3 + vsprojects/vcxproj/grpc/grpc.vcxproj.filters | 6 + .../grpc_test_util/grpc_test_util.vcxproj | 3 + .../grpc_test_util.vcxproj.filters | 6 + .../grpc_unsecure/grpc_unsecure.vcxproj | 3 + .../grpc_unsecure.vcxproj.filters | 6 + 26 files changed, 320 insertions(+), 48 deletions(-) create mode 100644 src/core/lib/channel/message_size_filter.c create mode 100644 src/core/lib/channel/message_size_filter.h diff --git a/BUILD b/BUILD index 12f0fd316e..7f661e9714 100644 --- a/BUILD +++ b/BUILD @@ -168,6 +168,7 @@ cc_library( "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", + "src/core/lib/channel/message_size_filter.h", "src/core/lib/compression/algorithm_metadata.h", "src/core/lib/compression/message_compress.h", "src/core/lib/debug/trace.h", @@ -323,6 +324,7 @@ cc_library( "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", + "src/core/lib/channel/message_size_filter.c", "src/core/lib/compression/compression.c", "src/core/lib/compression/message_compress.c", "src/core/lib/debug/trace.c", @@ -563,6 +565,7 @@ cc_library( "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", + "src/core/lib/channel/message_size_filter.h", "src/core/lib/compression/algorithm_metadata.h", "src/core/lib/compression/message_compress.h", "src/core/lib/debug/trace.h", @@ -705,6 +708,7 @@ cc_library( "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", + "src/core/lib/channel/message_size_filter.c", "src/core/lib/compression/compression.c", "src/core/lib/compression/message_compress.c", "src/core/lib/debug/trace.c", @@ -917,6 +921,7 @@ cc_library( "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", + "src/core/lib/channel/message_size_filter.h", "src/core/lib/compression/algorithm_metadata.h", "src/core/lib/compression/message_compress.h", "src/core/lib/debug/trace.h", @@ -1049,6 +1054,7 @@ cc_library( "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", + "src/core/lib/channel/message_size_filter.c", "src/core/lib/compression/compression.c", "src/core/lib/compression/message_compress.c", "src/core/lib/debug/trace.c", @@ -1816,6 +1822,7 @@ objc_library( "src/core/lib/channel/handshaker.c", "src/core/lib/channel/http_client_filter.c", "src/core/lib/channel/http_server_filter.c", + "src/core/lib/channel/message_size_filter.c", "src/core/lib/compression/compression.c", "src/core/lib/compression/message_compress.c", "src/core/lib/debug/trace.c", @@ -2035,6 +2042,7 @@ objc_library( "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", + "src/core/lib/channel/message_size_filter.h", "src/core/lib/compression/algorithm_metadata.h", "src/core/lib/compression/message_compress.h", "src/core/lib/debug/trace.h", diff --git a/CMakeLists.txt b/CMakeLists.txt index f847c9b0e6..567ec1fcb3 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -299,6 +299,7 @@ add_library(grpc src/core/lib/channel/handshaker.c src/core/lib/channel/http_client_filter.c src/core/lib/channel/http_server_filter.c + src/core/lib/channel/message_size_filter.c src/core/lib/compression/compression.c src/core/lib/compression/message_compress.c src/core/lib/debug/trace.c @@ -557,6 +558,7 @@ add_library(grpc_cronet src/core/lib/channel/handshaker.c src/core/lib/channel/http_client_filter.c src/core/lib/channel/http_server_filter.c + src/core/lib/channel/message_size_filter.c src/core/lib/compression/compression.c src/core/lib/compression/message_compress.c src/core/lib/debug/trace.c @@ -789,6 +791,7 @@ add_library(grpc_unsecure src/core/lib/channel/handshaker.c src/core/lib/channel/http_client_filter.c src/core/lib/channel/http_server_filter.c + src/core/lib/channel/message_size_filter.c src/core/lib/compression/compression.c src/core/lib/compression/message_compress.c src/core/lib/debug/trace.c diff --git a/Makefile b/Makefile index 6b619e310b..fdd1bfdfd1 100644 --- a/Makefile +++ b/Makefile @@ -2515,6 +2515,7 @@ LIBGRPC_SRC = \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ + src/core/lib/channel/message_size_filter.c \ src/core/lib/compression/compression.c \ src/core/lib/compression/message_compress.c \ src/core/lib/debug/trace.c \ @@ -2791,6 +2792,7 @@ LIBGRPC_CRONET_SRC = \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ + src/core/lib/channel/message_size_filter.c \ src/core/lib/compression/compression.c \ src/core/lib/compression/message_compress.c \ src/core/lib/debug/trace.c \ @@ -3057,6 +3059,7 @@ LIBGRPC_TEST_UTIL_SRC = \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ + src/core/lib/channel/message_size_filter.c \ src/core/lib/compression/compression.c \ src/core/lib/compression/message_compress.c \ src/core/lib/debug/trace.c \ @@ -3249,6 +3252,7 @@ LIBGRPC_UNSECURE_SRC = \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ + src/core/lib/channel/message_size_filter.c \ src/core/lib/compression/compression.c \ src/core/lib/compression/message_compress.c \ src/core/lib/debug/trace.c \ diff --git a/binding.gyp b/binding.gyp index a29cfda6fc..2cb43b0eb5 100644 --- a/binding.gyp +++ b/binding.gyp @@ -571,6 +571,7 @@ 'src/core/lib/channel/handshaker.c', 'src/core/lib/channel/http_client_filter.c', 'src/core/lib/channel/http_server_filter.c', + 'src/core/lib/channel/message_size_filter.c', 'src/core/lib/compression/compression.c', 'src/core/lib/compression/message_compress.c', 'src/core/lib/debug/trace.c', diff --git a/build.yaml b/build.yaml index 73c626351d..a0ff4ec859 100644 --- a/build.yaml +++ b/build.yaml @@ -163,6 +163,7 @@ filegroups: - src/core/lib/channel/handshaker.h - src/core/lib/channel/http_client_filter.h - src/core/lib/channel/http_server_filter.h + - src/core/lib/channel/message_size_filter.h - src/core/lib/compression/algorithm_metadata.h - src/core/lib/compression/message_compress.h - src/core/lib/debug/trace.h @@ -243,6 +244,7 @@ filegroups: - src/core/lib/channel/handshaker.c - src/core/lib/channel/http_client_filter.c - src/core/lib/channel/http_server_filter.c + - src/core/lib/channel/message_size_filter.c - src/core/lib/compression/compression.c - src/core/lib/compression/message_compress.c - src/core/lib/debug/trace.c diff --git a/config.m4 b/config.m4 index 5b47074bcd..c1b2fd9c9b 100644 --- a/config.m4 +++ b/config.m4 @@ -90,6 +90,7 @@ if test "$PHP_GRPC" != "no"; then src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ + src/core/lib/channel/message_size_filter.c \ src/core/lib/compression/compression.c \ src/core/lib/compression/message_compress.c \ src/core/lib/debug/trace.c \ diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 02a500f626..1046f503db 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -261,6 +261,7 @@ Pod::Spec.new do |s| 'src/core/lib/channel/handshaker.h', 'src/core/lib/channel/http_client_filter.h', 'src/core/lib/channel/http_server_filter.h', + 'src/core/lib/channel/message_size_filter.h', 'src/core/lib/compression/algorithm_metadata.h', 'src/core/lib/compression/message_compress.h', 'src/core/lib/debug/trace.h', @@ -420,6 +421,7 @@ Pod::Spec.new do |s| 'src/core/lib/channel/handshaker.c', 'src/core/lib/channel/http_client_filter.c', 'src/core/lib/channel/http_server_filter.c', + 'src/core/lib/channel/message_size_filter.c', 'src/core/lib/compression/compression.c', 'src/core/lib/compression/message_compress.c', 'src/core/lib/debug/trace.c', @@ -622,6 +624,7 @@ Pod::Spec.new do |s| 'src/core/lib/channel/handshaker.h', 'src/core/lib/channel/http_client_filter.h', 'src/core/lib/channel/http_server_filter.h', + 'src/core/lib/channel/message_size_filter.h', 'src/core/lib/compression/algorithm_metadata.h', 'src/core/lib/compression/message_compress.h', 'src/core/lib/debug/trace.h', diff --git a/grpc.gemspec b/grpc.gemspec index 9c82db8ee3..c69988f669 100755 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -180,6 +180,7 @@ Gem::Specification.new do |s| 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 ) + s.files += %w( src/core/lib/channel/message_size_filter.h ) s.files += %w( src/core/lib/compression/algorithm_metadata.h ) s.files += %w( src/core/lib/compression/message_compress.h ) s.files += %w( src/core/lib/debug/trace.h ) @@ -339,6 +340,7 @@ Gem::Specification.new do |s| 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 ) + s.files += %w( src/core/lib/channel/message_size_filter.c ) s.files += %w( src/core/lib/compression/compression.c ) s.files += %w( src/core/lib/compression/message_compress.c ) s.files += %w( src/core/lib/debug/trace.c ) diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h index e5a82883be..8b1ba63494 100644 --- a/include/grpc/impl/codegen/grpc_types.h +++ b/include/grpc/impl/codegen/grpc_types.h @@ -127,7 +127,11 @@ typedef struct { connection. Int valued. */ #define GRPC_ARG_MAX_CONCURRENT_STREAMS "grpc.max_concurrent_streams" /** Maximum message length that the channel can receive. Int valued, bytes. */ -#define GRPC_ARG_MAX_MESSAGE_LENGTH "grpc.max_message_length" +#define GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH "grpc.max_receive_message_length" +/** Backward compatibility. */ +#define GRPC_ARG_MAX_MESSAGE_LENGTH GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH +/** Maximum message length that the channel can send. Int valued, bytes. */ +#define GRPC_ARG_MAX_SEND_MESSAGE_LENGTH "grpc.max_send_message_length" /** Initial sequence number for http2 transports. Int valued. */ #define GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER \ "grpc.http2.initial_sequence_number" diff --git a/package.xml b/package.xml index be8c08d766..c2c8e0cdb5 100644 --- a/package.xml +++ b/package.xml @@ -188,6 +188,7 @@ <file baseinstalldir="/" name="src/core/lib/channel/handshaker.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/channel/http_client_filter.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/channel/http_server_filter.h" role="src" /> + <file baseinstalldir="/" name="src/core/lib/channel/message_size_filter.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/compression/algorithm_metadata.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/compression/message_compress.h" role="src" /> <file baseinstalldir="/" name="src/core/lib/debug/trace.h" role="src" /> @@ -347,6 +348,7 @@ <file baseinstalldir="/" name="src/core/lib/channel/handshaker.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/channel/http_client_filter.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/channel/http_server_filter.c" role="src" /> + <file baseinstalldir="/" name="src/core/lib/channel/message_size_filter.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/compression/compression.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/compression/message_compress.c" role="src" /> <file baseinstalldir="/" name="src/core/lib/debug/trace.c" role="src" /> 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..264c07ca8e --- /dev/null +++ b/src/core/lib/channel/message_size_filter.c @@ -0,0 +1,175 @@ +/* + * 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 <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + +/* the protobuf library will (by default) start warning at 100megs */ +#define DEFAULT_MAX_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 { + size_t max_send_size; + size_t 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)->length > chand->max_recv_size) { + char* message_string; + gpr_asprintf(&message_string, "Received message larger than max (%lu)", + chand->max_recv_size); + gpr_slice message = gpr_slice_from_copied_string(message_string); + gpr_free(message_string); + grpc_call_element_send_cancel_with_message( + exec_ctx, elem, GRPC_STATUS_INVALID_ARGUMENT, &message); + } + // Invoke the next callback. + calld->next_recv_message_ready->cb( + exec_ctx, calld->next_recv_message_ready->cb_arg, error); +} + +// 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 > chand->max_send_size) { + char* message_string; + gpr_asprintf(&message_string, "Sent message larger than max (%lu)", + chand->max_send_size); + gpr_slice message = gpr_slice_from_copied_string(message_string); + gpr_free(message_string); + grpc_call_element_send_cancel_with_message( + exec_ctx, elem, GRPC_STATUS_INVALID_ARGUMENT, &message); + } + // Inject callback for receiving a message. + if (op->recv_message_ready != NULL && chand->max_recv_size > 0) { + 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)); + 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) { + if (args->channel_args->args[i].type != GRPC_ARG_INTEGER) { + gpr_log(GPR_ERROR, "%s ignored: it must be an integer", + GRPC_ARG_MAX_SEND_MESSAGE_LENGTH); + } else if (args->channel_args->args[i].value.integer < 0) { + gpr_log(GPR_ERROR, "%s ignored: it must be >= 0", + GRPC_ARG_MAX_SEND_MESSAGE_LENGTH); + } else { + chand->max_send_size = + (size_t)args->channel_args->args[i].value.integer; + } + } + if (strcmp(args->channel_args->args[i].key, + GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH) == 0) { + if (args->channel_args->args[i].type != GRPC_ARG_INTEGER) { + gpr_log(GPR_ERROR, "%s ignored: it must be an integer", + GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH); + } else if (args->channel_args->args[i].value.integer < 0) { + gpr_log(GPR_ERROR, "%s ignored: it must be >= 0", + GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH); + } else { + chand->max_recv_size = + (size_t)args->channel_args->args[i].value.integer; + } + } + } +} + +// 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..367a322b37 --- /dev/null +++ b/src/core/lib/channel/message_size_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_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/surface/call.c b/src/core/lib/surface/call.c index 772681109a..6717024ab2 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -1160,17 +1160,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) && diff --git a/src/core/lib/surface/channel.c b/src/core/lib/surface/channel.c index 6d2b1c4935..0e8e5bbfcd 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); @@ -371,7 +356,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 4c62974346..23cc8656ca 100644 --- a/src/core/lib/surface/channel.h +++ b/src/core/lib/surface/channel.h @@ -64,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/init.c b/src/core/lib/surface/init.c index 5397913a21..1f2769f871 100644 --- a/src/core/lib/surface/init.c +++ b/src/core/lib/surface/init.c @@ -45,6 +45,7 @@ #include "src/core/lib/channel/connected_channel.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/executor.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_CHANNEL, 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); diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index 660e34d742..491e38c4cf 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -84,6 +84,7 @@ CORE_SOURCE_FILES = [ 'src/core/lib/channel/handshaker.c', 'src/core/lib/channel/http_client_filter.c', 'src/core/lib/channel/http_server_filter.c', + 'src/core/lib/channel/message_size_filter.c', 'src/core/lib/compression/compression.c', 'src/core/lib/compression/message_compress.c', 'src/core/lib/debug/trace.c', diff --git a/test/core/end2end/tests/max_message_length.c b/test/core/end2end/tests/max_message_length.c index 08d326ab4d..c8e0748f00 100644 --- a/test/core/end2end/tests/max_message_length.c +++ b/test/core/end2end/tests/max_message_length.c @@ -100,19 +100,22 @@ static void end_test(grpc_end2end_test_fixture *f) { grpc_completion_queue_destroy(f->cq); } -static void test_max_message_length(grpc_end2end_test_config config) { +static void test_max_message_length(grpc_end2end_test_config config, + bool send_limit) { + gpr_log(GPR_INFO, "testing with send_limit=%d", send_limit); + grpc_end2end_test_fixture f; - grpc_arg server_arg; - grpc_channel_args server_args; + grpc_arg channel_arg; + grpc_channel_args channel_args; grpc_call *c; - grpc_call *s; + grpc_call *s = NULL; cq_verifier *cqv; grpc_op ops[6]; grpc_op *op; gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world"); grpc_byte_buffer *request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); - grpc_byte_buffer *recv_payload; + grpc_byte_buffer *recv_payload = NULL; grpc_metadata_array initial_metadata_recv; grpc_metadata_array trailing_metadata_recv; grpc_metadata_array request_metadata_recv; @@ -123,14 +126,17 @@ static void test_max_message_length(grpc_end2end_test_config config) { size_t details_capacity = 0; int was_cancelled = 2; - server_arg.key = GRPC_ARG_MAX_MESSAGE_LENGTH; - server_arg.type = GRPC_ARG_INTEGER; - server_arg.value.integer = 5; + channel_arg.key = send_limit ? GRPC_ARG_MAX_SEND_MESSAGE_LENGTH + : GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH; + channel_arg.type = GRPC_ARG_INTEGER; + channel_arg.value.integer = 5; - server_args.num_args = 1; - server_args.args = &server_arg; + channel_args.num_args = 1; + channel_args.args = &channel_arg; - f = begin_test(config, "test_max_message_length", NULL, &server_args); + f = begin_test(config, "test_max_message_length", + send_limit ? &channel_args : NULL, + send_limit ? NULL : &channel_args); cqv = cq_verifier_create(f.cq); c = grpc_channel_create_call(f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq, @@ -175,6 +181,12 @@ static void test_max_message_length(grpc_end2end_test_config config) { error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL); GPR_ASSERT(GRPC_CALL_OK == error); + if (send_limit) { + cq_expect_completion(cqv, tag(1), 1); + cq_verify(cqv); + goto done; + } + error = grpc_server_request_call(f.server, &s, &call_details, &request_metadata_recv, f.cq, f.cq, tag(101)); @@ -201,11 +213,12 @@ static void test_max_message_length(grpc_end2end_test_config config) { cq_expect_completion(cqv, tag(1), 1); cq_verify(cqv); - GPR_ASSERT(status != GRPC_STATUS_OK); GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr:1234")); GPR_ASSERT(was_cancelled == 1); - GPR_ASSERT(recv_payload == NULL); + +done: + GPR_ASSERT(status == GRPC_STATUS_INVALID_ARGUMENT); gpr_free(details); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -213,9 +226,10 @@ static void test_max_message_length(grpc_end2end_test_config config) { grpc_metadata_array_destroy(&request_metadata_recv); grpc_call_details_destroy(&call_details); grpc_byte_buffer_destroy(request_payload); + grpc_byte_buffer_destroy(recv_payload); grpc_call_destroy(c); - grpc_call_destroy(s); + if (s != NULL) grpc_call_destroy(s); cq_verifier_destroy(cqv); @@ -224,7 +238,8 @@ static void test_max_message_length(grpc_end2end_test_config config) { } void max_message_length(grpc_end2end_test_config config) { - test_max_message_length(config); + test_max_message_length(config, true); + test_max_message_length(config, false); } void max_message_length_pre_init(void) {} diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 50cebfa84e..d8e1dbc613 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -799,6 +799,7 @@ src/core/lib/channel/context.h \ src/core/lib/channel/handshaker.h \ src/core/lib/channel/http_client_filter.h \ src/core/lib/channel/http_server_filter.h \ +src/core/lib/channel/message_size_filter.h \ src/core/lib/compression/algorithm_metadata.h \ src/core/lib/compression/message_compress.h \ src/core/lib/debug/trace.h \ @@ -958,6 +959,7 @@ src/core/lib/channel/connected_channel.c \ src/core/lib/channel/handshaker.c \ src/core/lib/channel/http_client_filter.c \ src/core/lib/channel/http_server_filter.c \ +src/core/lib/channel/message_size_filter.c \ src/core/lib/compression/compression.c \ src/core/lib/compression/message_compress.c \ src/core/lib/debug/trace.c \ diff --git a/tools/run_tests/sources_and_headers.json b/tools/run_tests/sources_and_headers.json index 84f7da8cb9..e3f43da8c7 100644 --- a/tools/run_tests/sources_and_headers.json +++ b/tools/run_tests/sources_and_headers.json @@ -5825,6 +5825,7 @@ "src/core/lib/channel/handshaker.h", "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.h", + "src/core/lib/channel/message_size_filter.h", "src/core/lib/compression/algorithm_metadata.h", "src/core/lib/compression/message_compress.h", "src/core/lib/debug/trace.h", @@ -5923,6 +5924,8 @@ "src/core/lib/channel/http_client_filter.h", "src/core/lib/channel/http_server_filter.c", "src/core/lib/channel/http_server_filter.h", + "src/core/lib/channel/message_size_filter.c", + "src/core/lib/channel/message_size_filter.h", "src/core/lib/compression/algorithm_metadata.h", "src/core/lib/compression/compression.c", "src/core/lib/compression/message_compress.c", diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj b/vsprojects/vcxproj/grpc/grpc.vcxproj index 298887e915..e8e3081897 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj @@ -308,6 +308,7 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\handshaker.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_client_filter.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.h" /> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\algorithm_metadata.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\debug\trace.h" /> @@ -478,6 +479,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.c"> diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters index 539dbad360..f8d97e9ab0 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters @@ -28,6 +28,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c"> <Filter>src\core\lib\channel</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.c"> + <Filter>src\core\lib\channel</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c"> <Filter>src\core\lib\compression</Filter> </ClCompile> @@ -692,6 +695,9 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.h"> <Filter>src\core\lib\channel</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.h"> + <Filter>src\core\lib\channel</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\algorithm_metadata.h"> <Filter>src\core\lib\compression</Filter> </ClInclude> diff --git a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj index 84a71d217d..2e5fecc490 100644 --- a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj +++ b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj @@ -198,6 +198,7 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\handshaker.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_client_filter.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.h" /> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\algorithm_metadata.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\debug\trace.h" /> @@ -321,6 +322,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.c"> 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 18fe926405..bfa0cdadcf 100644 --- a/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_test_util/grpc_test_util.vcxproj.filters @@ -76,6 +76,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c"> <Filter>src\core\lib\channel</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.c"> + <Filter>src\core\lib\channel</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c"> <Filter>src\core\lib\compression</Filter> </ClCompile> @@ -470,6 +473,9 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.h"> <Filter>src\core\lib\channel</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.h"> + <Filter>src\core\lib\channel</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\algorithm_metadata.h"> <Filter>src\core\lib\compression</Filter> </ClInclude> diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj index 6a89065654..f2bcac82ac 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj @@ -297,6 +297,7 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\handshaker.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_client_filter.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.h" /> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\algorithm_metadata.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.h" /> <ClInclude Include="$(SolutionDir)\..\src\core\lib\debug\trace.h" /> @@ -445,6 +446,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\message_compress.c"> diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters index 8325d910ec..679832388b 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters @@ -31,6 +31,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.c"> <Filter>src\core\lib\channel</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.c"> + <Filter>src\core\lib\channel</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\lib\compression\compression.c"> <Filter>src\core\lib\compression</Filter> </ClCompile> @@ -599,6 +602,9 @@ <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\http_server_filter.h"> <Filter>src\core\lib\channel</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\src\core\lib\channel\message_size_filter.h"> + <Filter>src\core\lib\channel</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\src\core\lib\compression\algorithm_metadata.h"> <Filter>src\core\lib\compression</Filter> </ClInclude> -- GitLab