1 /*
2 *
3 * Copyright 2015 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19 /* Posix implementation for gpr threads. */
20
21 #include <grpc/support/port_platform.h>
22
23 #ifdef GPR_POSIX_SYNC
24
25 #include "src/core/lib/gprpp/thd.h"
26
27 #include <grpc/support/alloc.h>
28 #include <grpc/support/log.h>
29 #include <grpc/support/sync.h>
30 #include <grpc/support/thd_id.h>
31 #include <pthread.h>
32 #include <stdlib.h>
33 #include <string.h>
34
35 #include "src/core/lib/gpr/useful.h"
36 #include "src/core/lib/gprpp/fork.h"
37 #include "src/core/lib/gprpp/memory.h"
38
39 namespace grpc_core {
40 namespace {
41 class ThreadInternalsPosix;
42 struct thd_arg {
43 ThreadInternalsPosix* thread;
44 void (*body)(void* arg); /* body of a thread */
45 void* arg; /* argument to a thread */
46 const char* name; /* name of thread. Can be nullptr. */
47 };
48
49 class ThreadInternalsPosix
50 : public grpc_core::internal::ThreadInternalsInterface {
51 public:
ThreadInternalsPosix(const char * thd_name,void (* thd_body)(void * arg),void * arg,bool * success)52 ThreadInternalsPosix(const char* thd_name, void (*thd_body)(void* arg),
53 void* arg, bool* success)
54 : started_(false) {
55 gpr_mu_init(&mu_);
56 gpr_cv_init(&ready_);
57 pthread_attr_t attr;
58 /* don't use gpr_malloc as we may cause an infinite recursion with
59 * the profiling code */
60 thd_arg* info = static_cast<thd_arg*>(malloc(sizeof(*info)));
61 GPR_ASSERT(info != nullptr);
62 info->thread = this;
63 info->body = thd_body;
64 info->arg = arg;
65 info->name = thd_name;
66 grpc_core::Fork::IncThreadCount();
67
68 GPR_ASSERT(pthread_attr_init(&attr) == 0);
69 GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE) ==
70 0);
71
72 *success =
73 (pthread_create(&pthread_id_, &attr,
74 [](void* v) -> void* {
75 thd_arg arg = *static_cast<thd_arg*>(v);
76 free(v);
77 if (arg.name != nullptr) {
78 #if GPR_APPLE_PTHREAD_NAME
79 /* Apple supports 64 characters, and will
80 * truncate if it's longer. */
81 pthread_setname_np(arg.name);
82 #elif GPR_LINUX_PTHREAD_NAME
83 /* Linux supports 16 characters max, and will
84 * error if it's longer. */
85 char buf[16];
86 size_t buf_len = GPR_ARRAY_SIZE(buf) - 1;
87 strncpy(buf, arg.name, buf_len);
88 buf[buf_len] = '\0';
89 pthread_setname_np(pthread_self(), buf);
90 #endif // GPR_APPLE_PTHREAD_NAME
91 }
92
93 gpr_mu_lock(&arg.thread->mu_);
94 while (!arg.thread->started_) {
95 gpr_cv_wait(&arg.thread->ready_, &arg.thread->mu_,
96 gpr_inf_future(GPR_CLOCK_MONOTONIC));
97 }
98 gpr_mu_unlock(&arg.thread->mu_);
99
100 (*arg.body)(arg.arg);
101 grpc_core::Fork::DecThreadCount();
102 return nullptr;
103 },
104 info) == 0);
105
106 GPR_ASSERT(pthread_attr_destroy(&attr) == 0);
107
108 if (!(*success)) {
109 /* don't use gpr_free, as this was allocated using malloc (see above) */
110 free(info);
111 grpc_core::Fork::DecThreadCount();
112 }
113 };
114
~ThreadInternalsPosix()115 ~ThreadInternalsPosix() override {
116 gpr_mu_destroy(&mu_);
117 gpr_cv_destroy(&ready_);
118 }
119
Start()120 void Start() override {
121 gpr_mu_lock(&mu_);
122 started_ = true;
123 gpr_cv_signal(&ready_);
124 gpr_mu_unlock(&mu_);
125 }
126
Join()127 void Join() override { pthread_join(pthread_id_, nullptr); }
128
129 private:
130 gpr_mu mu_;
131 gpr_cv ready_;
132 bool started_;
133 pthread_t pthread_id_;
134 };
135
136 } // namespace
137
Thread(const char * thd_name,void (* thd_body)(void * arg),void * arg,bool * success)138 Thread::Thread(const char* thd_name, void (*thd_body)(void* arg), void* arg,
139 bool* success) {
140 bool outcome = false;
141 impl_ =
142 grpc_core::New<ThreadInternalsPosix>(thd_name, thd_body, arg, &outcome);
143 if (outcome) {
144 state_ = ALIVE;
145 } else {
146 state_ = FAILED;
147 grpc_core::Delete(impl_);
148 impl_ = nullptr;
149 }
150
151 if (success != nullptr) {
152 *success = outcome;
153 }
154 }
155 } // namespace grpc_core
156
157 // The following is in the external namespace as it is exposed as C89 API
gpr_thd_currentid(void)158 gpr_thd_id gpr_thd_currentid(void) { return (gpr_thd_id)pthread_self(); }
159
160 #endif /* GPR_POSIX_SYNC */
161