Fixes

3 files changed, 22 insertions, 21 deletions

diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 84357b8075..855e490bc5 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -2034,6 +2034,8 @@ static void incoming_byte_stream_publish_error(
   grpc_closure_sched(exec_ctx, bs->on_next, GRPC_ERROR_REF(error));
   bs->on_next = NULL;
   GRPC_ERROR_UNREF(bs->error);
+  grpc_chttp2_cancel_stream(exec_ctx, bs->transport, bs->stream,
+                            GRPC_ERROR_REF(error));
   bs->error = error;
 }
 
diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h
index 23e85f20b3..ffacdac393 100644
--- a/src/core/lib/iomgr/error.h
+++ b/src/core/lib/iomgr/error.h
@@ -165,7 +165,7 @@ grpc_error *grpc_error_create(const char *file, int line, const char *desc,
 #define GRPC_ERROR_CREATE_REFERENCING(desc, errs, count) \
   grpc_error_create(__FILE__, __LINE__, desc, errs, count)
 
-#define GRPC_ERROR_REFCOUNT_DEBUG
+//#define GRPC_ERROR_REFCOUNT_DEBUG
 #ifdef GRPC_ERROR_REFCOUNT_DEBUG
 grpc_error *grpc_error_ref(grpc_error *err, const char *file, int line,
                            const char *func);
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index fde3430388..009e36fd87 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -1053,6 +1053,17 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
   }
 }
 
+static void finish_batch_step(grpc_exec_ctx *exec_ctx, batch_control *bctl,
+                              const char *but_why) {
+  gpr_log(GPR_DEBUG,
+          "finish_batch_step: call=%p bctl=%p why='%s' still_needed=%" PRIdPTR,
+          bctl->call, bctl, but_why,
+          gpr_atm_no_barrier_load(&bctl->steps_to_complete.count));
+  if (gpr_unref(&bctl->steps_to_complete)) {
+    post_batch_completion(exec_ctx, bctl);
+  }
+}
+
 static void continue_receiving_slices(grpc_exec_ctx *exec_ctx,
                                       batch_control *bctl) {
   grpc_call *call = bctl->call;
@@ -1063,9 +1074,7 @@ static void continue_receiving_slices(grpc_exec_ctx *exec_ctx,
       call->receiving_message = 0;
       grpc_byte_stream_destroy(exec_ctx, call->receiving_stream);
       call->receiving_stream = NULL;
-      if (gpr_unref(&bctl->steps_to_complete)) {
-        post_batch_completion(exec_ctx, bctl);
-      }
+      finish_batch_step(exec_ctx, bctl, "continue_receiving_slices");
       return;
     }
     if (grpc_byte_stream_next(exec_ctx, call->receiving_stream,
@@ -1096,9 +1105,7 @@ static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
     call->receiving_stream = NULL;
     grpc_byte_buffer_destroy(*call->receiving_buffer);
     *call->receiving_buffer = NULL;
-    if (gpr_unref(&bctl->steps_to_complete)) {
-      post_batch_completion(exec_ctx, bctl);
-    }
+    finish_batch_step(exec_ctx, bctl, "receiving_slice_ready with error");
   }
 }
 
@@ -1108,9 +1115,7 @@ static void process_data_after_md(grpc_exec_ctx *exec_ctx,
   if (call->receiving_stream == NULL) {
     *call->receiving_buffer = NULL;
     call->receiving_message = 0;
-    if (gpr_unref(&bctl->steps_to_complete)) {
-      post_batch_completion(exec_ctx, bctl);
-    }
+    finish_batch_step(exec_ctx, bctl, "no message");
   } else {
     call->test_only_last_message_flags = call->receiving_stream->flags;
     if ((call->receiving_stream->flags & GRPC_WRITE_INTERNAL_COMPRESS) &&
@@ -1130,18 +1135,17 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
                                    grpc_error *error) {
   batch_control *bctl = bctlp;
   grpc_call *call = bctl->call;
+  gpr_mu_lock(&bctl->call->mu);
   if (error != GRPC_ERROR_NONE) {
     cancel_with_error(exec_ctx, call, GRPC_ERROR_REF(error));
   }
-  gpr_mu_lock(&bctl->call->mu);
-  if (bctl->call->has_initial_md_been_received || error != GRPC_ERROR_NONE ||
+  if (call->has_initial_md_been_received || error != GRPC_ERROR_NONE ||
       call->receiving_stream == NULL) {
-    gpr_mu_unlock(&bctl->call->mu);
     process_data_after_md(exec_ctx, bctlp);
   } else {
     call->saved_receiving_stream_ready_bctlp = bctlp;
-    gpr_mu_unlock(&bctl->call->mu);
   }
+  gpr_mu_unlock(&bctl->call->mu);
 }
 
 static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx,
@@ -1239,9 +1243,7 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
 
   gpr_mu_unlock(&call->mu);
 
-  if (gpr_unref(&bctl->steps_to_complete)) {
-    post_batch_completion(exec_ctx, bctl);
-  }
+  finish_batch_step(exec_ctx, bctl, "receiving_initial_metadata_ready");
 }
 
 static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp,
@@ -1249,10 +1251,7 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp,
   batch_control *bctl = bctlp;
 
   add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error));
-
-  if (gpr_unref(&bctl->steps_to_complete)) {
-    post_batch_completion(exec_ctx, bctl);
-  }
+  finish_batch_step(exec_ctx, bctl, "finish_batch");
 }
 
 static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,