Merge pull request #1682 from ctiller/my-asan-is-better-than-yours

ASAN fixes for cancel_after_invoke tests

src/core/surface/call.c

@@ -401,6 +401,7 @@ static int is_op_live(grpc_call *call, grpc_ioreq_op op) {
 static void lock(grpc_call *call) { gpr_mu_lock(&call->mu); }
 
 static int need_more_data(grpc_call *call) {
+  if (call->read_state == READ_STATE_STREAM_CLOSED) return 0;
   return is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA) ||
          (is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) && grpc_bbq_empty(&call->incoming_queue)) ||
          is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA) ||
@@ -408,8 +409,7 @@ static int need_more_data(grpc_call *call) {
          is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS) ||
          (is_op_live(call, GRPC_IOREQ_RECV_CLOSE) &&
           grpc_bbq_empty(&call->incoming_queue)) ||
-         (call->write_state == WRITE_STATE_INITIAL && !call->is_client &&
+         (call->write_state == WRITE_STATE_INITIAL && !call->is_client);
-          call->read_state < READ_STATE_GOT_INITIAL_METADATA);
 }
 
 static void unlock(grpc_call *call) {

src/core/surface/server.c

@@ -427,6 +427,8 @@ static void server_on_recv(void *ptr, int success) {
         grpc_iomgr_add_callback(kill_zombie, elem);
       } else if (calld->state == PENDING) {
         call_list_remove(calld, PENDING_START);
+        calld->state = ZOMBIED;
+        grpc_iomgr_add_callback(kill_zombie, elem);
       }
       gpr_mu_unlock(&chand->server->mu);
       break;

src/core/transport/chttp2_transport.c

@@ -1142,6 +1142,7 @@ static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) {
 
   if (op->recv_ops) {
     GPR_ASSERT(s->incoming_sopb == NULL);
+    GPR_ASSERT(s->published_state != GRPC_STREAM_CLOSED);
     s->recv_done_closure.cb = op->on_done_recv;
     s->recv_done_closure.user_data = op->recv_user_data;
     s->incoming_sopb = op->recv_ops;