1 files changed, 19 insertions, 16 deletions

diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c
index a3184a1752..7f21176497 100644
--- a/src/core/ext/transport/cronet/transport/cronet_transport.c
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.c
@@ -372,13 +372,12 @@ static void remove_from_storage(struct stream_obj *s,
   This can get executed from the Cronet network thread via cronet callback
   or on the application supplied thread via the perform_stream_op function.
 */
-static void execute_from_storage(stream_obj *s) {
-  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+static void execute_from_storage(grpc_exec_ctx *exec_ctx, stream_obj *s) {
   gpr_mu_lock(&s->mu);
   for (struct op_and_state *curr = s->storage.head; curr != NULL;) {
     CRONET_LOG(GPR_DEBUG, "calling op at %p. done = %d", curr, curr->done);
     GPR_ASSERT(curr->done == 0);
-    enum e_op_result result = execute_stream_op(&exec_ctx, curr);
+    enum e_op_result result = execute_stream_op(exec_ctx, curr);
     CRONET_LOG(GPR_DEBUG, "execute_stream_op[%p] returns %s", curr,
                op_result_string(result));
     /* if this op is done, then remove it and free memory */
@@ -395,7 +394,6 @@ static void execute_from_storage(stream_obj *s) {
     }
   }
   gpr_mu_unlock(&s->mu);
-  grpc_exec_ctx_finish(&exec_ctx);
 }
 
 /*
@@ -420,7 +418,7 @@ static void on_failed(bidirectional_stream *stream, int net_error) {
   }
   null_and_maybe_free_read_buffer(s);
   gpr_mu_unlock(&s->mu);
-  execute_from_storage(s);
+  execute_from_storage(&exec_ctx, s);
   GRPC_CRONET_STREAM_UNREF(&exec_ctx, s, "cronet transport");
   grpc_exec_ctx_finish(&exec_ctx);
 }
@@ -447,7 +445,7 @@ static void on_canceled(bidirectional_stream *stream) {
   }
   null_and_maybe_free_read_buffer(s);
   gpr_mu_unlock(&s->mu);
-  execute_from_storage(s);
+  execute_from_storage(&exec_ctx, s);
   GRPC_CRONET_STREAM_UNREF(&exec_ctx, s, "cronet transport");
   grpc_exec_ctx_finish(&exec_ctx);
 }
@@ -466,7 +464,7 @@ static void on_succeeded(bidirectional_stream *stream) {
   s->cbs = NULL;
   null_and_maybe_free_read_buffer(s);
   gpr_mu_unlock(&s->mu);
-  execute_from_storage(s);
+  execute_from_storage(&exec_ctx, s);
   GRPC_CRONET_STREAM_UNREF(&exec_ctx, s, "cronet transport");
   grpc_exec_ctx_finish(&exec_ctx);
 }
@@ -476,6 +474,7 @@ static void on_succeeded(bidirectional_stream *stream) {
 */
 static void on_stream_ready(bidirectional_stream *stream) {
   CRONET_LOG(GPR_DEBUG, "W: on_stream_ready(%p)", stream);
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   stream_obj *s = (stream_obj *)stream->annotation;
   grpc_cronet_transport *t = (grpc_cronet_transport *)s->curr_ct;
   gpr_mu_lock(&s->mu);
@@ -495,7 +494,8 @@ static void on_stream_ready(bidirectional_stream *stream) {
     }
   }
   gpr_mu_unlock(&s->mu);
-  execute_from_storage(s);
+  execute_from_storage(&exec_ctx, s);
+  grpc_exec_ctx_finish(&exec_ctx);
 }
 
 /*
@@ -551,14 +551,15 @@ static void on_response_headers_received(
     s->state.pending_read_from_cronet = true;
   }
   gpr_mu_unlock(&s->mu);
+  execute_from_storage(&exec_ctx, s);
   grpc_exec_ctx_finish(&exec_ctx);
-  execute_from_storage(s);
 }
 
 /*
   Cronet callback
 */
 static void on_write_completed(bidirectional_stream *stream, const char *data) {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   stream_obj *s = (stream_obj *)stream->annotation;
   CRONET_LOG(GPR_DEBUG, "W: on_write_completed(%p, %s)", stream, data);
   gpr_mu_lock(&s->mu);
@@ -568,7 +569,8 @@ static void on_write_completed(bidirectional_stream *stream, const char *data) {
   }
   s->state.state_callback_received[OP_SEND_MESSAGE] = true;
   gpr_mu_unlock(&s->mu);
-  execute_from_storage(s);
+  execute_from_storage(&exec_ctx, s);
+  grpc_exec_ctx_finish(&exec_ctx);
 }
 
 /*
@@ -576,6 +578,7 @@ static void on_write_completed(bidirectional_stream *stream, const char *data) {
 */
 static void on_read_completed(bidirectional_stream *stream, char *data,
                               int count) {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   stream_obj *s = (stream_obj *)stream->annotation;
   CRONET_LOG(GPR_DEBUG, "R: on_read_completed(%p, %p, %d)", stream, data,
              count);
@@ -601,14 +604,15 @@ static void on_read_completed(bidirectional_stream *stream, char *data,
       gpr_mu_unlock(&s->mu);
     } else {
       gpr_mu_unlock(&s->mu);
-      execute_from_storage(s);
+      execute_from_storage(&exec_ctx, s);
     }
   } else {
     null_and_maybe_free_read_buffer(s);
     s->state.rs.read_stream_closed = true;
     gpr_mu_unlock(&s->mu);
-    execute_from_storage(s);
+    execute_from_storage(&exec_ctx, s);
   }
+  grpc_exec_ctx_finish(&exec_ctx);
 }
 
 /*
@@ -663,12 +667,11 @@ static void on_response_trailers_received(
     s->state.state_op_done[OP_SEND_TRAILING_METADATA] = true;
 
     gpr_mu_unlock(&s->mu);
-    grpc_exec_ctx_finish(&exec_ctx);
   } else {
     gpr_mu_unlock(&s->mu);
-    grpc_exec_ctx_finish(&exec_ctx);
-    execute_from_storage(s);
+    execute_from_storage(&exec_ctx, s);
   }
+  grpc_exec_ctx_finish(&exec_ctx);
 }
 
 /*
@@ -1411,7 +1414,7 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
   }
   stream_obj *s = (stream_obj *)gs;
   add_to_storage(s, op);
-  execute_from_storage(s);
+  execute_from_storage(exec_ctx, s);
 }
 
 static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,