aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib')
-rw-r--r--src/core/lib/iomgr/iomgr_uv.c4
-rw-r--r--src/core/lib/iomgr/iomgr_uv.h37
-rw-r--r--src/core/lib/iomgr/pollset_uv.c7
-rw-r--r--src/core/lib/iomgr/resolve_address_uv.c4
-rw-r--r--src/core/lib/iomgr/tcp_client_uv.c3
-rw-r--r--src/core/lib/iomgr/tcp_server_uv.c6
-rw-r--r--src/core/lib/iomgr/tcp_uv.c3
-rw-r--r--src/core/lib/iomgr/timer_uv.c4
8 files changed, 68 insertions, 0 deletions
diff --git a/src/core/lib/iomgr/iomgr_uv.c b/src/core/lib/iomgr/iomgr_uv.c
index 7a99a6efdd..c484a39c54 100644
--- a/src/core/lib/iomgr/iomgr_uv.c
+++ b/src/core/lib/iomgr/iomgr_uv.c
@@ -22,14 +22,18 @@
#include "src/core/lib/debug/trace.h"
#include "src/core/lib/iomgr/executor.h"
+#include "src/core/lib/iomgr/iomgr_uv.h"
#include "src/core/lib/iomgr/pollset_uv.h"
#include "src/core/lib/iomgr/tcp_uv.h"
+gpr_thd_id grpc_init_thread;
+
void grpc_iomgr_platform_init(void) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_pollset_global_init();
grpc_register_tracer("tcp", &grpc_tcp_trace);
grpc_executor_set_threading(&exec_ctx, false);
+ grpc_init_thread = gpr_thd_currentid();
grpc_exec_ctx_finish(&exec_ctx);
}
void grpc_iomgr_platform_flush(void) {}
diff --git a/src/core/lib/iomgr/iomgr_uv.h b/src/core/lib/iomgr/iomgr_uv.h
new file mode 100644
index 0000000000..35c073a033
--- /dev/null
+++ b/src/core/lib/iomgr/iomgr_uv.h
@@ -0,0 +1,37 @@
+/*
+ *
+ * Copyright 2017 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPC_CORE_LIB_IOMGR_IOMGR_UV_H
+#define GRPC_CORE_LIB_IOMGR_IOMGR_UV_H
+
+#include "src/core/lib/iomgr/iomgr_internal.h"
+
+#include <grpc/support/thd.h>
+
+/* The thread ID of the thread on which grpc was initialized. Used to verify
+ * that all calls into libuv are made on that same thread */
+extern gpr_thd_id grpc_init_thread;
+
+#ifdef GRPC_UV_THREAD_CHECK
+#define GRPC_ASSERT_SAME_THREAD() \
+ GPR_ASSERT(gpr_thd_currentid() == grpc_init_thread)
+#else
+#define GRPC_ASSERT_SAME_THREAD()
+#endif /* GRPC_UV_THREAD_CHECK */
+
+#endif /* GRPC_CORE_LIB_IOMGR_IOMGR_UV_H */
diff --git a/src/core/lib/iomgr/pollset_uv.c b/src/core/lib/iomgr/pollset_uv.c
index 1a54065a91..47c8c79cc2 100644
--- a/src/core/lib/iomgr/pollset_uv.c
+++ b/src/core/lib/iomgr/pollset_uv.c
@@ -28,6 +28,7 @@
#include <grpc/support/log.h>
#include <grpc/support/sync.h>
+#include "src/core/lib/iomgr/iomgr_uv.h"
#include "src/core/lib/iomgr/pollset.h"
#include "src/core/lib/iomgr/pollset_uv.h"
@@ -69,6 +70,7 @@ void grpc_pollset_global_init(void) {
}
void grpc_pollset_global_shutdown(void) {
+ GRPC_ASSERT_SAME_THREAD();
gpr_mu_destroy(&grpc_polling_mu);
uv_close((uv_handle_t *)dummy_uv_handle, dummy_handle_close_cb);
}
@@ -78,6 +80,7 @@ static void timer_run_cb(uv_timer_t *timer) {}
static void timer_close_cb(uv_handle_t *handle) { handle->data = (void *)1; }
void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu) {
+ GRPC_ASSERT_SAME_THREAD();
*mu = &grpc_polling_mu;
uv_timer_init(uv_default_loop(), &pollset->timer);
pollset->shutting_down = 0;
@@ -86,6 +89,7 @@ void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu) {
void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_closure *closure) {
GPR_ASSERT(!pollset->shutting_down);
+ GRPC_ASSERT_SAME_THREAD();
pollset->shutting_down = 1;
if (grpc_pollset_work_run_loop) {
// Drain any pending UV callbacks without blocking
@@ -98,6 +102,7 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
}
void grpc_pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
+ GRPC_ASSERT_SAME_THREAD();
uv_close((uv_handle_t *)&pollset->timer, timer_close_cb);
// timer.data is a boolean indicating that the timer has finished closing
pollset->timer.data = (void *)0;
@@ -112,6 +117,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) {
uint64_t timeout;
+ GRPC_ASSERT_SAME_THREAD();
gpr_mu_unlock(&grpc_polling_mu);
if (grpc_pollset_work_run_loop) {
if (gpr_time_cmp(deadline, now) >= 0) {
@@ -140,6 +146,7 @@ grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_error *grpc_pollset_kick(grpc_pollset *pollset,
grpc_pollset_worker *specific_worker) {
+ GRPC_ASSERT_SAME_THREAD();
uv_timer_start(dummy_uv_handle, dummy_timer_cb, 0, 0);
return GRPC_ERROR_NONE;
}
diff --git a/src/core/lib/iomgr/resolve_address_uv.c b/src/core/lib/iomgr/resolve_address_uv.c
index 5180357cb1..f910ec26b2 100644
--- a/src/core/lib/iomgr/resolve_address_uv.c
+++ b/src/core/lib/iomgr/resolve_address_uv.c
@@ -30,6 +30,7 @@
#include "src/core/lib/iomgr/closure.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/iomgr/iomgr_uv.h"
#include "src/core/lib/iomgr/resolve_address.h"
#include "src/core/lib/iomgr/sockaddr.h"
#include "src/core/lib/iomgr/sockaddr_utils.h"
@@ -174,6 +175,8 @@ static grpc_error *blocking_resolve_address_impl(
grpc_error *err;
int retry_status;
+ GRPC_ASSERT_SAME_THREAD();
+
req.addrinfo = NULL;
err = try_split_host_port(name, default_port, &host, &port);
@@ -228,6 +231,7 @@ static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name,
char *port = NULL;
grpc_error *err;
int s;
+ GRPC_ASSERT_SAME_THREAD();
err = try_split_host_port(name, default_port, &host, &port);
if (err != GRPC_ERROR_NONE) {
GRPC_CLOSURE_SCHED(exec_ctx, on_done, err);
diff --git a/src/core/lib/iomgr/tcp_client_uv.c b/src/core/lib/iomgr/tcp_client_uv.c
index 2f1d237c07..098ff7648d 100644
--- a/src/core/lib/iomgr/tcp_client_uv.c
+++ b/src/core/lib/iomgr/tcp_client_uv.c
@@ -26,6 +26,7 @@
#include <grpc/support/log.h>
#include "src/core/lib/iomgr/error.h"
+#include "src/core/lib/iomgr/iomgr_uv.h"
#include "src/core/lib/iomgr/sockaddr_utils.h"
#include "src/core/lib/iomgr/tcp_client.h"
#include "src/core/lib/iomgr/tcp_uv.h"
@@ -124,6 +125,8 @@ static void tcp_client_connect_impl(grpc_exec_ctx *exec_ctx,
(void)channel_args;
(void)interested_parties;
+ GRPC_ASSERT_SAME_THREAD();
+
if (channel_args != NULL) {
for (size_t i = 0; i < channel_args->num_args; i++) {
if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_RESOURCE_QUOTA)) {
diff --git a/src/core/lib/iomgr/tcp_server_uv.c b/src/core/lib/iomgr/tcp_server_uv.c
index a63a36bf58..7b45e08c9b 100644
--- a/src/core/lib/iomgr/tcp_server_uv.c
+++ b/src/core/lib/iomgr/tcp_server_uv.c
@@ -28,6 +28,7 @@
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/iomgr/iomgr_uv.h"
#include "src/core/lib/iomgr/sockaddr.h"
#include "src/core/lib/iomgr/sockaddr_utils.h"
#include "src/core/lib/iomgr/tcp_server.h"
@@ -107,6 +108,7 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx,
}
grpc_tcp_server *grpc_tcp_server_ref(grpc_tcp_server *s) {
+ GRPC_ASSERT_SAME_THREAD();
gpr_ref(&s->refs);
return s;
}
@@ -173,6 +175,7 @@ static void tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
}
void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
+ GRPC_ASSERT_SAME_THREAD();
if (gpr_unref(&s->refs)) {
/* Complete shutdown_starting work before destroying. */
grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT;
@@ -335,6 +338,8 @@ grpc_error *grpc_tcp_server_add_port(grpc_tcp_server *s,
int status;
grpc_error *error = GRPC_ERROR_NONE;
+ GRPC_ASSERT_SAME_THREAD();
+
if (s->tail != NULL) {
port_index = s->tail->port_index + 1;
}
@@ -415,6 +420,7 @@ void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *server,
grpc_tcp_listener *sp;
(void)pollsets;
(void)pollset_count;
+ GRPC_ASSERT_SAME_THREAD();
if (GRPC_TRACER_ON(grpc_tcp_trace)) {
gpr_log(GPR_DEBUG, "SERVER_START %p", server);
}
diff --git a/src/core/lib/iomgr/tcp_uv.c b/src/core/lib/iomgr/tcp_uv.c
index 0302af2a06..115297634b 100644
--- a/src/core/lib/iomgr/tcp_uv.c
+++ b/src/core/lib/iomgr/tcp_uv.c
@@ -30,6 +30,7 @@
#include <grpc/support/string_util.h>
#include "src/core/lib/iomgr/error.h"
+#include "src/core/lib/iomgr/iomgr_uv.h"
#include "src/core/lib/iomgr/network_status_tracker.h"
#include "src/core/lib/iomgr/resource_quota.h"
#include "src/core/lib/iomgr/tcp_uv.h"
@@ -183,6 +184,7 @@ static void uv_endpoint_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_tcp *tcp = (grpc_tcp *)ep;
int status;
grpc_error *error = GRPC_ERROR_NONE;
+ GRPC_ASSERT_SAME_THREAD();
GPR_ASSERT(tcp->read_cb == NULL);
tcp->read_cb = cb;
tcp->read_slices = read_slices;
@@ -236,6 +238,7 @@ static void uv_endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
unsigned int i;
grpc_slice *slice;
uv_write_t *write_req;
+ GRPC_ASSERT_SAME_THREAD();
if (GRPC_TRACER_ON(grpc_tcp_trace)) {
size_t j;
diff --git a/src/core/lib/iomgr/timer_uv.c b/src/core/lib/iomgr/timer_uv.c
index 4f204cfbf8..e004ff5a3a 100644
--- a/src/core/lib/iomgr/timer_uv.c
+++ b/src/core/lib/iomgr/timer_uv.c
@@ -24,6 +24,7 @@
#include <grpc/support/log.h>
#include "src/core/lib/debug/trace.h"
+#include "src/core/lib/iomgr/iomgr_uv.h"
#include "src/core/lib/iomgr/timer.h"
#include <uv.h>
@@ -42,6 +43,7 @@ static void stop_uv_timer(uv_timer_t *handle) {
void run_expired_timer(uv_timer_t *handle) {
grpc_timer *timer = (grpc_timer *)handle->data;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+ GRPC_ASSERT_SAME_THREAD();
GPR_ASSERT(timer->pending);
timer->pending = 0;
GRPC_CLOSURE_SCHED(&exec_ctx, timer->closure, GRPC_ERROR_NONE);
@@ -54,6 +56,7 @@ void grpc_timer_init(grpc_exec_ctx *exec_ctx, grpc_timer *timer,
gpr_timespec now) {
uint64_t timeout;
uv_timer_t *uv_timer;
+ GRPC_ASSERT_SAME_THREAD();
timer->closure = closure;
if (gpr_time_cmp(deadline, now) <= 0) {
timer->pending = 0;
@@ -74,6 +77,7 @@ void grpc_timer_init(grpc_exec_ctx *exec_ctx, grpc_timer *timer,
}
void grpc_timer_cancel(grpc_exec_ctx *exec_ctx, grpc_timer *timer) {
+ GRPC_ASSERT_SAME_THREAD();
if (timer->pending) {
timer->pending = 0;
GRPC_CLOSURE_SCHED(exec_ctx, timer->closure, GRPC_ERROR_CANCELLED);
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-18 13:16:23 +0000