Merge pull request #4142 from yang-g/cancel

Handle Cancel before start.

include/grpc++/client_context.h

@@ -53,15 +53,16 @@
 #include <memory>
 #include <string>
 
-#include <grpc/compression.h>
-#include <grpc/grpc.h>
-#include <grpc/support/log.h>
-#include <grpc/support/time.h>
+#include <grpc++/impl/sync.h>
 #include <grpc++/security/auth_context.h>
 #include <grpc++/support/config.h>
 #include <grpc++/support/status.h>
 #include <grpc++/support/string_ref.h>
 #include <grpc++/support/time.h>
+#include <grpc/compression.h>
+#include <grpc/grpc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/time.h>
 
 struct census_context;
 
@@ -315,7 +316,9 @@ class ClientContext {
 
   bool initial_metadata_received_;
   std::shared_ptr<Channel> channel_;
+  grpc::mutex mu_;
   grpc_call* call_;
+  bool call_canceled_;
   gpr_timespec deadline_;
   grpc::string authority_;
   std::shared_ptr<Credentials> creds_;

include/grpc++/support/time.h

@@ -35,6 +35,7 @@
 #define GRPCXX_SUPPORT_TIME_H
 
 #include <grpc++/support/config.h>
+#include <grpc/support/time.h>
 
 namespace grpc {
 

src/cpp/client/client_context.cc

@@ -48,6 +48,7 @@ namespace grpc {
 ClientContext::ClientContext()
     : initial_metadata_received_(false),
       call_(nullptr),
+      call_canceled_(false),
       deadline_(gpr_inf_future(GPR_CLOCK_REALTIME)),
       propagate_from_call_(nullptr) {}
 
@@ -72,6 +73,7 @@ void ClientContext::AddMetadata(const grpc::string& meta_key,
 
 void ClientContext::set_call(grpc_call* call,
                              const std::shared_ptr<Channel>& channel) {
+  grpc::unique_lock<grpc::mutex> lock(mu_);
   GPR_ASSERT(call_ == nullptr);
   call_ = call;
   channel_ = channel;
@@ -79,6 +81,9 @@ void ClientContext::set_call(grpc_call* call,
     grpc_call_cancel_with_status(call, GRPC_STATUS_CANCELLED,
                                  "Failed to set credentials to rpc.", nullptr);
   }
+  if (call_canceled_) {
+    grpc_call_cancel(call_, nullptr);
+  }
 }
 
 void ClientContext::set_compression_algorithm(
@@ -101,8 +106,11 @@ std::shared_ptr<const AuthContext> ClientContext::auth_context() const {
 }
 
 void ClientContext::TryCancel() {
+  grpc::unique_lock<grpc::mutex> lock(mu_);
   if (call_) {
     grpc_call_cancel(call_, nullptr);
+  } else {
+    call_canceled_ = true;
   }
 }
 

test/cpp/end2end/end2end_test.cc

@@ -575,6 +575,18 @@ void CancelRpc(ClientContext* context, int delay_us, TestServiceImpl* service) {
   context->TryCancel();
 }
 
+TEST_P(End2endTest, CancelRpcBeforeStart) {
+  ResetStub();
+  EchoRequest request;
+  EchoResponse response;
+  ClientContext context;
+  request.set_message("hello");
+  context.TryCancel();
+  Status s = stub_->Echo(&context, request, &response);
+  EXPECT_EQ("", response.message());
+  EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
+}
+
 // Client cancels request stream after sending two messages
 TEST_P(End2endTest, ClientCancelsRequestStream) {
   ResetStub();