Merge pull request #6481 from dgquintas/compression_md_level_bis

Allow servers to select compression level via initial MD & overall compression cleanup

3 files changed, 66 insertions, 19 deletions

diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc
index 32c7794ade..d3e5ce0c4a 100644
--- a/src/cpp/client/client_context.cc
+++ b/src/cpp/client/client_context.cc
@@ -33,15 +33,14 @@
 
 #include <grpc++/client_context.h>
 
-#include <grpc++/security/credentials.h>
-#include <grpc++/server_context.h>
-#include <grpc++/support/time.h>
 #include <grpc/compression.h>
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/string_util.h>
 
-#include "src/core/lib/channel/compress_filter.h"
+#include <grpc++/security/credentials.h>
+#include <grpc++/server_context.h>
+#include <grpc++/support/time.h>
 
 namespace grpc {
 
@@ -112,7 +111,7 @@ void ClientContext::set_compression_algorithm(
     abort();
   }
   GPR_ASSERT(algorithm_name != nullptr);
-  AddMetadata(GRPC_COMPRESS_REQUEST_ALGORITHM_KEY, algorithm_name);
+  AddMetadata(GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY, algorithm_name);
 }
 
 void ClientContext::TryCancel() {
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index 54feac3982..279981744a 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -37,6 +37,8 @@
 #include <grpc++/server.h>
 #include <grpc/support/cpu.h>
 #include <grpc/support/log.h>
+
+#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<ServerBuilderPlugin> 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<ServerCompletionQueue> ServerBuilder::AddCompletionQueue(
@@ -67,35 +75,65 @@ std::unique_ptr<ServerCompletionQueue> ServerBuilder::AddCompletionQueue(
   return std::unique_ptr<ServerCompletionQueue>(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<ServerBuilderOption> option) {
+ServerBuilder& ServerBuilder::SetOption(
+    std::unique_ptr<ServerBuilderOption> 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<ServerCredentials> 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<ServerCredentials> creds,
+    int* selected_port) {
   Port port = {addr, creds, selected_port};
   ports_.push_back(port);
+  return *this;
 }
 
 std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
@@ -128,7 +166,15 @@ std::unique_ptr<Server> 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);
+              enabled_compression_algorithms_bitset_);
+  if (maybe_default_compression_level_.is_set) {
+    args.SetInt(GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL,
+                maybe_default_compression_level_.level);
+  }
+  if (maybe_default_compression_algorithm_.is_set) {
+    args.SetInt(GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM,
+                maybe_default_compression_algorithm_.algorithm);
+  }
   std::unique_ptr<Server> 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 <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 
-#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 {