Merge pull request #3627 from yang-g/flake2

Deflake end2end_test

src/core/iomgr/pollset_posix.c

@@ -352,6 +352,7 @@ static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args, int success) {
   if (pollset->shutting_down) {
     /* We don't care about this pollset anymore. */
     if (pollset->in_flight_cbs == 0 && !pollset->called_shutdown) {
+      pollset->called_shutdown = 1;
       finish_shutdown(exec_ctx, pollset);
     }
   } else if (grpc_fd_is_orphaned(fd)) {

src/core/transport/chttp2_transport.c

@@ -1136,7 +1136,7 @@ static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success) {
     grpc_chttp2_publish_reads(exec_ctx, &t->global, &t->parsing);
     t->parsing_active = 0;
   }
-  if (!success || i != t->read_buffer.count) {
+  if (!success || i != t->read_buffer.count || t->closed) {
     drop_connection(exec_ctx, t);
     read_error_locked(exec_ctx, t);
   } else if (!t->closed) {

test/cpp/end2end/end2end_test.cc

@@ -683,10 +683,14 @@ TEST_P(End2endTest, RequestStreamServerEarlyCancelTest) {
   auto stream = stub_->RequestStream(&context, &response);
   request.set_message("hello");
   int send_messages = 20;
-  while (send_messages > 0) {
+  while (send_messages > 10) {
     EXPECT_TRUE(stream->Write(request));
     send_messages--;
   }
+  while (send_messages > 0) {
+    stream->Write(request);
+    send_messages--;
+  }
   stream->WritesDone();
   Status s = stream->Finish();
   EXPECT_EQ(s.error_code(), StatusCode::CANCELLED);