Porting Aaron Jacob's unit test that detecting the race

test/cpp/end2end/generic_end2end_test.cc

@@ -92,6 +92,7 @@ class GenericEnd2endTest : public ::testing::Test {
   void ResetStub() {
     std::shared_ptr<Channel> channel = grpc::CreateChannel(
         server_address_.str(), InsecureChannelCredentials());
+    stub_ = grpc::testing::EchoTestService::NewStub(channel);
     generic_stub_.reset(new GenericStub(channel));
   }
 
@@ -177,6 +178,54 @@ class GenericEnd2endTest : public ::testing::Test {
     }
   }
 
+  // Return errors to up to one call that comes in on the supplied completion
+  // queue, until the CQ is being shut down (and therefore we can no longer
+  // enqueue further events).
+  void DriveCompletionQueue() {
+    enum class Event : uintptr_t {
+      kCallReceived,
+      kResponseSent,
+    };
+    // Request the call, but only if the main thread hasn't beaten us to
+    // shutting down the CQ.
+    grpc::GenericServerContext server_context;
+    grpc::GenericServerAsyncReaderWriter reader_writer(&server_context);
+
+    {
+      std::lock_guard<std::mutex> lock(shutting_down_mu_);
+      if (!shutting_down_) {
+        generic_service_.RequestCall(
+            &server_context, &reader_writer, srv_cq_.get(), srv_cq_.get(),
+            reinterpret_cast<void*>(Event::kCallReceived));
+      }
+    }
+    // Process events.
+    {
+      Event event;
+      bool ok;
+      while (srv_cq_->Next(reinterpret_cast<void**>(&event), &ok)) {
+        std::lock_guard<std::mutex> lock(shutting_down_mu_);
+        if (shutting_down_) {
+          // The main thread has started shutting down. Simply continue to drain
+          // events.
+          continue;
+        }
+
+        switch (event) {
+          case Event::kCallReceived:
+            reader_writer.Finish(
+                ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "go away"),
+                reinterpret_cast<void*>(Event::kResponseSent));
+            break;
+
+          case Event::kResponseSent:
+            // We are done.
+            break;
+        }
+      }
+    }
+  }
+
   CompletionQueue cli_cq_;
   std::unique_ptr<ServerCompletionQueue> srv_cq_;
   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
@@ -185,6 +234,8 @@ class GenericEnd2endTest : public ::testing::Test {
   AsyncGenericService generic_service_;
   const grpc::string server_host_;
   std::ostringstream server_address_;
+  bool shutting_down_;
+  std::mutex shutting_down_mu_;
 };
 
 TEST_F(GenericEnd2endTest, SimpleRpc) {
@@ -330,6 +381,32 @@ TEST_F(GenericEnd2endTest, Deadline) {
                        gpr_time_from_seconds(10, GPR_TIMESPAN)));
 }
 
+TEST_F(GenericEnd2endTest, ShortDeadline) {
+  ResetStub();
+
+  ClientContext cli_ctx;
+  EchoRequest request;
+  EchoResponse response;
+
+  {
+    std::lock_guard<std::mutex> lock(shutting_down_mu_);
+    shutting_down_ = false;
+  }
+  std::thread driver([=] { DriveCompletionQueue(); });
+
+  request.set_message("");
+  cli_ctx.set_deadline(gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
+                                    gpr_time_from_micros(500, GPR_TIMESPAN)));
+  Status s = stub_->Echo(&cli_ctx, request, &response);
+  EXPECT_FALSE(s.ok());
+  {
+    std::lock_guard<std::mutex> lock(shutting_down_mu_);
+    shutting_down_ = true;
+  }
+  TearDown();
+  driver.join();
+}
+
 }  // namespace
 }  // namespace testing
 }  // namespace grpc