Merge branch 'master' into callback_codegen_client_unary

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/core/lib/surface/call.cc

@@ -207,7 +207,7 @@ struct grpc_call {
       grpc_server* server;
     } server;
   } final_op;
-  grpc_error* status_error;
+  gpr_atm status_error;
 
   /* recv_state can contain one of the following values:
      RECV_NONE :                 :  no initial metadata and messages received
@@ -519,10 +519,12 @@ static void destroy_call(void* call, grpc_error* error) {
     GRPC_CQ_INTERNAL_UNREF(c->cq, "bind");
   }
 
-  grpc_error_get_status(c->status_error, c->send_deadline,
+  grpc_error* status_error =
+      reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&c->status_error));
+  grpc_error_get_status(status_error, c->send_deadline,
                         &c->final_info.final_status, nullptr, nullptr,
                         &(c->final_info.error_string));
-  GRPC_ERROR_UNREF(c->status_error);
+  GRPC_ERROR_UNREF(status_error);
   c->final_info.stats.latency =
       gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), c->start_time);
 
@@ -705,7 +707,7 @@ static void set_final_status(grpc_call* call, grpc_error* error) {
                           call->final_op.client.error_string);
     // explicitly take a ref
     grpc_slice_ref_internal(*call->final_op.client.status_details);
-    call->status_error = error;
+    gpr_atm_rel_store(&call->status_error, reinterpret_cast<gpr_atm>(error));
     grpc_core::channelz::ChannelNode* channelz_channel =
         grpc_channel_get_channelz_node(call->channel);
     if (channelz_channel != nullptr) {
@@ -717,7 +719,9 @@ static void set_final_status(grpc_call* call, grpc_error* error) {
     }
   } else {
     *call->final_op.server.cancelled =
-        error != GRPC_ERROR_NONE || call->status_error != GRPC_ERROR_NONE;
+        error != GRPC_ERROR_NONE ||
+        reinterpret_cast<grpc_error*>(gpr_atm_acq_load(&call->status_error)) !=
+            GRPC_ERROR_NONE;
     grpc_core::channelz::ServerNode* channelz_server =
         grpc_server_get_channelz_node(call->final_op.server.server);
     if (channelz_server != nullptr) {
@@ -1686,7 +1690,8 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
           }
         }
 
-        call->status_error = status_error;
+        gpr_atm_rel_store(&call->status_error,
+                          reinterpret_cast<gpr_atm>(status_error));
         if (!prepare_application_metadata(
                 call,
                 static_cast<int>(

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_; }

src/cpp/ext/filters/census/server_filter.cc

@@ -93,7 +93,7 @@ void CensusServerCallData::OnDoneRecvInitialMetadataCb(void* user_data,
     FilterInitialMetadata(initial_metadata, &sml);
     calld->path_ = grpc_slice_ref_internal(sml.path);
     calld->method_ = GetMethod(&calld->path_);
-    calld->qualified_method_ = StrCat("Recv.", calld->method_);
+    calld->qualified_method_ = absl::StrCat("Recv.", calld->method_);
     const char* tracing_str =
         GRPC_SLICE_IS_EMPTY(sml.tracing_slice)
             ? ""

src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi

@@ -144,8 +144,14 @@ cdef class SSLChannelCredentials(ChannelCredentials):
       return grpc_ssl_credentials_create(
           c_pem_root_certificates, NULL, NULL, NULL)
     else:
-      c_pem_key_certificate_pair.private_key = self._private_key
-      c_pem_key_certificate_pair.certificate_chain = self._certificate_chain
+      if self._private_key:
+        c_pem_key_certificate_pair.private_key = self._private_key
+      else:
+        c_pem_key_certificate_pair.private_key = NULL
+      if self._certificate_chain:
+        c_pem_key_certificate_pair.certificate_chain = self._certificate_chain
+      else:
+        c_pem_key_certificate_pair.certificate_chain = NULL
       return grpc_ssl_credentials_create(
           c_pem_root_certificates, &c_pem_key_certificate_pair, NULL, NULL)
 

src/ruby/lib/grpc/generic/rpc_server.rb

@@ -136,7 +136,7 @@ module GRPC
         begin
           blk, args = worker_queue.pop
           blk.call(*args)
-        rescue StandardError => e
+        rescue StandardError, GRPC::Core::CallError => e
           GRPC.logger.warn('Error in worker thread')
           GRPC.logger.warn(e)
         end

test/cpp/end2end/client_callback_end2end_test.cc

@@ -65,6 +65,7 @@ class ClientCallbackEnd2endTest : public ::testing::Test {
     }
   }
 
+<<<<<<< HEAD
   void SendRpcs(int num_rpcs) {
     grpc::string test_string("");
     for (int i = 0; i < num_rpcs; i++) {
@@ -96,6 +97,9 @@ class ClientCallbackEnd2endTest : public ::testing::Test {
   }
 
   void SendRpcsGeneric(int num_rpcs) {
+=======
+  void SendRpcs(int num_rpcs, bool maybe_except) {
+>>>>>>> master
     const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo");
     grpc::string test_string("");
     for (int i = 0; i < num_rpcs; i++) {
@@ -113,7 +117,7 @@ class ClientCallbackEnd2endTest : public ::testing::Test {
       bool done = false;
       generic_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;
@@ -122,6 +126,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) {
@@ -140,12 +149,12 @@ 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);
 }
 
 TEST_F(ClientCallbackEnd2endTest, SequentialGenericRpcs) {
@@ -153,6 +162,13 @@ TEST_F(ClientCallbackEnd2endTest, SequentialGenericRpcs) {
   SendRpcsGeneric(10);
 }
 
+#if GRPC_ALLOW_EXCEPTIONS
+TEST_F(ClientCallbackEnd2endTest, ExceptingRpc) {
+  ResetStub();
+  SendRpcs(10, true);
+}
+#endif
+
 }  // namespace
 }  // namespace testing
 }  // namespace grpc

test/cpp/microbenchmarks/helpers.cc

@@ -38,6 +38,7 @@ void TrackCounters::AddLabel(const grpc::string& label) {
 }
 
 void TrackCounters::AddToLabel(std::ostream& out, benchmark::State& state) {
+#ifdef GRPC_COLLECT_STATS
   grpc_stats_data stats_end;
   grpc_stats_collect(&stats_end);
   grpc_stats_data stats;
@@ -53,6 +54,7 @@ void TrackCounters::AddToLabel(std::ostream& out, benchmark::State& state) {
         << " " << grpc_stats_histogram_name[i] << "-99p:"
         << grpc_stats_histo_percentile(&stats, (grpc_stats_histograms)i, 99.0);
   }
+#endif
 #ifdef GPR_LOW_LEVEL_COUNTERS
   grpc_memory_counters counters_at_end = grpc_memory_counters_snapshot();
   out << " locks/iter:"

tools/interop_matrix/client_matrix.py

@@ -99,6 +99,9 @@ LANG_RELEASE_MATRIX = {
         {
             'v1.14.1': None
         },
+        {
+            'v1.15.0': None
+        },
     ],
     'go': [
         {
@@ -231,6 +234,9 @@ LANG_RELEASE_MATRIX = {
         {
             'v1.14.1': None
         },
+        {
+            'v1.15.0': None
+        },
     ],
     'node': [
         {
@@ -319,6 +325,9 @@ LANG_RELEASE_MATRIX = {
         {
             'v1.14.1': None
         },
+        {
+            'v1.15.0': None
+        },
     ],
     'php': [
         {
@@ -363,6 +372,9 @@ LANG_RELEASE_MATRIX = {
         {
             'v1.14.1': None
         },
+        {
+            'v1.15.0': None
+        },
     ],
     'csharp': [
         {
@@ -412,6 +424,9 @@ LANG_RELEASE_MATRIX = {
         {
             'v1.14.1': None
         },
+        {
+            'v1.15.0': None
+        },
     ],
 }
 

tools/run_tests/run_tests_matrix.py

@@ -53,12 +53,7 @@ def _safe_report_name(name):
 
 
 def _report_filename(name):
-    """Generates report file name"""
-    return 'report_%s_%s' % (_safe_report_name(name), _REPORT_SUFFIX)
-
-
-def _report_filename_internal_ci(name):
-    """Generates report file name that leads to better presentation by internal CI"""
+    """Generates report file name with directory structure that leads to better presentation by internal CI"""
     return '%s/%s' % (_safe_report_name(name), _REPORT_SUFFIX)
 
 
@@ -507,8 +502,9 @@ if __name__ == "__main__":
         default=False,
         action='store_const',
         const=True,
-        help='Put reports into subdirectories to improve presentation of '
-        'results by Internal CI.')
+        help=
+        '(Deprecated, has no effect) Put reports into subdirectories to improve presentation of '
+        'results by Kokoro.')
     argp.add_argument(
         '--bq_result_table',
         default='',
@@ -517,9 +513,6 @@ if __name__ == "__main__":
         help='Upload test results to a specified BQ table.')
     args = argp.parse_args()
 
-    if args.internal_ci:
-        _report_filename = _report_filename_internal_ci  # override the function
-
     extra_args = []
     if args.build_only:
         extra_args.append('--build_only')