• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 #include <unistd.h>
35 
36 #include "src/core/lib/gpr/useful.h"
37 #include "src/core/lib/gprpp/fork.h"
38 #include "src/core/lib/gprpp/memory.h"
39 
40 namespace grpc_core {
41 namespace {
42 class ThreadInternalsPosix;
43 struct thd_arg {
44   ThreadInternalsPosix* thread;
45   void (*body)(void* arg); /* body of a thread */
46   void* arg;               /* argument to a thread */
47   const char* name;        /* name of thread. Can be nullptr. */
48   bool joinable;
49   bool tracked;
50 };
51 
RoundUpToPageSize(size_t size)52 size_t RoundUpToPageSize(size_t size) {
53   // TODO(yunjiaw): Change this variable (page_size) to a function-level static
54   // when possible
55   size_t page_size = static_cast<size_t>(sysconf(_SC_PAGESIZE));
56   return (size + page_size - 1) & ~(page_size - 1);
57 }
58 
59 // Returns the minimum valid stack size that can be passed to
60 // pthread_attr_setstacksize.
MinValidStackSize(size_t request_size)61 size_t MinValidStackSize(size_t request_size) {
62   size_t min_stacksize = sysconf(_SC_THREAD_STACK_MIN);
63   if (request_size < min_stacksize) {
64     request_size = min_stacksize;
65   }
66 
67   // On some systems, pthread_attr_setstacksize() can fail if stacksize is
68   // not a multiple of the system page size.
69   return RoundUpToPageSize(request_size);
70 }
71 
72 class ThreadInternalsPosix : public internal::ThreadInternalsInterface {
73  public:
ThreadInternalsPosix(const char * thd_name,void (* thd_body)(void * arg),void * arg,bool * success,const Thread::Options & options)74   ThreadInternalsPosix(const char* thd_name, void (*thd_body)(void* arg),
75                        void* arg, bool* success, const Thread::Options& options)
76       : started_(false) {
77     gpr_mu_init(&mu_);
78     gpr_cv_init(&ready_);
79     pthread_attr_t attr;
80     /* don't use gpr_malloc as we may cause an infinite recursion with
81      * the profiling code */
82     thd_arg* info = static_cast<thd_arg*>(malloc(sizeof(*info)));
83     GPR_ASSERT(info != nullptr);
84     info->thread = this;
85     info->body = thd_body;
86     info->arg = arg;
87     info->name = thd_name;
88     info->joinable = options.joinable();
89     info->tracked = options.tracked();
90     if (options.tracked()) {
91       Fork::IncThreadCount();
92     }
93 
94     GPR_ASSERT(pthread_attr_init(&attr) == 0);
95     if (options.joinable()) {
96       GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE) ==
97                  0);
98     } else {
99       GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) ==
100                  0);
101     }
102 
103     if (options.stack_size() != 0) {
104       size_t stack_size = MinValidStackSize(options.stack_size());
105       GPR_ASSERT(pthread_attr_setstacksize(&attr, stack_size) == 0);
106     }
107 
108     *success = (pthread_create(
109                     &pthread_id_, &attr,
110                     [](void* v) -> void* {
111                       thd_arg arg = *static_cast<thd_arg*>(v);
112                       free(v);
113                       if (arg.name != nullptr) {
114 #if GPR_APPLE_PTHREAD_NAME
115                         /* Apple supports 64 characters, and will
116                          * truncate if it's longer. */
117                         pthread_setname_np(arg.name);
118 #elif GPR_LINUX_PTHREAD_NAME
119                         /* Linux supports 16 characters max, and will
120                          * error if it's longer. */
121                         char buf[16];
122                         size_t buf_len = GPR_ARRAY_SIZE(buf) - 1;
123                         strncpy(buf, arg.name, buf_len);
124                         buf[buf_len] = '\0';
125                         pthread_setname_np(pthread_self(), buf);
126 #endif  // GPR_APPLE_PTHREAD_NAME
127                       }
128 
129                       gpr_mu_lock(&arg.thread->mu_);
130                       while (!arg.thread->started_) {
131                         gpr_cv_wait(&arg.thread->ready_, &arg.thread->mu_,
132                                     gpr_inf_future(GPR_CLOCK_MONOTONIC));
133                       }
134                       gpr_mu_unlock(&arg.thread->mu_);
135 
136                       if (!arg.joinable) {
137                         delete arg.thread;
138                       }
139 
140                       (*arg.body)(arg.arg);
141                       if (arg.tracked) {
142                         Fork::DecThreadCount();
143                       }
144                       return nullptr;
145                     },
146                     info) == 0);
147 
148     GPR_ASSERT(pthread_attr_destroy(&attr) == 0);
149 
150     if (!(*success)) {
151       /* don't use gpr_free, as this was allocated using malloc (see above) */
152       free(info);
153       if (options.tracked()) {
154         Fork::DecThreadCount();
155       }
156     }
157   }
158 
~ThreadInternalsPosix()159   ~ThreadInternalsPosix() override {
160     gpr_mu_destroy(&mu_);
161     gpr_cv_destroy(&ready_);
162   }
163 
Start()164   void Start() override {
165     gpr_mu_lock(&mu_);
166     started_ = true;
167     gpr_cv_signal(&ready_);
168     gpr_mu_unlock(&mu_);
169   }
170 
Join()171   void Join() override { pthread_join(pthread_id_, nullptr); }
172 
173  private:
174   gpr_mu mu_;
175   gpr_cv ready_;
176   bool started_;
177   pthread_t pthread_id_;
178 };
179 
180 }  // namespace
181 
Thread(const char * thd_name,void (* thd_body)(void * arg),void * arg,bool * success,const Options & options)182 Thread::Thread(const char* thd_name, void (*thd_body)(void* arg), void* arg,
183                bool* success, const Options& options)
184     : options_(options) {
185   bool outcome = false;
186   impl_ = new ThreadInternalsPosix(thd_name, thd_body, arg, &outcome, options);
187   if (outcome) {
188     state_ = ALIVE;
189   } else {
190     state_ = FAILED;
191     delete impl_;
192     impl_ = nullptr;
193   }
194 
195   if (success != nullptr) {
196     *success = outcome;
197   }
198 }
199 }  // namespace grpc_core
200 
201 // The following is in the external namespace as it is exposed as C89 API
gpr_thd_currentid(void)202 gpr_thd_id gpr_thd_currentid(void) {
203   // Use C-style casting because Linux and OSX have different definitions
204   // of pthread_t so that a single C++ cast doesn't handle it.
205   // NOLINTNEXTLINE(google-readability-casting)
206   return (gpr_thd_id)pthread_self();
207 }
208 
209 #endif /* GPR_POSIX_SYNC */
210