From 9e9f7b62c706245cd74de1a7f01c03ba0bd96b0b Mon Sep 17 00:00:00 2001 From: David Garcia Quintas Date: Mon, 16 May 2016 19:12:12 -0700 Subject: Refactorings and renamings related to compression. Also added levels to the channel args options. --- src/cpp/server/server_builder.cc | 7 +++++++ src/cpp/server/server_context.cc | 6 ++++-- 2 files changed, 11 insertions(+), 2 deletions(-) (limited to 'src/cpp/server') diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index 61f0f6ae2a..e6050c3b52 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -125,6 +125,13 @@ std::unique_ptr ServerBuilder::BuildAndStart() { } args.SetInt(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET, compression_options_.enabled_algorithms_bitset); + if (compression_options_.default_level.is_set) { + args.SetInt(GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL, + compression_options_.default_level.level); + } else if (compression_options_.default_algorithm.is_set) { + args.SetInt(GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM, + compression_options_.default_algorithm.algorithm); + } std::unique_ptr server( new Server(thread_pool.release(), true, max_message_size_, &args)); ServerInitializer* initializer = server->initializer(); diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index 204fef1b09..43117fd1e9 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -42,7 +42,6 @@ #include #include -#include "src/core/lib/channel/compress_filter.h" #include "src/core/lib/surface/call.h" namespace grpc { @@ -196,6 +195,9 @@ bool ServerContext::IsCancelled() const { } void ServerContext::set_compression_level(grpc_compression_level level) { + // TODO(dgq): get rid of grpc_call_compression_for_level and propagate the + // compression level by adding a new argument to + // CallOpSendInitialMetadata::SendInitialMetadata. const grpc_compression_algorithm algorithm_for_level = grpc_call_compression_for_level(call_, level); set_compression_algorithm(algorithm_for_level); @@ -210,7 +212,7 @@ void ServerContext::set_compression_algorithm( abort(); } GPR_ASSERT(algorithm_name != NULL); - AddInitialMetadata(GRPC_COMPRESS_REQUEST_ALGORITHM_KEY, algorithm_name); + AddInitialMetadata(GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY, algorithm_name); } grpc::string ServerContext::peer() const { -- cgit v1.2.3 From 2d02456e785053735172186868300d0f4dde3d9e Mon Sep 17 00:00:00 2001 From: David Garcia Quintas Date: Tue, 17 May 2016 23:24:39 -0700 Subject: got rid of grpc_compression_options --- include/grpc++/server_builder.h | 48 +++++++++++----- include/grpc/compression.h | 14 ----- include/grpc/impl/codegen/compression_types.h | 26 --------- src/core/lib/compression/compression_algorithm.c | 22 -------- src/cpp/server/server_builder.cc | 71 ++++++++++++++++++------ test/core/compression/compression_test.c | 30 ---------- 6 files changed, 90 insertions(+), 121 deletions(-) (limited to 'src/cpp/server') diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h index ad629521cb..c48c86d2b8 100644 --- a/include/grpc++/server_builder.h +++ b/include/grpc++/server_builder.h @@ -66,29 +66,43 @@ class ServerBuilder { /// The service must exist for the lifetime of the \a Server instance returned /// by \a BuildAndStart(). /// Matches requests with any :authority - void RegisterService(Service* service); + ServerBuilder& RegisterService(Service* service); /// Register a generic service. /// Matches requests with any :authority - void RegisterAsyncGenericService(AsyncGenericService* service); + ServerBuilder& RegisterAsyncGenericService(AsyncGenericService* service); /// Register a service. This call does not take ownership of the service. /// The service must exist for the lifetime of the \a Server instance returned /// by BuildAndStart(). /// Only matches requests with :authority \a host - void RegisterService(const grpc::string& host, Service* service); + ServerBuilder& RegisterService(const grpc::string& host, Service* service); /// Set max message size in bytes. - void SetMaxMessageSize(int max_message_size) { + ServerBuilder& SetMaxMessageSize(int max_message_size) { max_message_size_ = max_message_size; + return *this; } - /// Set the compression options to be used by the server. - void SetCompressionOptions(const grpc_compression_options& options) { - compression_options_ = options; - } + /// Set the support status for compression algorithms. All algorithms are + /// enabled by default. + /// + /// Incoming calls compressed with an unsupported algorithm will fail with + /// GRPC_STATUS_UNIMPLEMENTED. + ServerBuilder& SetCompressionAlgorithmSupportStatus( + grpc_compression_algorithm algorithm, bool enabled); + + /// The default compression level to use for all channel calls in the + /// absence of a call-specific level. + ServerBuilder& SetDefaultCompressionLevel(grpc_compression_level level); + + /// The default compression algorithm to use for all channel calls in the + /// absence of a call-specific level. Note that it overrides any compression + /// level set by \a SetDefaultCompressionLevel. + ServerBuilder& SetDefaultCompressionAlgorithm( + grpc_compression_algorithm algorithm); - void SetOption(std::unique_ptr option); + ServerBuilder& SetOption(std::unique_ptr option); /// Tries to bind \a server to the given \a addr. /// @@ -101,9 +115,9 @@ class ServerBuilder { /// number. \a nullptr otherwise. /// // TODO(dgq): the "port" part seems to be a misnomer. - void AddListeningPort(const grpc::string& addr, - std::shared_ptr creds, - int* selected_port = nullptr); + ServerBuilder& AddListeningPort(const grpc::string& addr, + std::shared_ptr creds, + int* selected_port = nullptr); /// Add a completion queue for handling asynchronous services /// Caller is required to keep this completion queue live until @@ -136,7 +150,6 @@ class ServerBuilder { }; int max_message_size_; - grpc_compression_options compression_options_; std::vector> options_; std::vector> services_; std::vector ports_; @@ -144,6 +157,15 @@ class ServerBuilder { std::shared_ptr creds_; std::map> plugins_; AsyncGenericService* generic_service_; + struct { + bool is_set; + grpc_compression_level level; + } maybe_default_compression_level_; + struct { + bool is_set; + grpc_compression_algorithm algorithm; + } maybe_default_compression_algorithm_; + uint32_t enabled_compression_algorithms_bitset_; }; } // namespace grpc diff --git a/include/grpc/compression.h b/include/grpc/compression.h index 04816b9f3a..3eba00c986 100644 --- a/include/grpc/compression.h +++ b/include/grpc/compression.h @@ -55,20 +55,6 @@ GRPCAPI int grpc_compression_algorithm_parse( GRPCAPI int grpc_compression_algorithm_name( grpc_compression_algorithm algorithm, char **name); -GRPCAPI void grpc_compression_options_init(grpc_compression_options *opts); - -/** Mark \a algorithm as enabled in \a opts. */ -GRPCAPI void grpc_compression_options_enable_algorithm( - grpc_compression_options *opts, grpc_compression_algorithm algorithm); - -/** Mark \a algorithm as disabled in \a opts. */ -GRPCAPI void grpc_compression_options_disable_algorithm( - grpc_compression_options *opts, grpc_compression_algorithm algorithm); - -/** Returns true if \a algorithm is marked as enabled in \a opts. */ -GRPCAPI int grpc_compression_options_is_algorithm_enabled( - const grpc_compression_options *opts, grpc_compression_algorithm algorithm); - #ifdef __cplusplus } #endif diff --git a/include/grpc/impl/codegen/compression_types.h b/include/grpc/impl/codegen/compression_types.h index 7cf68bfaf8..19c2cefcf4 100644 --- a/include/grpc/impl/codegen/compression_types.h +++ b/include/grpc/impl/codegen/compression_types.h @@ -74,32 +74,6 @@ typedef enum { GRPC_COMPRESS_LEVEL_COUNT } grpc_compression_level; -typedef struct grpc_compression_options { - /** All algs are enabled by default. This option corresponds to the channel - * argument key behind \a GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET - */ - uint32_t enabled_algorithms_bitset; - - /** The default channel compression level. It'll be used in the absence of - * call specific settings. This option corresponds to the channel argument key - * behind \a GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL. If present, takes - * precedence over \a default_algorithm. - * TODO(dgq): currently only available for server channels. */ - struct { - bool is_set; - grpc_compression_algorithm level; - } default_level; - - /** The default channel compression algorithm. It'll be used in the absence of - * call specific settings. This option corresponds to the channel argument key - * behind \a GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM. */ - struct { - bool is_set; - grpc_compression_algorithm algorithm; - } default_algorithm; - -} grpc_compression_options; - #ifdef __cplusplus } #endif diff --git a/src/core/lib/compression/compression_algorithm.c b/src/core/lib/compression/compression_algorithm.c index c7f834d989..6cbdbd81b0 100644 --- a/src/core/lib/compression/compression_algorithm.c +++ b/src/core/lib/compression/compression_algorithm.c @@ -124,25 +124,3 @@ grpc_mdelem *grpc_compression_encoding_mdelem( } return NULL; } - -void grpc_compression_options_init(grpc_compression_options *opts) { - opts->enabled_algorithms_bitset = (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1; - opts->default_level.is_set = false; - opts->default_algorithm.is_set = false; -} - -void grpc_compression_options_enable_algorithm( - grpc_compression_options *opts, grpc_compression_algorithm algorithm) { - GPR_BITSET(&opts->enabled_algorithms_bitset, algorithm); -} - -void grpc_compression_options_disable_algorithm( - grpc_compression_options *opts, grpc_compression_algorithm algorithm) { - GPR_BITCLEAR(&opts->enabled_algorithms_bitset, algorithm); -} - -int grpc_compression_options_is_algorithm_enabled( - const grpc_compression_options *opts, - grpc_compression_algorithm algorithm) { - return GPR_BITGET(opts->enabled_algorithms_bitset, algorithm); -} diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index e6050c3b52..bbea9355ba 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -37,6 +37,8 @@ #include #include #include + +#include "include/grpc/support/useful.h" #include "src/cpp/server/thread_pool_interface.h" namespace grpc { @@ -52,12 +54,18 @@ static void do_plugin_list_init(void) { ServerBuilder::ServerBuilder() : max_message_size_(-1), generic_service_(nullptr) { - grpc_compression_options_init(&compression_options_); gpr_once_init(&once_init_plugin_list, do_plugin_list_init); for (auto factory : (*g_plugin_factory_list)) { std::unique_ptr plugin = factory(); plugins_[plugin->name()] = std::move(plugin); } + // all compression algorithms enabled by default. + enabled_compression_algorithms_bitset_ = + (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1; + memset(&maybe_default_compression_level_, 0, + sizeof(maybe_default_compression_level_)); + memset(&maybe_default_compression_algorithm_, 0, + sizeof(maybe_default_compression_algorithm_)); } std::unique_ptr ServerBuilder::AddCompletionQueue() { @@ -66,35 +74,65 @@ std::unique_ptr ServerBuilder::AddCompletionQueue() { return std::unique_ptr(cq); } -void ServerBuilder::RegisterService(Service* service) { +ServerBuilder& ServerBuilder::RegisterService(Service* service) { services_.emplace_back(new NamedService(service)); + return *this; } -void ServerBuilder::RegisterService(const grpc::string& addr, - Service* service) { +ServerBuilder& ServerBuilder::RegisterService(const grpc::string& addr, + Service* service) { services_.emplace_back(new NamedService(addr, service)); + return *this; } -void ServerBuilder::RegisterAsyncGenericService(AsyncGenericService* service) { +ServerBuilder& ServerBuilder::RegisterAsyncGenericService( + AsyncGenericService* service) { if (generic_service_) { gpr_log(GPR_ERROR, "Adding multiple AsyncGenericService is unsupported for now. " "Dropping the service %p", service); - return; + } else { + generic_service_ = service; } - generic_service_ = service; + return *this; } -void ServerBuilder::SetOption(std::unique_ptr option) { +ServerBuilder& ServerBuilder::SetOption( + std::unique_ptr option) { options_.push_back(std::move(option)); + return *this; +} + +ServerBuilder& ServerBuilder::SetCompressionAlgorithmSupportStatus( + grpc_compression_algorithm algorithm, bool enabled) { + if (enabled) { + GPR_BITSET(&enabled_compression_algorithms_bitset_, algorithm); + } else { + GPR_BITCLEAR(&enabled_compression_algorithms_bitset_, algorithm); + } + return *this; } -void ServerBuilder::AddListeningPort(const grpc::string& addr, - std::shared_ptr creds, - int* selected_port) { +ServerBuilder& ServerBuilder::SetDefaultCompressionLevel( + grpc_compression_level level) { + maybe_default_compression_level_.level = level; + return *this; +} + +ServerBuilder& ServerBuilder::SetDefaultCompressionAlgorithm( + grpc_compression_algorithm algorithm) { + maybe_default_compression_algorithm_.is_set = true; + maybe_default_compression_algorithm_.algorithm = algorithm; + return *this; +} + +ServerBuilder& ServerBuilder::AddListeningPort( + const grpc::string& addr, std::shared_ptr creds, + int* selected_port) { Port port = {addr, creds, selected_port}; ports_.push_back(port); + return *this; } std::unique_ptr ServerBuilder::BuildAndStart() { @@ -124,13 +162,14 @@ std::unique_ptr ServerBuilder::BuildAndStart() { args.SetInt(GRPC_ARG_MAX_MESSAGE_LENGTH, max_message_size_); } args.SetInt(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET, - compression_options_.enabled_algorithms_bitset); - if (compression_options_.default_level.is_set) { + enabled_compression_algorithms_bitset_); + if (maybe_default_compression_level_.is_set) { args.SetInt(GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL, - compression_options_.default_level.level); - } else if (compression_options_.default_algorithm.is_set) { + maybe_default_compression_level_.level); + } + if (maybe_default_compression_algorithm_.is_set) { args.SetInt(GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM, - compression_options_.default_algorithm.algorithm); + maybe_default_compression_algorithm_.algorithm); } std::unique_ptr server( new Server(thread_pool.release(), true, max_message_size_, &args)); diff --git a/test/core/compression/compression_test.c b/test/core/compression/compression_test.c index ed6f9e7214..5ee1805222 100644 --- a/test/core/compression/compression_test.c +++ b/test/core/compression/compression_test.c @@ -92,40 +92,10 @@ static void test_compression_algorithm_name(void) { /* the value of "name" is undefined upon failure */ } -static void test_compression_enable_disable_algorithm(void) { - grpc_compression_options options; - grpc_compression_algorithm algorithm; - - gpr_log(GPR_DEBUG, "test_compression_enable_disable_algorithm"); - - grpc_compression_options_init(&options); - for (algorithm = GRPC_COMPRESS_NONE; - algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT; algorithm++) { - /* all algorithms are enabled by default */ - GPR_ASSERT(grpc_compression_options_is_algorithm_enabled(&options, - algorithm) != 0); - } - /* disable one by one */ - for (algorithm = GRPC_COMPRESS_NONE; - algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT; algorithm++) { - grpc_compression_options_disable_algorithm(&options, algorithm); - GPR_ASSERT(grpc_compression_options_is_algorithm_enabled(&options, - algorithm) == 0); - } - /* re-enable one by one */ - for (algorithm = GRPC_COMPRESS_NONE; - algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT; algorithm++) { - grpc_compression_options_enable_algorithm(&options, algorithm); - GPR_ASSERT(grpc_compression_options_is_algorithm_enabled(&options, - algorithm) != 0); - } -} - int main(int argc, char **argv) { grpc_init(); test_compression_algorithm_parse(); test_compression_algorithm_name(); - test_compression_enable_disable_algorithm(); grpc_shutdown(); return 0; -- cgit v1.2.3