Merge pull request #6741 from yang-g/its_ok_to_have_no_read_on_cancelled_call

Let Next set ok=true when receiving error status

include/grpc++/impl/codegen/async_stream.h

@@ -172,6 +172,7 @@ class ClientAsyncWriter GRPC_FINAL : public ClientAsyncWriterInterface<W> {
                     R* response, void* tag)
       : context_(context), call_(channel->CreateCall(method, context, cq)) {
     finish_ops_.RecvMessage(response);
+    finish_ops_.AllowNoMessage();
 
     init_ops_.set_output_tag(tag);
     init_ops_.SendInitialMetadata(context->send_initial_metadata_,

include/grpc++/impl/codegen/async_unary_call.h

@@ -91,6 +91,7 @@ class ClientAsyncResponseReader GRPC_FINAL
       collection_->finish_buf_.RecvInitialMetadata(context_);
     }
     collection_->finish_buf_.RecvMessage(msg);
+    collection_->finish_buf_.AllowNoMessage();
     collection_->finish_buf_.ClientRecvStatus(context_, status);
     call_.PerformOps(&collection_->finish_buf_);
   }

include/grpc++/impl/codegen/call.h

@@ -261,10 +261,16 @@ Status CallOpSendMessage::SendMessage(const M& message) {
 template <class R>
 class CallOpRecvMessage {
  public:
-  CallOpRecvMessage() : got_message(false), message_(nullptr) {}
+  CallOpRecvMessage()
+      : got_message(false),
+        message_(nullptr),
+        allow_not_getting_message_(false) {}
 
   void RecvMessage(R* message) { message_ = message; }
 
+  // Do not change status if no message is received.
+  void AllowNoMessage() { allow_not_getting_message_ = true; }
+
   bool got_message;
 
  protected:
@@ -290,7 +296,9 @@ class CallOpRecvMessage {
       }
     } else {
       got_message = false;
-      *status = false;
+      if (!allow_not_getting_message_) {
+        *status = false;
+      }
     }
     message_ = nullptr;
   }
@@ -298,6 +306,7 @@ class CallOpRecvMessage {
  private:
   R* message_;
   grpc_byte_buffer* recv_buf_;
+  bool allow_not_getting_message_;
 };
 
 namespace CallOpGenericRecvMessageHelper {
@@ -325,7 +334,8 @@ class DeserializeFuncType GRPC_FINAL : public DeserializeFunc {
 
 class CallOpGenericRecvMessage {
  public:
-  CallOpGenericRecvMessage() : got_message(false) {}
+  CallOpGenericRecvMessage()
+      : got_message(false), allow_not_getting_message_(false) {}
 
   template <class R>
   void RecvMessage(R* message) {
@@ -336,6 +346,9 @@ class CallOpGenericRecvMessage {
     deserialize_.reset(func);
   }
 
+  // Do not change status if no message is received.
+  void AllowNoMessage() { allow_not_getting_message_ = true; }
+
   bool got_message;
 
  protected:
@@ -360,7 +373,9 @@ class CallOpGenericRecvMessage {
       }
     } else {
       got_message = false;
-      *status = false;
+      if (!allow_not_getting_message_) {
+        *status = false;
+      }
     }
     deserialize_.reset();
   }
@@ -368,6 +383,7 @@ class CallOpGenericRecvMessage {
  private:
   std::unique_ptr<CallOpGenericRecvMessageHelper::DeserializeFunc> deserialize_;
   grpc_byte_buffer* recv_buf_;
+  bool allow_not_getting_message_;
 };
 
 class CallOpClientSendClose {

include/grpc++/impl/codegen/sync_stream.h

@@ -189,6 +189,7 @@ class ClientWriter : public ClientWriterInterface<W> {
                ClientContext* context, R* response)
       : context_(context), call_(channel->CreateCall(method, context, &cq_)) {
     finish_ops_.RecvMessage(response);
+    finish_ops_.AllowNoMessage();
 
     CallOpSet<CallOpSendInitialMetadata> ops;
     ops.SendInitialMetadata(context->send_initial_metadata_,

test/cpp/end2end/async_end2end_test.cc

@@ -819,7 +819,7 @@ TEST_P(AsyncEnd2endTest, ServerCheckCancellation) {
   EXPECT_TRUE(srv_ctx.IsCancelled());
 
   response_reader->Finish(&recv_response, &recv_status, tag(4));
-  Verifier(GetParam().disable_blocking).Expect(4, false).Verify(cq_.get());
+  Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get());
 
   EXPECT_EQ(StatusCode::CANCELLED, recv_status.error_code());
 }
@@ -881,7 +881,7 @@ TEST_P(AsyncEnd2endTest, UnimplementedRpc) {
       stub->AsyncUnimplemented(&cli_ctx, send_request, cq_.get()));
 
   response_reader->Finish(&recv_response, &recv_status, tag(4));
-  Verifier(GetParam().disable_blocking).Expect(4, false).Verify(cq_.get());
+  Verifier(GetParam().disable_blocking).Expect(4, true).Verify(cq_.get());
 
   EXPECT_EQ(StatusCode::UNIMPLEMENTED, recv_status.error_code());
   EXPECT_EQ("", recv_status.error_message());
@@ -1026,9 +1026,7 @@ class AsyncEnd2endServerTryCancelTest : public AsyncEnd2endTest {
 
     // Client will see the cancellation
     cli_stream->Finish(&recv_status, tag(10));
-    // TODO(sreek): The expectation here should be true. This is a bug (github
-    // issue #4972)
-    Verifier(GetParam().disable_blocking).Expect(10, false).Verify(cq_.get());
+    Verifier(GetParam().disable_blocking).Expect(10, true).Verify(cq_.get());
     EXPECT_FALSE(recv_status.ok());
     EXPECT_EQ(::grpc::StatusCode::CANCELLED, recv_status.error_code());
   }