From ec0bc8b4ed4760ff0ab1e51d505f1b235fc9d60d Mon Sep 17 00:00:00 2001 From: "Mark D. Roth" Date: Wed, 15 Jun 2016 14:02:57 -0700 Subject: [PATCH] Initial attempt at a C++ API for defining channel filters. --- BUILD | 4 + Makefile | 4 + build.yaml | 2 + include/grpc++/channel_filter.h | 211 ++++++++++++++++++ src/cpp/common/channel_filter.cc | 98 ++++++++ tools/doxygen/Doxyfile.c++ | 1 + tools/doxygen/Doxyfile.c++.internal | 2 + tools/run_tests/sources_and_headers.json | 3 + vsprojects/vcxproj/grpc++/grpc++.vcxproj | 3 + .../vcxproj/grpc++/grpc++.vcxproj.filters | 6 + .../grpc++_unsecure/grpc++_unsecure.vcxproj | 3 + .../grpc++_unsecure.vcxproj.filters | 6 + 12 files changed, 343 insertions(+) create mode 100644 include/grpc++/channel_filter.h create mode 100644 src/cpp/common/channel_filter.cc diff --git a/BUILD b/BUILD index f049e3c4056..2940fed33f2 100644 --- a/BUILD +++ b/BUILD @@ -1247,6 +1247,7 @@ cc_library( "src/cpp/client/generic_stub.cc", "src/cpp/client/insecure_credentials.cc", "src/cpp/common/channel_arguments.cc", + "src/cpp/common/channel_filter.cc", "src/cpp/common/completion_queue.cc", "src/cpp/common/core_codegen.cc", "src/cpp/common/rpc_method.cc", @@ -1269,6 +1270,7 @@ cc_library( hdrs = [ "include/grpc++/alarm.h", "include/grpc++/channel.h", + "include/grpc++/channel_filter.h", "include/grpc++/client_context.h", "include/grpc++/completion_queue.h", "include/grpc++/create_channel.h", @@ -1473,6 +1475,7 @@ cc_library( "src/cpp/client/generic_stub.cc", "src/cpp/client/insecure_credentials.cc", "src/cpp/common/channel_arguments.cc", + "src/cpp/common/channel_filter.cc", "src/cpp/common/completion_queue.cc", "src/cpp/common/core_codegen.cc", "src/cpp/common/rpc_method.cc", @@ -1495,6 +1498,7 @@ cc_library( hdrs = [ "include/grpc++/alarm.h", "include/grpc++/channel.h", + "include/grpc++/channel_filter.h", "include/grpc++/client_context.h", "include/grpc++/completion_queue.h", "include/grpc++/create_channel.h", diff --git a/Makefile b/Makefile index 6eccd06952c..2f88411c95b 100644 --- a/Makefile +++ b/Makefile @@ -3463,6 +3463,7 @@ LIBGRPC++_SRC = \ src/cpp/client/generic_stub.cc \ src/cpp/client/insecure_credentials.cc \ src/cpp/common/channel_arguments.cc \ + src/cpp/common/channel_filter.cc \ src/cpp/common/completion_queue.cc \ src/cpp/common/core_codegen.cc \ src/cpp/common/rpc_method.cc \ @@ -3485,6 +3486,7 @@ LIBGRPC++_SRC = \ PUBLIC_HEADERS_CXX += \ include/grpc++/alarm.h \ include/grpc++/channel.h \ + include/grpc++/channel_filter.h \ include/grpc++/client_context.h \ include/grpc++/completion_queue.h \ include/grpc++/create_channel.h \ @@ -3950,6 +3952,7 @@ LIBGRPC++_UNSECURE_SRC = \ src/cpp/client/generic_stub.cc \ src/cpp/client/insecure_credentials.cc \ src/cpp/common/channel_arguments.cc \ + src/cpp/common/channel_filter.cc \ src/cpp/common/completion_queue.cc \ src/cpp/common/core_codegen.cc \ src/cpp/common/rpc_method.cc \ @@ -3972,6 +3975,7 @@ LIBGRPC++_UNSECURE_SRC = \ PUBLIC_HEADERS_CXX += \ include/grpc++/alarm.h \ include/grpc++/channel.h \ + include/grpc++/channel_filter.h \ include/grpc++/client_context.h \ include/grpc++/completion_queue.h \ include/grpc++/create_channel.h \ diff --git a/build.yaml b/build.yaml index a83ebc595ad..b6b1f758532 100644 --- a/build.yaml +++ b/build.yaml @@ -634,6 +634,7 @@ filegroups: public_headers: - include/grpc++/alarm.h - include/grpc++/channel.h + - include/grpc++/channel_filter.h - include/grpc++/client_context.h - include/grpc++/completion_queue.h - include/grpc++/create_channel.h @@ -693,6 +694,7 @@ filegroups: - src/cpp/client/generic_stub.cc - src/cpp/client/insecure_credentials.cc - src/cpp/common/channel_arguments.cc + - src/cpp/common/channel_filter.cc - src/cpp/common/completion_queue.cc - src/cpp/common/core_codegen.cc - src/cpp/common/rpc_method.cc diff --git a/include/grpc++/channel_filter.h b/include/grpc++/channel_filter.h new file mode 100644 index 00000000000..8067ca9c603 --- /dev/null +++ b/include/grpc++/channel_filter.h @@ -0,0 +1,211 @@ +/* + * + * 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 GRPCXX_CHANNEL_FILTER_H +#define GRPCXX_CHANNEL_FILTER_H + +#include + +#include + +#include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/surface/channel_init.h" + +// +// An interface to define filters. +// +// To define a filter, implement a subclass of each of CallData and +// ChannelData. Then register the filter like this: +// RegisterChannelFilter( +// "name-of-filter", GRPC_SERVER_CHANNEL, INT_MAX); +// + +namespace grpc { + +// Represents call data. +// Note: Must be copyable. +class CallData { + public: + // Do not override the destructor. Instead, put clean-up code in the + // Destroy() method. + virtual ~CallData() {} + + virtual void Destroy() {} + + virtual void StartTransportStreamOp( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_transport_stream_op *op); + + virtual void SetPollsetOrPollsetSet( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_polling_entity *pollent); + + virtual char* GetPeer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem); + + protected: + CallData() {} +}; + +// Represents channel data. +// Note: Must be copyable. +class ChannelData { + public: + // Do not override the destructor. Instead, put clean-up code in the + // Destroy() method. + virtual ~ChannelData() {} + + virtual void Destroy() {} + + virtual void StartTransportOp( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, + grpc_transport_op *op); + + protected: + ChannelData() {} +}; + +namespace internal { + +// Defines static members for passing to C core. +template +class ChannelFilter { + static const size_t call_data_size = sizeof(CallDataType); + + static void InitCallElement( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_call_element_args *args) { + CallDataType* call_data = elem->call_data; + *call_data = CallDataType(); + } + + static void DestroyCallElement( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + const grpc_call_stats *stats, void *and_free_memory) { + CallDataType* call_data = elem->call_data; + // Can't destroy the object here, since it's not allocated by + // itself; instead, it's part of a larger allocation managed by + // C-core. So instead, we call the Destroy() method. + call_data->Destroy(); + } + + static void StartTransportStreamOp( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_transport_stream_op *op) { + CallDataType* call_data = elem->call_data; + call_data->StartTransportStreamOp(exec_ctx, op); + } + + static void SetPollsetOrPollsetSet( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_polling_entity *pollent) { + CallDataType* call_data = elem->call_data; + call_data->SetPollsetOrPollsetSet(exec_ctx, pollent); + } + + static char* GetPeer( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { + CallDataType* call_data = elem->call_data; + return call_data->GetPeer(exec_ctx); + } + + static const size_t channel_data_size = sizeof(ChannelDataType); + + static void InitChannelElement( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, + grpc_channel_element_args *args) { + ChannelDataType* channel_data = elem->channel_data; + *channel_data = ChannelDataType(); + } + + static void DestroyChannelElement( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem) { + ChannelDataType* channel_data = elem->channel_data; + // Can't destroy the object here, since it's not allocated by + // itself; instead, it's part of a larger allocation managed by + // C-core. So instead, we call the Destroy() method. + channel_data->Destroy(); + } + + static void StartTransportOp( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, + grpc_transport_op *op) { + ChannelDataType* channel_data = elem->channel_data; + channel_data->StartTransportOp(exec_ctx, op); + } +}; + +struct FilterRecord { + grpc_channel_stack_type stack_type; + int priority; + grpc_channel_filter filter; +}; +extern std::vector* channel_filters; + +void ChannelFilterPluginInit(); +void ChannelFilterPluginShutdown() {} + +} // namespace internal + +// Registers a new filter. +// Must be called by only one thread at a time. +template +void RegisterChannelFilter(const char* name, + grpc_channel_stack_type stack_type, int priority) { + // If we haven't been called before, initialize channel_filters and + // call grpc_register_plugin(). + if (internal::channel_filters == nullptr) { + grpc_register_plugin(internal::ChannelFilterPluginInit, + internal::ChannelFilterPluginShutdown); + internal::channel_filters = new std::vector(); + } + // Add an entry to channel_filters. The filter will be added when the + // C-core initialization code calls ChannelFilterPluginInit(). + typedef internal::ChannelFilter FilterType; + internal::channel_filters->emplace_back({ + stack_type, priority, { + FilterType::StartTransportStreamOp, + FilterType::StartTransportOp, + FilterType::call_data_size, + FilterType::InitCallElement, + FilterType::SetPollsetOrPollsetSet, + FilterType::DestroyCallElement, + FilterType::channel_data_size, + FilterType::InitChannelElement, + FilterType::DestroyChannelElement, + FilterType::GetPeer, + name}}); +} + +} // namespace grpc + +#endif // GRPCXX_CHANNEL_FILTER_H diff --git a/src/cpp/common/channel_filter.cc b/src/cpp/common/channel_filter.cc new file mode 100644 index 00000000000..9a80195e867 --- /dev/null +++ b/src/cpp/common/channel_filter.cc @@ -0,0 +1,98 @@ +/* + * + * 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 + +#include "src/core/lib/channel/channel_stack.h" + +namespace grpc { + +// +// CallData +// + +void CallData::StartTransportStreamOp( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_transport_stream_op *op) { + grpc_call_next_op(exec_ctx, elem, op); +} + +void CallData::SetPollsetOrPollsetSet( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_polling_entity *pollent) { + grpc_call_stack_ignore_set_pollset_or_pollset_set(exec_ctx, elem, pollent); +} + +char* CallData::GetPeer( + grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { + return grpc_call_next_get_peer(exec_ctx, elem); +} + +// +// ChannelData +// + +void ChannelData::StartTransportOp( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, + grpc_transport_op *op) { + grpc_channel_next_op(exec_ctx, elem, op); +} + +// +// RegisterChannelFilter() +// + +namespace internal { + +std::vector* channel_filters = nullptr; + +namespace { + +bool AppendFilter(grpc_channel_stack_builder* builder, void* arg) { + return grpc_channel_stack_builder_append_filter( + builder, (const grpc_channel_filter *)arg, nullptr, nullptr); +} + +} // namespace + +void ChannelFilterPluginInit() { + for (size_t i = 0; i < channel_filters->size(); ++i) { + FilterRecord& filter = (*channel_filters)[i]; + grpc_channel_init_register_stage(filter.stack_type, filter.priority, + AppendFilter, (void*)&filter.filter); + } +} + +} // namespace internal + +} // namespace grpc diff --git a/tools/doxygen/Doxyfile.c++ b/tools/doxygen/Doxyfile.c++ index 7f9d2df6f6c..e770574cb1f 100644 --- a/tools/doxygen/Doxyfile.c++ +++ b/tools/doxygen/Doxyfile.c++ @@ -762,6 +762,7 @@ WARN_LOGFILE = INPUT = include/grpc++/alarm.h \ include/grpc++/channel.h \ +include/grpc++/channel_filter.h \ include/grpc++/client_context.h \ include/grpc++/completion_queue.h \ include/grpc++/create_channel.h \ diff --git a/tools/doxygen/Doxyfile.c++.internal b/tools/doxygen/Doxyfile.c++.internal index dcf1a4c8c40..a3c4a109264 100644 --- a/tools/doxygen/Doxyfile.c++.internal +++ b/tools/doxygen/Doxyfile.c++.internal @@ -762,6 +762,7 @@ WARN_LOGFILE = INPUT = include/grpc++/alarm.h \ include/grpc++/channel.h \ +include/grpc++/channel_filter.h \ include/grpc++/client_context.h \ include/grpc++/completion_queue.h \ include/grpc++/create_channel.h \ @@ -880,6 +881,7 @@ src/cpp/client/credentials.cc \ src/cpp/client/generic_stub.cc \ src/cpp/client/insecure_credentials.cc \ src/cpp/common/channel_arguments.cc \ +src/cpp/common/channel_filter.cc \ src/cpp/common/completion_queue.cc \ src/cpp/common/core_codegen.cc \ src/cpp/common/rpc_method.cc \ diff --git a/tools/run_tests/sources_and_headers.json b/tools/run_tests/sources_and_headers.json index 4aad52c69d0..00ea2fdc058 100644 --- a/tools/run_tests/sources_and_headers.json +++ b/tools/run_tests/sources_and_headers.json @@ -6497,6 +6497,7 @@ "headers": [ "include/grpc++/alarm.h", "include/grpc++/channel.h", + "include/grpc++/channel_filter.h", "include/grpc++/client_context.h", "include/grpc++/completion_queue.h", "include/grpc++/create_channel.h", @@ -6551,6 +6552,7 @@ "src": [ "include/grpc++/alarm.h", "include/grpc++/channel.h", + "include/grpc++/channel_filter.h", "include/grpc++/client_context.h", "include/grpc++/completion_queue.h", "include/grpc++/create_channel.h", @@ -6606,6 +6608,7 @@ "src/cpp/client/generic_stub.cc", "src/cpp/client/insecure_credentials.cc", "src/cpp/common/channel_arguments.cc", + "src/cpp/common/channel_filter.cc", "src/cpp/common/completion_queue.cc", "src/cpp/common/core_codegen.cc", "src/cpp/common/rpc_method.cc", diff --git a/vsprojects/vcxproj/grpc++/grpc++.vcxproj b/vsprojects/vcxproj/grpc++/grpc++.vcxproj index cb9e41ea22f..b882c302bbb 100644 --- a/vsprojects/vcxproj/grpc++/grpc++.vcxproj +++ b/vsprojects/vcxproj/grpc++/grpc++.vcxproj @@ -260,6 +260,7 @@ + @@ -397,6 +398,8 @@ + + diff --git a/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters b/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters index a9051182b3c..08fffb74b2f 100644 --- a/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters +++ b/vsprojects/vcxproj/grpc++/grpc++.vcxproj.filters @@ -46,6 +46,9 @@ src\cpp\common + + src\cpp\common + src\cpp\common @@ -108,6 +111,9 @@ include\grpc++ + + include\grpc++ + include\grpc++ diff --git a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj index 03be485b297..b5a27f624d1 100644 --- a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj @@ -260,6 +260,7 @@ + @@ -383,6 +384,8 @@ + + diff --git a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters index ba99bc53c8c..68d9a47973d 100644 --- a/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc++_unsecure/grpc++_unsecure.vcxproj.filters @@ -31,6 +31,9 @@ src\cpp\common + + src\cpp\common + src\cpp\common @@ -93,6 +96,9 @@ include\grpc++ + + include\grpc++ + include\grpc++