|
|
|
@ -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, |
|
|
|
|