Catch exceptions in user callbacks

include/grpcpp/impl/codegen/callback_common.h

@@ -35,6 +35,10 @@ class CQCallbackInterface;
 namespace grpc {
 namespace internal {
 
+// The contract on these tags is that they are single-shot. They must be
+// constructed and then fired at exactly one point. There is no expectation
+// that they can be reused without reconstruction.
+
 class CallbackWithStatusTag {
  public:
   // always allocated against a call arena, no memory free required

src/cpp/common/callback_common.cc

@@ -26,8 +26,21 @@
 
 namespace grpc {
 namespace internal {
-
 namespace {
+
+template <class Func, class Arg>
+void CatchingCallback(Func&& func, Arg&& arg) {
+#if GRPC_ALLOW_EXCEPTIONS
+  try {
+    func(arg);
+  } catch (...) {
+    // nothing to return or change here, just don't crash the library
+  }
+#else   // GRPC_ALLOW_EXCEPTIONS
+  func(arg);
+#endif  // GRPC_ALLOW_EXCEPTIONS
+}
+
 class CallbackWithSuccessImpl : public grpc_core::CQCallbackInterface {
  public:
   static void operator delete(void* ptr, std::size_t size) {
@@ -52,8 +65,11 @@ class CallbackWithSuccessImpl : public grpc_core::CQCallbackInterface {
     bool new_ok = ok;
     GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &new_ok));
     GPR_ASSERT(ignored == parent_->ops());
-    func_(ok);
-    func_ = nullptr;  // release the function
+
+    // Last use of func_ or ok, so ok to move them out for rvalue call above
+    CatchingCallback(std::move(func_), std::move(ok));
+
+    func_ = nullptr;  // reset to clear this out for sure
     grpc_call_unref(call_);
   }
 
@@ -88,8 +104,10 @@ class CallbackWithStatusImpl : public grpc_core::CQCallbackInterface {
     GPR_ASSERT(parent_->ops()->FinalizeResult(&ignored, &ok));
     GPR_ASSERT(ignored == parent_->ops());
 
-    func_(status_);
-    func_ = nullptr;  // release the function
+    // Last use of func_ or status_, so ok to move them out
+    CatchingCallback(std::move(func_), std::move(status_));
+
+    func_ = nullptr;  // reset to clear this out for sure
     grpc_call_unref(call_);
   }
   Status* status_ptr() { return &status_; }

test/cpp/end2end/client_callback_end2end_test.cc

@@ -64,7 +64,7 @@ class ClientCallbackEnd2endTest : public ::testing::Test {
     }
   }
 
-  void SendRpcs(int num_rpcs) {
+  void SendRpcs(int num_rpcs, bool maybe_except) {
     const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo");
     grpc::string test_string("");
     for (int i = 0; i < num_rpcs; i++) {
@@ -82,7 +82,7 @@ class ClientCallbackEnd2endTest : public ::testing::Test {
       bool done = false;
       stub_->experimental().UnaryCall(
           &cli_ctx, kMethodName, send_buf.get(), &recv_buf,
-          [&request, &recv_buf, &done, &mu, &cv](Status s) {
+          [&request, &recv_buf, &done, &mu, &cv, maybe_except](Status s) {
             GPR_ASSERT(s.ok());
 
             EchoResponse response;
@@ -91,6 +91,11 @@ class ClientCallbackEnd2endTest : public ::testing::Test {
             std::lock_guard<std::mutex> l(mu);
             done = true;
             cv.notify_one();
+#if GRPC_ALLOW_EXCEPTIONS
+            if (maybe_except) {
+              throw - 1;
+            }
+#endif
           });
       std::unique_lock<std::mutex> l(mu);
       while (!done) {
@@ -107,13 +112,20 @@ class ClientCallbackEnd2endTest : public ::testing::Test {
 
 TEST_F(ClientCallbackEnd2endTest, SimpleRpc) {
   ResetStub();
-  SendRpcs(1);
+  SendRpcs(1, false);
 }
 
 TEST_F(ClientCallbackEnd2endTest, SequentialRpcs) {
   ResetStub();
-  SendRpcs(10);
+  SendRpcs(10, false);
+}
+
+#if GRPC_ALLOW_EXCEPTIONS
+TEST_F(ClientCallbackEnd2endTest, ExceptingRpc) {
+  ResetStub();
+  SendRpcs(10, true);
 }
+#endif
 
 }  // namespace
 }  // namespace testing