src/core/lib/gprpp/thd_posix.cc

   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 <pthread.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <unistd.h>
  29
  30 #include <grpc/support/alloc.h>
  31 #include <grpc/support/log.h>
  32 #include <grpc/support/sync.h>
  33 #include <grpc/support/thd_id.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 #include "src/core/lib/gprpp/thd.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
  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.
  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:
  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
 159   ~ThreadInternalsPosix() override {
 160     gpr_mu_destroy(&mu_);
 161     gpr_cv_destroy(&ready_);
 162   }
 163
 164   void Start() override {
 165     gpr_mu_lock(&mu_);
 166     started_ = true;
 167     gpr_cv_signal(&ready_);
 168     gpr_mu_unlock(&mu_);
 169   }
 170
 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
 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
 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 */