diff options
Diffstat (limited to 'src')
21 files changed, 217 insertions, 160 deletions
diff --git a/src/core/lib/gpr/thd.h b/src/core/lib/gpr/thd.h deleted file mode 100644 index 9d8140114c..0000000000 --- a/src/core/lib/gpr/thd.h +++ /dev/null @@ -1,48 +0,0 @@ -/* - * - * Copyright 2015 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_GPR_THD_H -#define GRPC_CORE_LIB_GPR_THD_H -/** Internal thread interface for GPR. - - Types - gpr_thd_options options used when creating a thread - */ - -#include <grpc/support/port_platform.h> -#include <grpc/support/thd_id.h> -#include <grpc/support/time.h> - -/** Create a new thread running (*thd_body)(arg) and place its thread identifier - in *t, and return true. If there are insufficient resources, return false. - thd_name is the name of the thread for identification purposes on platforms - that support thread naming. - The thread must be joined. */ -int gpr_thd_new(gpr_thd_id* t, const char* thd_name, - void (*thd_body)(void* arg), void* arg); - -/** Blocks until the specified thread properly terminates. */ -void gpr_thd_join(gpr_thd_id t); - -/* Internal interfaces between modules within the gpr support library. */ -void gpr_thd_init(); - -/* Wait for all outstanding threads to finish, up to deadline */ -int gpr_await_threads(gpr_timespec deadline); - -#endif /* GRPC_CORE_LIB_GPR_THD_H */ diff --git a/src/core/lib/gprpp/thd.h b/src/core/lib/gprpp/thd.h new file mode 100644 index 0000000000..c453b8f758 --- /dev/null +++ b/src/core/lib/gprpp/thd.h @@ -0,0 +1,79 @@ +/* + * + * Copyright 2015 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_GPRPP_THD_H +#define GRPC_CORE_LIB_GPRPP_THD_H + +/** Internal thread interface. */ + +#include <grpc/support/port_platform.h> + +#include <grpc/support/log.h> +#include <grpc/support/sync.h> +#include <grpc/support/thd_id.h> +#include <grpc/support/time.h> + +namespace grpc_core { + +class Thread { + public: + /// Default constructor only to allow use in structs that lack constructors + /// Does not produce a validly-constructed thread; must later + /// use placement new to construct a real thread. Does not init mu_ and cv_ + Thread(): real_(false), alive_(false), started_(false), joined_(false) {} + + Thread(const char* thd_name, void (*thd_body)(void* arg), void* arg, + bool* success = nullptr); + ~Thread() { + if (!alive_) { + // This thread never existed, so nothing to do + } else { + GPR_ASSERT(joined_); + } + if (real_) { + gpr_mu_destroy(&mu_); + gpr_cv_destroy(&ready_); + } + } + + void Start() { + gpr_mu_lock(&mu_); + if (alive_) { + started_ = true; + gpr_cv_signal(&ready_); + } + gpr_mu_unlock(&mu_); + } + void Join(); + + static void Init(); + static bool AwaitAll(gpr_timespec deadline); + private: + gpr_mu mu_; + gpr_cv ready_; + + gpr_thd_id id_; + bool real_; + bool alive_; + bool started_; + bool joined_; +}; + +} // namespace grpc_core + +#endif /* GRPC_CORE_LIB_GPRPP_THD_H */ diff --git a/src/core/lib/gpr/thd_posix.cc b/src/core/lib/gprpp/thd_posix.cc index a5640d8cc4..1d166bb405 100644 --- a/src/core/lib/gpr/thd_posix.cc +++ b/src/core/lib/gprpp/thd_posix.cc @@ -22,7 +22,7 @@ #ifdef GPR_POSIX_SYNC -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> @@ -35,51 +35,56 @@ #include "src/core/lib/gpr/fork.h" #include "src/core/lib/gpr/useful.h" -static gpr_mu g_mu; -static gpr_cv g_cv; -static int g_thread_count; -static int g_awaiting_threads; +namespace grpc_core { +namespace { +gpr_mu g_mu; +gpr_cv g_cv; +int g_thread_count; +int g_awaiting_threads; struct thd_arg { + Thread* thread; void (*body)(void* arg); /* body of a thread */ void* arg; /* argument to a thread */ const char* name; /* name of thread. Can be nullptr. */ }; -static void inc_thd_count(); -static void dec_thd_count(); +/***************************************** + * Only used when fork support is enabled + */ -/* Body of every thread started via gpr_thd_new. */ -static void* thread_body(void* v) { - struct thd_arg a = *static_cast<struct thd_arg*>(v); - free(v); - if (a.name != nullptr) { -#if GPR_APPLE_PTHREAD_NAME - /* Apple supports 64 characters, and will truncate if it's longer. */ - pthread_setname_np(a.name); -#elif GPR_LINUX_PTHREAD_NAME - /* Linux supports 16 characters max, and will error if it's longer. */ - char buf[16]; - size_t buf_len = GPR_ARRAY_SIZE(buf) - 1; - strncpy(buf, a.name, buf_len); - buf[buf_len] = '\0'; - pthread_setname_np(pthread_self(), buf); -#endif // GPR_APPLE_PTHREAD_NAME +void inc_thd_count() { + if (grpc_fork_support_enabled()) { + gpr_mu_lock(&g_mu); + g_thread_count++; + gpr_mu_unlock(&g_mu); + } +} + +void dec_thd_count() { + if (grpc_fork_support_enabled()) { + gpr_mu_lock(&g_mu); + g_thread_count--; + if (g_awaiting_threads && g_thread_count == 0) { + gpr_cv_signal(&g_cv); + } + gpr_mu_unlock(&g_mu); } - (*a.body)(a.arg); - dec_thd_count(); - return nullptr; } -int gpr_thd_new(gpr_thd_id* t, const char* thd_name, - void (*thd_body)(void* arg), void* arg) { - int thread_started; +} // namespace + +Thread::Thread(const char* thd_name, void (*thd_body)(void* arg), void* arg, + bool* success): + real_(true), alive_(false), started_(false), joined_(false) { + gpr_mu_init(&mu_); + gpr_cv_init(&ready_); pthread_attr_t attr; - pthread_t p; /* don't use gpr_malloc as we may cause an infinite recursion with * the profiling code */ - struct thd_arg* a = static_cast<struct thd_arg*>(malloc(sizeof(*a))); + thd_arg* a = static_cast<thd_arg*>(malloc(sizeof(*a))); GPR_ASSERT(a != nullptr); + a->thread = this; a->body = thd_body; a->arg = arg; a->name = thd_name; @@ -88,54 +93,63 @@ int gpr_thd_new(gpr_thd_id* t, const char* thd_name, GPR_ASSERT(pthread_attr_init(&attr) == 0); GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE) == 0); - thread_started = (pthread_create(&p, &attr, &thread_body, a) == 0); - + pthread_t p; + alive_ = (pthread_create(&p, &attr, [](void *v) -> void* { + thd_arg a = *static_cast<thd_arg*>(v); + free(v); + if (a.name != nullptr) { +#if GPR_APPLE_PTHREAD_NAME + /* Apple supports 64 characters, and will truncate if it's longer. */ + pthread_setname_np(a.name); +#elif GPR_LINUX_PTHREAD_NAME + /* Linux supports 16 characters max, and will error if it's longer. */ + char buf[16]; + size_t buf_len = GPR_ARRAY_SIZE(buf) - 1; + strncpy(buf, a.name, buf_len); + buf[buf_len] = '\0'; + pthread_setname_np(pthread_self(), buf); +#endif // GPR_APPLE_PTHREAD_NAME + } + + gpr_mu_lock(&a.thread->mu_); + if (!a.thread->started_) { + gpr_cv_wait(&a.thread->ready_, &a.thread->mu_, + gpr_inf_future(GPR_CLOCK_MONOTONIC)); + } + gpr_mu_unlock(&a.thread->mu_); + + (*a.body)(a.arg); + dec_thd_count(); + return nullptr; + }, a) == 0); + + if (success != nullptr) { *success = alive_; } + + id_ = reinterpret_cast<gpr_thd_id>(p); GPR_ASSERT(pthread_attr_destroy(&attr) == 0); - if (!thread_started) { + if (!alive_) { /* don't use gpr_free, as this was allocated using malloc (see above) */ free(a); dec_thd_count(); } - *t = (gpr_thd_id)p; - return thread_started; } -gpr_thd_id gpr_thd_currentid(void) { return (gpr_thd_id)pthread_self(); } - -void gpr_thd_join(gpr_thd_id t) { pthread_join((pthread_t)t, nullptr); } - -/***************************************** - * Only used when fork support is enabled - */ - -static void inc_thd_count() { - if (grpc_fork_support_enabled()) { - gpr_mu_lock(&g_mu); - g_thread_count++; - gpr_mu_unlock(&g_mu); +void Thread::Join() { + if (alive_) { + pthread_join(reinterpret_cast<pthread_t>(id_), nullptr); } + joined_ = true; } -static void dec_thd_count() { - if (grpc_fork_support_enabled()) { - gpr_mu_lock(&g_mu); - g_thread_count--; - if (g_awaiting_threads && g_thread_count == 0) { - gpr_cv_signal(&g_cv); - } - gpr_mu_unlock(&g_mu); - } -} - -void gpr_thd_init() { +void Thread::Init() { gpr_mu_init(&g_mu); gpr_cv_init(&g_cv); g_thread_count = 0; g_awaiting_threads = 0; } -int gpr_await_threads(gpr_timespec deadline) { +bool Thread::AwaitAll(gpr_timespec deadline) { gpr_mu_lock(&g_mu); g_awaiting_threads = 1; int res = 0; @@ -147,4 +161,9 @@ int gpr_await_threads(gpr_timespec deadline) { return res == 0; } +} // namespace grpc_core + +// The following is in the external namespace as it is exposed as C89 API +gpr_thd_id gpr_thd_currentid(void) { return (gpr_thd_id)pthread_self(); } + #endif /* GPR_POSIX_SYNC */ diff --git a/src/core/lib/gpr/thd_windows.cc b/src/core/lib/gprpp/thd_windows.cc index 53093e80f3..14310caa59 100644 --- a/src/core/lib/gpr/thd_windows.cc +++ b/src/core/lib/gprpp/thd_windows.cc @@ -22,7 +22,7 @@ #ifdef GPR_WINDOWS -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> diff --git a/src/core/lib/iomgr/ev_poll_posix.cc b/src/core/lib/iomgr/ev_poll_posix.cc index 0d1e67dfdf..4553845309 100644 --- a/src/core/lib/iomgr/ev_poll_posix.cc +++ b/src/core/lib/iomgr/ev_poll_posix.cc @@ -22,6 +22,7 @@ #include "src/core/lib/iomgr/ev_poll_posix.h" +#include <new> #include <assert.h> #include <errno.h> #include <limits.h> @@ -36,7 +37,7 @@ #include "src/core/lib/debug/stats.h" #include "src/core/lib/gpr/murmur_hash.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/gpr/tls.h" #include "src/core/lib/gpr/useful.h" #include "src/core/lib/iomgr/block_annotate.h" @@ -253,7 +254,7 @@ typedef struct poll_result { } poll_result; typedef struct poll_args { - gpr_thd_id poller_thd; + grpc_core::Thread poller_thd; gpr_cv trigger; int trigger_set; gpr_cv harvest; @@ -1377,7 +1378,8 @@ static poll_args* get_poller_locked(struct pollfd* fds, nfds_t count) { init_result(pargs); cache_poller_locked(pargs); gpr_ref(&g_cvfds.pollcount); - GPR_ASSERT(gpr_thd_new(&pargs->poller_thd, "grpc_poller", &run_poll, pargs)); + new (&pargs->poller_thd) grpc_core::Thread("grpc_poller", &run_poll, pargs); + pargs->poller_thd.Start(); return pargs; } @@ -1462,7 +1464,8 @@ static void cache_harvest_locked() { } gpr_cv_signal(&args->harvest); gpr_cv_wait(&args->join, &g_cvfds.mu, gpr_inf_future(GPR_CLOCK_MONOTONIC)); - gpr_thd_join(args->poller_thd); + args->poller_thd.Join(); + args->poller_thd.~Thread(); gpr_free(args); } } diff --git a/src/core/lib/iomgr/exec_ctx.cc b/src/core/lib/iomgr/exec_ctx.cc index 1a2284f474..e9fa08aeb8 100644 --- a/src/core/lib/iomgr/exec_ctx.cc +++ b/src/core/lib/iomgr/exec_ctx.cc @@ -21,7 +21,7 @@ #include <grpc/support/log.h> #include <grpc/support/sync.h> -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/profiling/timers.h" diff --git a/src/core/lib/iomgr/executor.cc b/src/core/lib/iomgr/executor.cc index baf0cfc48f..ad1b306ffa 100644 --- a/src/core/lib/iomgr/executor.cc +++ b/src/core/lib/iomgr/executor.cc @@ -18,6 +18,7 @@ #include "src/core/lib/iomgr/executor.h" +#include <new> #include <string.h> #include <grpc/support/alloc.h> @@ -27,7 +28,7 @@ #include "src/core/lib/debug/stats.h" #include "src/core/lib/gpr/spinlock.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/gpr/tls.h" #include "src/core/lib/gpr/useful.h" #include "src/core/lib/iomgr/exec_ctx.h" @@ -41,7 +42,7 @@ typedef struct { size_t depth; bool shutdown; bool queued_long_job; - gpr_thd_id id; + grpc_core::Thread thd; } thread_state; static thread_state* g_thread_state; @@ -99,11 +100,14 @@ void grpc_executor_set_threading(bool threading) { for (size_t i = 0; i < g_max_threads; i++) { gpr_mu_init(&g_thread_state[i].mu); gpr_cv_init(&g_thread_state[i].cv); + new (&g_thread_state[i].thd) grpc_core::Thread(); g_thread_state[i].elems = GRPC_CLOSURE_LIST_INIT; } - gpr_thd_new(&g_thread_state[0].id, "grpc_executor", executor_thread, - &g_thread_state[0]); + new (&g_thread_state[0].thd) grpc_core::Thread("grpc_executor", + executor_thread, + &g_thread_state[0]); + g_thread_state[0].thd.Start(); } else { if (cur_threads == 0) return; for (size_t i = 0; i < g_max_threads; i++) { @@ -117,10 +121,11 @@ void grpc_executor_set_threading(bool threading) { gpr_spinlock_lock(&g_adding_thread_lock); gpr_spinlock_unlock(&g_adding_thread_lock); for (gpr_atm i = 0; i < g_cur_threads; i++) { - gpr_thd_join(g_thread_state[i].id); + g_thread_state[i].thd.Join(); } gpr_atm_no_barrier_store(&g_cur_threads, 0); for (size_t i = 0; i < g_max_threads; i++) { + g_thread_state[i].thd.~Thread(); gpr_mu_destroy(&g_thread_state[i].mu); gpr_cv_destroy(&g_thread_state[i].cv); run_closures(g_thread_state[i].elems); @@ -260,8 +265,10 @@ static void executor_push(grpc_closure* closure, grpc_error* error, if (cur_thread_count < g_max_threads) { gpr_atm_no_barrier_store(&g_cur_threads, cur_thread_count + 1); - gpr_thd_new(&g_thread_state[cur_thread_count].id, "gpr_executor", - executor_thread, &g_thread_state[cur_thread_count]); + new (&g_thread_state[cur_thread_count].thd) + grpc_core::Thread("grpc_executor", executor_thread, + &g_thread_state[cur_thread_count]); + g_thread_state[cur_thread_count].thd.Start(); } gpr_spinlock_unlock(&g_adding_thread_lock); } diff --git a/src/core/lib/iomgr/fork_posix.cc b/src/core/lib/iomgr/fork_posix.cc index c9a65c5702..e8ba65d1d3 100644 --- a/src/core/lib/iomgr/fork_posix.cc +++ b/src/core/lib/iomgr/fork_posix.cc @@ -27,7 +27,7 @@ #include "src/core/lib/gpr/env.h" #include "src/core/lib/gpr/fork.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/iomgr/ev_posix.h" #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/iomgr/timer_manager.h" @@ -51,7 +51,7 @@ void grpc_prefork() { grpc_timer_manager_set_threading(false); grpc_executor_set_threading(false); grpc_core::ExecCtx::Get()->Flush(); - if (!gpr_await_threads( + if (!grpc_core::Thread::AwaitAll( gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_seconds(3, GPR_TIMESPAN)))) { gpr_log(GPR_ERROR, "gRPC thread still active! Cannot fork!"); diff --git a/src/core/lib/iomgr/iocp_windows.cc b/src/core/lib/iomgr/iocp_windows.cc index 4716872ce4..24af0c1db8 100644 --- a/src/core/lib/iomgr/iocp_windows.cc +++ b/src/core/lib/iomgr/iocp_windows.cc @@ -28,7 +28,7 @@ #include <grpc/support/log_windows.h> #include "src/core/lib/debug/stats.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/iomgr/iocp_windows.h" #include "src/core/lib/iomgr/iomgr_internal.h" #include "src/core/lib/iomgr/socket_windows.h" diff --git a/src/core/lib/iomgr/iomgr.cc b/src/core/lib/iomgr/iomgr.cc index 70a80e1998..072670f8be 100644 --- a/src/core/lib/iomgr/iomgr.cc +++ b/src/core/lib/iomgr/iomgr.cc @@ -31,7 +31,7 @@ #include "src/core/lib/gpr/env.h" #include "src/core/lib/gpr/string.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/gpr/useful.h" #include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/iomgr/executor.h" diff --git a/src/core/lib/iomgr/pollset_windows.cc b/src/core/lib/iomgr/pollset_windows.cc index 240a24dee1..7c4196c064 100644 --- a/src/core/lib/iomgr/pollset_windows.cc +++ b/src/core/lib/iomgr/pollset_windows.cc @@ -22,7 +22,7 @@ #include <grpc/support/log.h> -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/iomgr/iocp_windows.h" #include "src/core/lib/iomgr/iomgr_internal.h" #include "src/core/lib/iomgr/pollset.h" diff --git a/src/core/lib/iomgr/resolve_address_posix.cc b/src/core/lib/iomgr/resolve_address_posix.cc index 3dc1d871a1..6509ec1de9 100644 --- a/src/core/lib/iomgr/resolve_address_posix.cc +++ b/src/core/lib/iomgr/resolve_address_posix.cc @@ -33,7 +33,7 @@ #include "src/core/lib/gpr/host_port.h" #include "src/core/lib/gpr/string.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/gpr/useful.h" #include "src/core/lib/iomgr/block_annotate.h" #include "src/core/lib/iomgr/executor.h" diff --git a/src/core/lib/iomgr/resolve_address_windows.cc b/src/core/lib/iomgr/resolve_address_windows.cc index 8f4fd04402..2044d85a94 100644 --- a/src/core/lib/iomgr/resolve_address_windows.cc +++ b/src/core/lib/iomgr/resolve_address_windows.cc @@ -35,7 +35,7 @@ #include "src/core/lib/gpr/host_port.h" #include "src/core/lib/gpr/string.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/iomgr/block_annotate.h" #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/iomgr/iomgr_internal.h" diff --git a/src/core/lib/iomgr/timer_manager.cc b/src/core/lib/iomgr/timer_manager.cc index 39ef575d12..f4c289af6a 100644 --- a/src/core/lib/iomgr/timer_manager.cc +++ b/src/core/lib/iomgr/timer_manager.cc @@ -18,20 +18,21 @@ #include "src/core/lib/iomgr/timer_manager.h" +#include <new> +#include <inttypes.h> + #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/port_platform.h> -#include <inttypes.h> - #include "src/core/lib/debug/trace.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/iomgr/timer.h" -typedef struct completed_thread { - gpr_thd_id t; - struct completed_thread* next; -} completed_thread; +struct completed_thread { + grpc_core::Thread thd; + completed_thread* next; +}; extern grpc_core::TraceFlag grpc_timer_check_trace; @@ -67,7 +68,8 @@ static void gc_completed_threads(void) { g_completed_threads = nullptr; gpr_mu_unlock(&g_mu); while (to_gc != nullptr) { - gpr_thd_join(to_gc->t); + to_gc->thd.Join(); + to_gc->thd.~Thread(); completed_thread* next = to_gc->next; gpr_free(to_gc); to_gc = next; @@ -86,14 +88,8 @@ static void start_timer_thread_and_unlock(void) { } completed_thread* ct = static_cast<completed_thread*>(gpr_malloc(sizeof(*ct))); - // The call to gpr_thd_new() has to be under the same lock used by - // gc_completed_threads(), particularly due to ct->t, which is written here - // (internally by gpr_thd_new) and read there. Otherwise it's possible for ct - // to leak through g_completed_threads and be freed in gc_completed_threads() - // before "&ct->t" is written to, causing a use-after-free. - gpr_mu_lock(&g_mu); - gpr_thd_new(&ct->t, "grpc_global_timer", timer_thread, ct); - gpr_mu_unlock(&g_mu); + new (&ct->thd) grpc_core::Thread("grpc_global_timer", timer_thread, ct); + ct->thd.Start(); } void grpc_timer_manager_tick() { diff --git a/src/core/lib/iomgr/wakeup_fd_cv.cc b/src/core/lib/iomgr/wakeup_fd_cv.cc index 41d35cb1fd..fd78d5ee03 100644 --- a/src/core/lib/iomgr/wakeup_fd_cv.cc +++ b/src/core/lib/iomgr/wakeup_fd_cv.cc @@ -30,7 +30,7 @@ #include <grpc/support/sync.h> #include <grpc/support/time.h> -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/gpr/useful.h" #define MAX_TABLE_RESIZE 256 diff --git a/src/core/lib/profiling/basic_timers.cc b/src/core/lib/profiling/basic_timers.cc index 07e21f5180..13ae1d8b7b 100644 --- a/src/core/lib/profiling/basic_timers.cc +++ b/src/core/lib/profiling/basic_timers.cc @@ -30,7 +30,7 @@ #include <string.h> #include "src/core/lib/gpr/env.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" typedef enum { BEGIN = '{', END = '}', MARK = '.' } marker_type; @@ -68,7 +68,7 @@ static pthread_cond_t g_cv; static gpr_timer_log_list g_in_progress_logs; static gpr_timer_log_list g_done_logs; static int g_shutdown; -static gpr_thd_id g_writing_thread; +static grpc_core::Thread* g_writing_thread; static __thread int g_thread_id; static int g_next_thread_id; static int g_writing_enabled = 1; @@ -182,7 +182,8 @@ static void finish_writing(void) { g_shutdown = 1; pthread_cond_signal(&g_cv); pthread_mutex_unlock(&g_mu); - gpr_thd_join(g_writing_thread); + g_writing_thread->Join(); + delete g_writing_thread; gpr_log(GPR_INFO, "flushing logs"); @@ -201,8 +202,8 @@ void gpr_timers_set_log_filename(const char* filename) { } static void init_output() { - GPR_ASSERT(gpr_thd_new(&g_writing_thread, "timer_output_thread", - writing_thread, NULL)); + g_writing_thread = new grpc_core::Thread("timer_output_thread", + writing_thread, nullptr); atexit(finish_writing); } diff --git a/src/core/lib/surface/init.cc b/src/core/lib/surface/init.cc index 7bc24a5049..ac9f9e6066 100644 --- a/src/core/lib/surface/init.cc +++ b/src/core/lib/surface/init.cc @@ -32,7 +32,7 @@ #include "src/core/lib/debug/stats.h" #include "src/core/lib/debug/trace.h" #include "src/core/lib/gpr/fork.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/http/parser.h" #include "src/core/lib/iomgr/call_combiner.h" #include "src/core/lib/iomgr/combiner.h" @@ -123,7 +123,7 @@ void grpc_init(void) { gpr_mu_lock(&g_init_mu); if (++g_initializations == 1) { gpr_time_init(); - gpr_thd_init(); + grpc_core::Thread::Init(); grpc_stats_init(); grpc_slice_intern_init(); grpc_mdctx_global_init(); diff --git a/src/core/tsi/alts_transport_security.cc b/src/core/tsi/alts_transport_security.cc index 1654d893d0..5c73c629cc 100644 --- a/src/core/tsi/alts_transport_security.cc +++ b/src/core/tsi/alts_transport_security.cc @@ -54,7 +54,7 @@ void grpc_tsi_alts_shutdown() { grpc_tsi_alts_wait_for_cq_drain(); grpc_completion_queue_destroy(g_alts_resource.cq); grpc_channel_destroy(g_alts_resource.channel); - gpr_thd_join(g_alts_resource.thread_id); + g_alts_resource.thread.Join(); } gpr_cv_destroy(&g_alts_resource.cv); gpr_mu_destroy(&g_alts_resource.mu); diff --git a/src/core/tsi/alts_transport_security.h b/src/core/tsi/alts_transport_security.h index 37febd1e28..c45ffd3fd6 100644 --- a/src/core/tsi/alts_transport_security.h +++ b/src/core/tsi/alts_transport_security.h @@ -22,10 +22,10 @@ #include <grpc/grpc.h> #include <grpc/support/sync.h> -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" typedef struct alts_shared_resource { - gpr_thd_id thread_id; + grpc_core::Thread thread; grpc_channel* channel; grpc_completion_queue* cq; gpr_mu mu; diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc index cba5984f4b..867f31f025 100644 --- a/src/cpp/client/channel_cc.cc +++ b/src/cpp/client/channel_cc.cc @@ -42,7 +42,7 @@ #include <grpcpp/support/time.h> #include "src/core/lib/gpr/env.h" #include "src/core/lib/gpr/string.h" -#include "src/core/lib/gpr/thd.h" +#include "src/core/lib/gprpp/thd.h" #include "src/core/lib/profiling/timers.h" namespace grpc { diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index df6adc4884..f26162f4a4 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -41,8 +41,6 @@ CORE_SOURCE_FILES = [ 'src/core/lib/gpr/sync.cc', 'src/core/lib/gpr/sync_posix.cc', 'src/core/lib/gpr/sync_windows.cc', - 'src/core/lib/gpr/thd_posix.cc', - 'src/core/lib/gpr/thd_windows.cc', 'src/core/lib/gpr/time.cc', 'src/core/lib/gpr/time_posix.cc', 'src/core/lib/gpr/time_precise.cc', @@ -52,6 +50,8 @@ CORE_SOURCE_FILES = [ 'src/core/lib/gpr/tmpfile_posix.cc', 'src/core/lib/gpr/tmpfile_windows.cc', 'src/core/lib/gpr/wrap_memcpy.cc', + 'src/core/lib/gprpp/thd_posix.cc', + 'src/core/lib/gprpp/thd_windows.cc', 'src/core/lib/profiling/basic_timers.cc', 'src/core/lib/profiling/stap_timers.cc', 'src/core/lib/surface/init.cc', |