legacy-libs/grpc-cloned/deps/grpc/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 "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
  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   if (request_size < _SC_THREAD_STACK_MIN) {
  63     request_size = _SC_THREAD_STACK_MIN;
  64   }
  65
  66   // On some systems, pthread_attr_setstacksize() can fail if stacksize is
  67   // not a multiple of the system page size.
  68   return RoundUpToPageSize(request_size);
  69 }
  70
  71 class ThreadInternalsPosix : public internal::ThreadInternalsInterface {
  72  public:
  73   ThreadInternalsPosix(const char* thd_name, void (*thd_body)(void* arg),
  74                        void* arg, bool* success, const Thread::Options& options)
  75       : started_(false) {
  76     gpr_mu_init(&mu_);
  77     gpr_cv_init(&ready_);
  78     pthread_attr_t attr;
  79     /* don't use gpr_malloc as we may cause an infinite recursion with
  80      * the profiling code */
  81     thd_arg* info = static_cast<thd_arg*>(malloc(sizeof(*info)));
  82     GPR_ASSERT(info != nullptr);
  83     info->thread = this;
  84     info->body = thd_body;
  85     info->arg = arg;
  86     info->name = thd_name;
  87     info->joinable = options.joinable();
  88     info->tracked = options.tracked();
  89     if (options.tracked()) {
  90       Fork::IncThreadCount();
  91     }
  92
  93     GPR_ASSERT(pthread_attr_init(&attr) == 0);
  94     if (options.joinable()) {
  95       GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE) ==
  96                  0);
  97     } else {
  98       GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) ==
  99                  0);
 100     }
 101
 102     if (options.stack_size() != 0) {
 103       size_t stack_size = MinValidStackSize(options.stack_size());
 104       GPR_ASSERT(pthread_attr_setstacksize(&attr, stack_size) == 0);
 105     }
 106
 107     *success =
 108         (pthread_create(&pthread_id_, &attr,
 109                         [](void* v) -> void* {
 110                           thd_arg arg = *static_cast<thd_arg*>(v);
 111                           free(v);
 112                           if (arg.name != nullptr) {
 113 #if GPR_APPLE_PTHREAD_NAME
 114                             /* Apple supports 64 characters, and will
 115                              * truncate if it's longer. */
 116                             pthread_setname_np(arg.name);
 117 #elif GPR_LINUX_PTHREAD_NAME
 118                             /* Linux supports 16 characters max, and will
 119                              * error if it's longer. */
 120                             char buf[16];
 121                             size_t buf_len = GPR_ARRAY_SIZE(buf) - 1;
 122                             strncpy(buf, arg.name, buf_len);
 123                             buf[buf_len] = '\0';
 124                             pthread_setname_np(pthread_self(), buf);
 125 #endif  // GPR_APPLE_PTHREAD_NAME
 126                           }
 127
 128                           gpr_mu_lock(&arg.thread->mu_);
 129                           while (!arg.thread->started_) {
 130                             gpr_cv_wait(&arg.thread->ready_, &arg.thread->mu_,
 131                                         gpr_inf_future(GPR_CLOCK_MONOTONIC));
 132                           }
 133                           gpr_mu_unlock(&arg.thread->mu_);
 134
 135                           if (!arg.joinable) {
 136                             Delete(arg.thread);
 137                           }
 138
 139                           (*arg.body)(arg.arg);
 140                           if (arg.tracked) {
 141                             Fork::DecThreadCount();
 142                           }
 143                           return nullptr;
 144                         },
 145                         info) == 0);
 146
 147     GPR_ASSERT(pthread_attr_destroy(&attr) == 0);
 148
 149     if (!(*success)) {
 150       /* don't use gpr_free, as this was allocated using malloc (see above) */
 151       free(info);
 152       if (options.tracked()) {
 153         Fork::DecThreadCount();
 154       }
 155     }
 156   }
 157
 158   ~ThreadInternalsPosix() override {
 159     gpr_mu_destroy(&mu_);
 160     gpr_cv_destroy(&ready_);
 161   }
 162
 163   void Start() override {
 164     gpr_mu_lock(&mu_);
 165     started_ = true;
 166     gpr_cv_signal(&ready_);
 167     gpr_mu_unlock(&mu_);
 168   }
 169
 170   void Join() override { pthread_join(pthread_id_, nullptr); }
 171
 172  private:
 173   gpr_mu mu_;
 174   gpr_cv ready_;
 175   bool started_;
 176   pthread_t pthread_id_;
 177 };
 178
 179 }  // namespace
 180
 181 Thread::Thread(const char* thd_name, void (*thd_body)(void* arg), void* arg,
 182                bool* success, const Options& options)
 183     : options_(options) {
 184   bool outcome = false;
 185   impl_ = New<ThreadInternalsPosix>(thd_name, thd_body, arg, &outcome, options);
 186   if (outcome) {
 187     state_ = ALIVE;
 188   } else {
 189     state_ = FAILED;
 190     Delete(impl_);
 191     impl_ = nullptr;
 192   }
 193
 194   if (success != nullptr) {
 195     *success = outcome;
 196   }
 197 }
 198 }  // namespace grpc_core
 199
 200 // The following is in the external namespace as it is exposed as C89 API
 201 gpr_thd_id gpr_thd_currentid(void) { return (gpr_thd_id)pthread_self(); }
 202
 203 #endif /* GPR_POSIX_SYNC */