3 files changed, 22 insertions, 9 deletions

diff --git a/src/core/support/subprocess_posix.c b/src/core/support/subprocess_posix.c
index 642520bb47..b4631fa0ed 100644
--- a/src/core/support/subprocess_posix.c
+++ b/src/core/support/subprocess_posix.c
@@ -57,7 +57,7 @@ struct gpr_subprocess {
 
 const char *gpr_subprocess_binary_extension() { return ""; }
 
-gpr_subprocess *gpr_subprocess_create(int argc, char **argv) {
+gpr_subprocess *gpr_subprocess_create(int argc, const char **argv) {
   gpr_subprocess *r;
   int pid;
   char **exec_args;
@@ -92,7 +92,11 @@ void gpr_subprocess_destroy(gpr_subprocess *p) {
 
 int gpr_subprocess_join(gpr_subprocess *p) {
   int status;
+retry:
   if (waitpid(p->pid, &status, 0) == -1) {
+    if (errno == EINTR) {
+      goto retry;
+    }
     gpr_log(GPR_ERROR, "waitpid failed: %s", strerror(errno));
     return -1;
   }
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 50df36cae9..8b1ef0f2be 100644
--- a/src/core/surface/call.c
+++ b/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;
-          } else {
+          call->request_set[i] = REQSET_EMPTY;
+          if (!master->success) {
             call->write_state = WRITE_STATE_WRITE_CLOSED;
           }
           break;
@@ -588,6 +587,7 @@ static void call_on_done_send(void *pc, int success) {
   lock(call);
   if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_INITIAL_METADATA)) {
     finish_ioreq_op(call, GRPC_IOREQ_SEND_INITIAL_METADATA, success);
+    call->write_state = WRITE_STATE_STARTED;
   }
   if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_MESSAGE)) {
     finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, success);
@@ -596,6 +596,10 @@ static void call_on_done_send(void *pc, int success) {
     finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, success);
     finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, success);
     finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, 1);
+    call->write_state = WRITE_STATE_WRITE_CLOSED;
+  }
+  if (!success) {
+    call->write_state = WRITE_STATE_WRITE_CLOSED;
   }
   call->last_send_contains = 0;
   call->sending = 0;
@@ -810,7 +814,6 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op) {
       op->send_ops = &call->send_ops;
       op->bind_pollset = grpc_cq_pollset(call->cq);
       call->last_send_contains |= 1 << GRPC_IOREQ_SEND_INITIAL_METADATA;
-      call->write_state = WRITE_STATE_STARTED;
       call->send_initial_metadata_count = 0;
     /* fall through intended */
     case WRITE_STATE_STARTED:
@@ -826,7 +829,6 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op) {
         op->is_last_send = 1;
         op->send_ops = &call->send_ops;
         call->last_send_contains |= 1 << GRPC_IOREQ_SEND_CLOSE;
-        call->write_state = WRITE_STATE_WRITE_CLOSED;
         if (!call->is_client) {
           /* send trailing metadata */
           data = call->request_data[GRPC_IOREQ_SEND_TRAILING_METADATA];
@@ -1175,6 +1177,10 @@ static void set_cancelled_value(grpc_status_code status, void *dest) {
 }
 
 static void finish_batch(grpc_call *call, int success, void *tag) {
+  grpc_cq_end_op(call->cq, tag, call, success);
+}
+
+static void finish_batch_with_close(grpc_call *call, int success, void *tag) {
   grpc_cq_end_op(call->cq, tag, call, 1);
 }
 
@@ -1185,6 +1191,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
   size_t out;
   const grpc_op *op;
   grpc_ioreq *req;
+  void (*finish_func)(grpc_call *, int, void *) = finish_batch;
 
   GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, tag);
 
@@ -1268,6 +1275,7 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
             op->data.recv_status_on_client.trailing_metadata;
         req = &reqs[out++];
         req->op = GRPC_IOREQ_RECV_CLOSE;
+        finish_func = finish_batch_with_close;
         break;
       case GRPC_OP_RECV_CLOSE_ON_SERVER:
         req = &reqs[out++];
@@ -1277,13 +1285,14 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
             op->data.recv_close_on_server.cancelled;
         req = &reqs[out++];
         req->op = GRPC_IOREQ_RECV_CLOSE;
+        finish_func = finish_batch_with_close;
         break;
     }
   }
 
   grpc_cq_begin_op(call->cq, call);
 
-  return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_batch,
+  return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_func,
                                              tag);
 }
 
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 48910afda3..8c9ca48a05 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -275,14 +275,14 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
       gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
       memset(&ret, 0, sizeof(ret));
       ret.type = GRPC_QUEUE_TIMEOUT;
-      GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ev->base);
+      GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
       return ret;
     }
   }
   gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
   ret = ev->base;
   gpr_free(ev);
-  GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ev->base);
+  GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
   return ret;
 }