aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/ext/filters/client_channel/client_channel.c20
-rw-r--r--src/core/ext/filters/client_channel/subchannel.c3
-rw-r--r--src/core/lib/iomgr/ev_epoll_linux.c4
-rw-r--r--src/core/lib/iomgr/ev_poll_posix.c4
-rw-r--r--src/core/lib/iomgr/pollset.h4
-rw-r--r--src/core/lib/iomgr/pollset_windows.c4
-rw-r--r--src/core/lib/surface/completion_queue.c5
7 files changed, 23 insertions, 21 deletions
diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c
index 93ad53aab9..83e3b8f118 100644
--- a/src/core/ext/filters/client_channel/client_channel.c
+++ b/src/core/ext/filters/client_channel/client_channel.c
@@ -914,14 +914,14 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
.arena = calld->arena};
grpc_error *new_error = grpc_connected_subchannel_create_call(
exec_ctx, calld->connected_subchannel, &call_args, &subchannel_call);
+ gpr_atm_rel_store(&calld->subchannel_call,
+ (gpr_atm)(uintptr_t)subchannel_call);
if (new_error != GRPC_ERROR_NONE) {
new_error = grpc_error_add_child(new_error, error);
- subchannel_call = CANCELLED_CALL;
fail_locked(exec_ctx, calld, new_error);
+ } else {
+ retry_waiting_locked(exec_ctx, calld);
}
- gpr_atm_rel_store(&calld->subchannel_call,
- (gpr_atm)(uintptr_t)subchannel_call);
- retry_waiting_locked(exec_ctx, calld);
}
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
}
@@ -1152,16 +1152,16 @@ static void start_transport_stream_op_batch_locked_inner(
.arena = calld->arena};
grpc_error *error = grpc_connected_subchannel_create_call(
exec_ctx, calld->connected_subchannel, &call_args, &subchannel_call);
+ gpr_atm_rel_store(&calld->subchannel_call,
+ (gpr_atm)(uintptr_t)subchannel_call);
if (error != GRPC_ERROR_NONE) {
- subchannel_call = CANCELLED_CALL;
fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error));
grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, op, error);
+ } else {
+ retry_waiting_locked(exec_ctx, calld);
+ /* recurse to retry */
+ start_transport_stream_op_batch_locked_inner(exec_ctx, op, elem);
}
- gpr_atm_rel_store(&calld->subchannel_call,
- (gpr_atm)(uintptr_t)subchannel_call);
- retry_waiting_locked(exec_ctx, calld);
- /* recurse to retry */
- start_transport_stream_op_batch_locked_inner(exec_ctx, op, elem);
/* early out */
return;
}
diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c
index 29a1a09555..9a7a7a0ee5 100644
--- a/src/core/ext/filters/client_channel/subchannel.c
+++ b/src/core/ext/filters/client_channel/subchannel.c
@@ -769,7 +769,7 @@ grpc_error *grpc_connected_subchannel_create_call(
*call = gpr_arena_alloc(
args->arena, sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(*call);
- (*call)->connection = con; // Ref is added below.
+ (*call)->connection = GRPC_CONNECTED_SUBCHANNEL_REF(con, "subchannel_call");
const grpc_call_element_args call_args = {.call_stack = callstk,
.server_transport_data = NULL,
.context = NULL,
@@ -784,7 +784,6 @@ grpc_error *grpc_connected_subchannel_create_call(
gpr_log(GPR_ERROR, "error: %s", error_string);
return error;
}
- GRPC_CONNECTED_SUBCHANNEL_REF(con, "subchannel_call");
grpc_call_stack_set_pollset_or_pollset_set(exec_ctx, callstk, args->pollent);
return GRPC_ERROR_NONE;
}
diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c
index 7014b98349..e5cf54f10a 100644
--- a/src/core/lib/iomgr/ev_epoll_linux.c
+++ b/src/core/lib/iomgr/ev_epoll_linux.c
@@ -1717,7 +1717,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
worker.pt_id = pthread_self();
gpr_atm_no_barrier_store(&worker.is_kicked, (gpr_atm)0);
- *worker_hdl = &worker;
+ if (worker_hdl) *worker_hdl = &worker;
gpr_tls_set(&g_current_thread_pollset, (intptr_t)pollset);
gpr_tls_set(&g_current_thread_worker, (intptr_t)&worker);
@@ -1795,7 +1795,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
gpr_mu_lock(&pollset->po.mu);
}
- *worker_hdl = NULL;
+ if (worker_hdl) *worker_hdl = NULL;
gpr_tls_set(&g_current_thread_pollset, (intptr_t)0);
gpr_tls_set(&g_current_thread_worker, (intptr_t)0);
diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c
index d90f223362..9834cdd197 100644
--- a/src/core/lib/iomgr/ev_poll_posix.c
+++ b/src/core/lib/iomgr/ev_poll_posix.c
@@ -871,7 +871,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_pollset_worker **worker_hdl,
gpr_timespec now, gpr_timespec deadline) {
grpc_pollset_worker worker;
- *worker_hdl = &worker;
+ if (worker_hdl) *worker_hdl = &worker;
grpc_error *error = GRPC_ERROR_NONE;
/* Avoid malloc for small number of elements. */
@@ -1092,7 +1092,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
gpr_mu_lock(&pollset->mu);
}
}
- *worker_hdl = NULL;
+ if (worker_hdl) *worker_hdl = NULL;
GPR_TIMER_END("pollset_work", 0);
GRPC_LOG_IF_ERROR("pollset_work", GRPC_ERROR_REF(error));
return error;
diff --git a/src/core/lib/iomgr/pollset.h b/src/core/lib/iomgr/pollset.h
index e19ce697b8..9bf3cdac89 100644
--- a/src/core/lib/iomgr/pollset.h
+++ b/src/core/lib/iomgr/pollset.h
@@ -75,6 +75,10 @@ void grpc_pollset_destroy(grpc_pollset *pollset);
and it is guaranteed that it will not be released by grpc_pollset_work
AFTER worker has been destroyed.
+ It's legal for worker to be NULL: in that case, this specific thread can not
+ be directly woken with a kick, but maybe be indirectly (with a kick against
+ the pollset as a whole).
+
Tries not to block past deadline.
May call grpc_closure_list_run on grpc_closure_list, without holding the
pollset
diff --git a/src/core/lib/iomgr/pollset_windows.c b/src/core/lib/iomgr/pollset_windows.c
index 17043c1ea1..04c6b71747 100644
--- a/src/core/lib/iomgr/pollset_windows.c
+++ b/src/core/lib/iomgr/pollset_windows.c
@@ -120,7 +120,7 @@ grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_pollset_worker **worker_hdl,
gpr_timespec now, gpr_timespec deadline) {
grpc_pollset_worker worker;
- *worker_hdl = &worker;
+ if (worker_hdl) *worker_hdl = &worker;
int added_worker = 0;
worker.links[GRPC_POLLSET_WORKER_LINK_POLLSET].next =
@@ -185,7 +185,7 @@ done:
remove_worker(&worker, GRPC_POLLSET_WORKER_LINK_POLLSET);
}
gpr_cv_destroy(&worker.cv);
- *worker_hdl = NULL;
+ if (worker_hdl) *worker_hdl = NULL;
return GRPC_ERROR_NONE;
}
diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c
index b4594817e4..3273addf1d 100644
--- a/src/core/lib/surface/completion_queue.c
+++ b/src/core/lib/surface/completion_queue.c
@@ -345,7 +345,6 @@ static void dump_pending_tags(grpc_completion_queue *cc) {}
grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
gpr_timespec deadline, void *reserved) {
grpc_event ret;
- grpc_pollset_worker *worker = NULL;
gpr_timespec now;
GPR_TIMER_BEGIN("grpc_completion_queue_next", 0);
@@ -426,8 +425,8 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
gpr_mu_lock(cc->mu);
continue;
} else {
- grpc_error *err = grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc),
- &worker, now, iteration_deadline);
+ grpc_error *err = grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc), NULL,
+ now, iteration_deadline);
if (err != GRPC_ERROR_NONE) {
gpr_mu_unlock(cc->mu);
const char *msg = grpc_error_string(err);
generated by cgit v1.2.3 (git 2.39.1) at 2024-09-13 21:49:06 +0000