Merge pull request #13697 from vjpai/nostdthread

Stop using std::thread in C++ library since it can trigger exceptions

6 files changed, 65 insertions, 8 deletions

diff --git a/include/grpc++/impl/codegen/byte_buffer.h b/include/grpc++/impl/codegen/byte_buffer.h
index fe73ce7a83..9c0246e617 100644
--- a/include/grpc++/impl/codegen/byte_buffer.h
+++ b/include/grpc++/impl/codegen/byte_buffer.h
@@ -41,6 +41,8 @@ template <class ServiceType, class RequestType, class ResponseType>
 class RpcMethodHandler;
 template <class ServiceType, class RequestType, class ResponseType>
 class ServerStreamingHandler;
+template <StatusCode code>
+class ErrorMethodHandler;
 template <class R>
 class DeserializeFuncType;
 }  // namespace internal
@@ -107,6 +109,8 @@ class ByteBuffer final {
   friend class internal::RpcMethodHandler;
   template <class ServiceType, class RequestType, class ResponseType>
   friend class internal::ServerStreamingHandler;
+  template <StatusCode code>
+  friend class internal::ErrorMethodHandler;
   template <class R>
   friend class internal::DeserializeFuncType;
 
diff --git a/include/grpc++/impl/codegen/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h
index b8a7862578..452eac6646 100644
--- a/include/grpc++/impl/codegen/completion_queue.h
+++ b/include/grpc++/impl/codegen/completion_queue.h
@@ -78,7 +78,8 @@ template <class ServiceType, class RequestType, class ResponseType>
 class ServerStreamingHandler;
 template <class ServiceType, class RequestType, class ResponseType>
 class BidiStreamingHandler;
-class UnknownMethodHandler;
+template <StatusCode code>
+class ErrorMethodHandler;
 template <class Streamer, bool WriteNeeded>
 class TemplatedBidiStreamingHandler;
 template <class InputMessage, class OutputMessage>
@@ -221,7 +222,8 @@ class CompletionQueue : private GrpcLibraryCodegen {
   friend class ::grpc::internal::ServerStreamingHandler;
   template <class Streamer, bool WriteNeeded>
   friend class ::grpc::internal::TemplatedBidiStreamingHandler;
-  friend class ::grpc::internal::UnknownMethodHandler;
+  template <StatusCode code>
+  friend class ::grpc::internal::ErrorMethodHandler;
   friend class ::grpc::Server;
   friend class ::grpc::ServerContext;
   friend class ::grpc::ServerInterface;
diff --git a/include/grpc++/impl/codegen/method_handler_impl.h b/include/grpc++/impl/codegen/method_handler_impl.h
index c0af4ca130..93b7826e8f 100644
--- a/include/grpc++/impl/codegen/method_handler_impl.h
+++ b/include/grpc++/impl/codegen/method_handler_impl.h
@@ -242,12 +242,14 @@ class SplitServerStreamingHandler
             ServerSplitStreamer<RequestType, ResponseType>, false>(func) {}
 };
 
-/// Handle unknown method by returning UNIMPLEMENTED error.
-class UnknownMethodHandler : public MethodHandler {
+/// General method handler class for errors that prevent real method use
+/// e.g., handle unknown method by returning UNIMPLEMENTED error.
+template <StatusCode code>
+class ErrorMethodHandler : public MethodHandler {
  public:
   template <class T>
   static void FillOps(ServerContext* context, T* ops) {
-    Status status(StatusCode::UNIMPLEMENTED, "");
+    Status status(code, "");
     if (!context->sent_initial_metadata_) {
       ops->SendInitialMetadata(context->initial_metadata_,
                                context->initial_metadata_flags());
@@ -264,9 +266,18 @@ class UnknownMethodHandler : public MethodHandler {
     FillOps(param.server_context, &ops);
     param.call->PerformOps(&ops);
     param.call->cq()->Pluck(&ops);
+    // We also have to destroy any request payload in the handler parameter
+    ByteBuffer* payload = param.request.bbuf_ptr();
+    if (payload != nullptr) {
+      payload->Clear();
+    }
   }
 };
 
+typedef ErrorMethodHandler<StatusCode::UNIMPLEMENTED> UnknownMethodHandler;
+typedef ErrorMethodHandler<StatusCode::RESOURCE_EXHAUSTED>
+    ResourceExhaustedHandler;
+
 }  // namespace internal
 }  // namespace grpc
 
diff --git a/include/grpc++/impl/codegen/server_context.h b/include/grpc++/impl/codegen/server_context.h
index a2d6967bf8..9f20335a2a 100644
--- a/include/grpc++/impl/codegen/server_context.h
+++ b/include/grpc++/impl/codegen/server_context.h
@@ -63,7 +63,8 @@ template <class ServiceType, class RequestType, class ResponseType>
 class ServerStreamingHandler;
 template <class ServiceType, class RequestType, class ResponseType>
 class BidiStreamingHandler;
-class UnknownMethodHandler;
+template <StatusCode code>
+class ErrorMethodHandler;
 template <class Streamer, bool WriteNeeded>
 class TemplatedBidiStreamingHandler;
 class Call;
@@ -255,7 +256,8 @@ class ServerContext {
   friend class ::grpc::internal::ServerStreamingHandler;
   template <class Streamer, bool WriteNeeded>
   friend class ::grpc::internal::TemplatedBidiStreamingHandler;
-  friend class ::grpc::internal::UnknownMethodHandler;
+  template <StatusCode code>
+  friend class ::grpc::internal::ErrorMethodHandler;
   friend class ::grpc::ClientContext;
 
   /// Prevent copying.
diff --git a/include/grpc++/server.h b/include/grpc++/server.h
index 01c4a60d21..cf590185d1 100644
--- a/include/grpc++/server.h
+++ b/include/grpc++/server.h
@@ -35,6 +35,7 @@
 #include <grpc++/support/config.h>
 #include <grpc++/support/status.h>
 #include <grpc/compression.h>
+#include <grpc/support/thd.h>
 
 struct grpc_server;
 
@@ -138,10 +139,20 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen {
   ///
   /// \param sync_cq_timeout_msec The timeout to use when calling AsyncNext() on
   /// server completion queues passed via sync_server_cqs param.
+  ///
+  /// \param thread_creator The thread creation function for the sync
+  /// server. Typically gpr_thd_new
+  ///
+  /// \param thread_joiner The thread joining function for the sync
+  /// server. Typically gpr_thd_join
   Server(int max_message_size, ChannelArguments* args,
          std::shared_ptr<std::vector<std::unique_ptr<ServerCompletionQueue>>>
              sync_server_cqs,
-         int min_pollers, int max_pollers, int sync_cq_timeout_msec);
+         int min_pollers, int max_pollers, int sync_cq_timeout_msec,
+         std::function<int(gpr_thd_id*, const char*, void (*)(void*), void*,
+                           const gpr_thd_options*)>
+             thread_creator,
+         std::function<void(gpr_thd_id)> thread_joiner);
 
   /// Register a service. This call does not take ownership of the service.
   /// The service must exist for the lifetime of the Server instance.
@@ -220,6 +231,14 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen {
 
   std::unique_ptr<HealthCheckServiceInterface> health_check_service_;
   bool health_check_service_disabled_;
+
+  std::function<int(gpr_thd_id*, const char*, void (*)(void*), void*,
+                    const gpr_thd_options*)>
+      thread_creator_;
+  std::function<void(gpr_thd_id)> thread_joiner_;
+
+  // A special handler for resource exhausted in sync case
+  std::unique_ptr<internal::MethodHandler> resource_exhausted_handler_;
 };
 
 }  // namespace grpc
diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h
index e2bae4b41f..25bbacbbc7 100644
--- a/include/grpc++/server_builder.h
+++ b/include/grpc++/server_builder.h
@@ -20,6 +20,7 @@
 #define GRPCXX_SERVER_BUILDER_H
 
 #include <climits>
+#include <functional>
 #include <map>
 #include <memory>
 #include <vector>
@@ -30,6 +31,7 @@
 #include <grpc++/support/config.h>
 #include <grpc/compression.h>
 #include <grpc/support/cpu.h>
+#include <grpc/support/thd.h>
 #include <grpc/support/useful.h>
 #include <grpc/support/workaround_list.h>
 
@@ -47,6 +49,7 @@ class Service;
 
 namespace testing {
 class ServerBuilderPluginTest;
+class ServerBuilderThreadCreatorOverrideTest;
 }  // namespace testing
 
 /// A builder class for the creation and startup of \a grpc::Server instances.
@@ -213,6 +216,17 @@ class ServerBuilder {
 
  private:
   friend class ::grpc::testing::ServerBuilderPluginTest;
+  friend class ::grpc::testing::ServerBuilderThreadCreatorOverrideTest;
+
+  ServerBuilder& SetThreadFunctions(
+      std::function<int(gpr_thd_id*, const char*, void (*)(void*), void*,
+                        const gpr_thd_options*)>
+          thread_creator,
+      std::function<void(gpr_thd_id)> thread_joiner) {
+    thread_creator_ = thread_creator;
+    thread_joiner_ = thread_joiner;
+    return *this;
+  }
 
   struct Port {
     grpc::string addr;
@@ -272,6 +286,11 @@ class ServerBuilder {
     grpc_compression_algorithm algorithm;
   } maybe_default_compression_algorithm_;
   uint32_t enabled_compression_algorithms_bitset_;
+
+  std::function<int(gpr_thd_id*, const char*, void (*)(void*), void*,
+                    const gpr_thd_options*)>
+      thread_creator_;
+  std::function<void(gpr_thd_id)> thread_joiner_;
 };
 
 }  // namespace grpc