Add response streams to server crash test, fix them

src/core/surface/call.c

@@ -536,9 +536,8 @@ static void finish_live_ioreq_op(grpc_call *call, grpc_ioreq_op op,
       switch ((grpc_ioreq_op)i) {
         case GRPC_IOREQ_RECV_MESSAGE:
         case GRPC_IOREQ_SEND_MESSAGE:
-          if (master->success) {
+          call->request_set[i] = REQSET_EMPTY;
-            call->request_set[i] = REQSET_EMPTY;
+          if (!master->success) {
-          } else {
             call->write_state = WRITE_STATE_WRITE_CLOSED;
           }
           break;

test/cpp/end2end/server_crash_test.cc

@@ -79,6 +79,20 @@ class ServiceImpl GRPC_FINAL : public ::grpc::cpp::test::util::TestService::Serv
       gpr_log(GPR_INFO, "recv msg %s", request.message().c_str());
       response.set_message(request.message());
       stream->Write(response);
+      gpr_sleep_until(gpr_time_add(gpr_now(), gpr_time_from_seconds(1)));
+    }
+    return Status::OK;
+  }
+
+  Status ResponseStream(ServerContext* context, const EchoRequest* request,
+                        ServerWriter<EchoResponse>* writer) GRPC_OVERRIDE {
+    EchoResponse response;
+    for (int i = 0;; i++) {
+      std::ostringstream msg;
+      msg << "Hello " << i;
+      response.set_message(msg.str());
+      if (!writer->Write(response)) break;
+      gpr_sleep_until(gpr_time_add(gpr_now(), gpr_time_from_seconds(1)));
     }
     return Status::OK;
   }
@@ -89,7 +103,7 @@ class CrashTest : public ::testing::Test {
   CrashTest() {}
 
   std::unique_ptr<Server>
-  CreateServerAndClient() {
+  CreateServerAndClient(const std::string& mode) {
     auto port = grpc_pick_unused_port_or_die();
     std::ostringstream addr_stream;
     addr_stream << "localhost:" << port;
@@ -97,6 +111,7 @@ class CrashTest : public ::testing::Test {
     client_.reset(new SubProcess({
       g_root + "/server_crash_test_client",
       "--address=" + addr,
+      "--mode=" + mode
     }));
     GPR_ASSERT(client_);
 
@@ -115,8 +130,16 @@ class CrashTest : public ::testing::Test {
   ServiceImpl service_;
 };
 
-TEST_F(CrashTest, Kill) {
+TEST_F(CrashTest, ResponseStream) {
-  auto server = CreateServerAndClient();
+  auto server = CreateServerAndClient("response");
+
+  gpr_sleep_until(gpr_time_add(gpr_now(), gpr_time_from_seconds(5)));
+  KillClient();
+  server->Shutdown();
+}
+
+TEST_F(CrashTest, BidiStream) {
+  auto server = CreateServerAndClient("bidi");
 
   gpr_sleep_until(gpr_time_add(gpr_now(), gpr_time_from_seconds(5)));
   KillClient();

test/cpp/end2end/server_crash_test_client.cc

@@ -46,6 +46,7 @@
 #include "test/cpp/util/echo.grpc.pb.h"
 
 DEFINE_string(address, "", "Address to connect to");
+DEFINE_string(mode, "", "Test mode to use");
 
 using grpc::cpp::test::util::EchoRequest;
 using grpc::cpp::test::util::EchoResponse;
@@ -66,14 +67,26 @@ int main(int argc, char** argv) {
   EchoResponse response;
   grpc::ClientContext context;
 
-  auto stream = stub->BidiStream(&context);
+  if (FLAGS_mode == "bidi") {
-  for (int i = 0;; i++) {
+    auto stream = stub->BidiStream(&context);
-    std::ostringstream msg;
+    for (int i = 0;; i++) {
-    msg << "Hello " << i;
+      std::ostringstream msg;
-    request.set_message(msg.str());
+      msg << "Hello " << i;
-    GPR_ASSERT(stream->Write(request));
+      request.set_message(msg.str());
-    GPR_ASSERT(stream->Read(&response));
+      GPR_ASSERT(stream->Write(request));
-    GPR_ASSERT(response.message() == request.message());
+      GPR_ASSERT(stream->Read(&response));
+      GPR_ASSERT(response.message() == request.message());
+    }
+  } else if (FLAGS_mode == "response") {
+    EchoRequest request;
+    request.set_message("Hello");
+    auto stream = stub->ResponseStream(&context, request);
+    for (;;) {
+      GPR_ASSERT(stream->Read(&response));
+    }
+  } else {
+    gpr_log(GPR_ERROR, "invalid test mode '%s'", FLAGS_mode.c_str());
+    return 1;
   }
 
   return 0;