src/core/lib/gprpp/fork.cc

   1 /*
   2  *
   3  * Copyright 2017 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 #include <grpc/support/port_platform.h>
  20
  21 #include "src/core/lib/gprpp/fork.h"
  22
  23 #include <string.h>
  24
  25 #include <grpc/support/alloc.h>
  26 #include <grpc/support/sync.h>
  27 #include <grpc/support/time.h>
  28
  29 #include "src/core/lib/gpr/env.h"
  30 #include "src/core/lib/gpr/useful.h"
  31 #include "src/core/lib/gprpp/memory.h"
  32
  33 /*
  34  * NOTE: FORKING IS NOT GENERALLY SUPPORTED, THIS IS ONLY INTENDED TO WORK
  35  *       AROUND VERY SPECIFIC USE CASES.
  36  */
  37
  38 namespace grpc_core {
  39 namespace internal {
  40 // The exec_ctx_count has 2 modes, blocked and unblocked.
  41 // When unblocked, the count is 2-indexed; exec_ctx_count=2 indicates
  42 // 0 active ExecCtxs, exex_ctx_count=3 indicates 1 active ExecCtxs...
  43
  44 // When blocked, the exec_ctx_count is 0-indexed.  Note that ExecCtx
  45 // creation can only be blocked if there is exactly 1 outstanding ExecCtx,
  46 // meaning that BLOCKED and UNBLOCKED counts partition the integers
  47 #define UNBLOCKED(n) (n + 2)
  48 #define BLOCKED(n) (n)
  49
  50 class ExecCtxState {
  51  public:
  52   ExecCtxState() : fork_complete_(true) {
  53     gpr_mu_init(&mu_);
  54     gpr_cv_init(&cv_);
  55     gpr_atm_no_barrier_store(&count_, UNBLOCKED(0));
  56   }
  57
  58   void IncExecCtxCount() {
  59     gpr_atm count = gpr_atm_no_barrier_load(&count_);
  60     while (true) {
  61       if (count <= BLOCKED(1)) {
  62         // This only occurs if we are trying to fork.  Wait until the fork()
  63         // operation completes before allowing new ExecCtxs.
  64         gpr_mu_lock(&mu_);
  65         if (gpr_atm_no_barrier_load(&count_) <= BLOCKED(1)) {
  66           while (!fork_complete_) {
  67             gpr_cv_wait(&cv_, &mu_, gpr_inf_future(GPR_CLOCK_REALTIME));
  68           }
  69         }
  70         gpr_mu_unlock(&mu_);
  71       } else if (gpr_atm_no_barrier_cas(&count_, count, count + 1)) {
  72         break;
  73       }
  74       count = gpr_atm_no_barrier_load(&count_);
  75     }
  76   }
  77
  78   void DecExecCtxCount() { gpr_atm_no_barrier_fetch_add(&count_, -1); }
  79
  80   bool BlockExecCtx() {
  81     // Assumes there is an active ExecCtx when this function is called
  82     if (gpr_atm_no_barrier_cas(&count_, UNBLOCKED(1), BLOCKED(1))) {
  83       gpr_mu_lock(&mu_);
  84       fork_complete_ = false;
  85       gpr_mu_unlock(&mu_);
  86       return true;
  87     }
  88     return false;
  89   }
  90
  91   void AllowExecCtx() {
  92     gpr_mu_lock(&mu_);
  93     gpr_atm_no_barrier_store(&count_, UNBLOCKED(0));
  94     fork_complete_ = true;
  95     gpr_cv_broadcast(&cv_);
  96     gpr_mu_unlock(&mu_);
  97   }
  98
  99   ~ExecCtxState() {
 100     gpr_mu_destroy(&mu_);
 101     gpr_cv_destroy(&cv_);
 102   }
 103
 104  private:
 105   bool fork_complete_;
 106   gpr_mu mu_;
 107   gpr_cv cv_;
 108   gpr_atm count_;
 109 };
 110
 111 class ThreadState {
 112  public:
 113   ThreadState() : awaiting_threads_(false), threads_done_(false), count_(0) {
 114     gpr_mu_init(&mu_);
 115     gpr_cv_init(&cv_);
 116   }
 117
 118   void IncThreadCount() {
 119     gpr_mu_lock(&mu_);
 120     count_++;
 121     gpr_mu_unlock(&mu_);
 122   }
 123
 124   void DecThreadCount() {
 125     gpr_mu_lock(&mu_);
 126     count_--;
 127     if (awaiting_threads_ && count_ == 0) {
 128       threads_done_ = true;
 129       gpr_cv_signal(&cv_);
 130     }
 131     gpr_mu_unlock(&mu_);
 132   }
 133   void AwaitThreads() {
 134     gpr_mu_lock(&mu_);
 135     awaiting_threads_ = true;
 136     threads_done_ = (count_ == 0);
 137     while (!threads_done_) {
 138       gpr_cv_wait(&cv_, &mu_, gpr_inf_future(GPR_CLOCK_REALTIME));
 139     }
 140     awaiting_threads_ = true;
 141     gpr_mu_unlock(&mu_);
 142   }
 143
 144   ~ThreadState() {
 145     gpr_mu_destroy(&mu_);
 146     gpr_cv_destroy(&cv_);
 147   }
 148
 149  private:
 150   bool awaiting_threads_;
 151   bool threads_done_;
 152   gpr_mu mu_;
 153   gpr_cv cv_;
 154   int count_;
 155 };
 156
 157 }  // namespace
 158
 159 void Fork::GlobalInit() {
 160   if (!override_enabled_) {
 161 #ifdef GRPC_ENABLE_FORK_SUPPORT
 162     support_enabled_ = true;
 163 #endif
 164     bool env_var_set = false;
 165     char* env = gpr_getenv("GRPC_ENABLE_FORK_SUPPORT");
 166     if (env != nullptr) {
 167       static const char* truthy[] = {"yes",  "Yes",  "YES", "true",
 168                                      "True", "TRUE", "1"};
 169       static const char* falsey[] = {"no",    "No",    "NO", "false",
 170                                      "False", "FALSE", "0"};
 171       for (size_t i = 0; i < GPR_ARRAY_SIZE(truthy); i++) {
 172         if (0 == strcmp(env, truthy[i])) {
 173           support_enabled_ = true;
 174           env_var_set = true;
 175           break;
 176         }
 177       }
 178       if (!env_var_set) {
 179         for (size_t i = 0; i < GPR_ARRAY_SIZE(falsey); i++) {
 180           if (0 == strcmp(env, falsey[i])) {
 181             support_enabled_ = false;
 182             env_var_set = true;
 183             break;
 184           }
 185         }
 186       }
 187       gpr_free(env);
 188     }
 189   }
 190   if (support_enabled_) {
 191     exec_ctx_state_ = grpc_core::New<internal::ExecCtxState>();
 192     thread_state_ = grpc_core::New<internal::ThreadState>();
 193   }
 194 }
 195
 196 void Fork::GlobalShutdown() {
 197   if (support_enabled_) {
 198     grpc_core::Delete(exec_ctx_state_);
 199     grpc_core::Delete(thread_state_);
 200   }
 201 }
 202
 203 bool Fork::Enabled() { return support_enabled_; }
 204
 205 // Testing Only
 206 void Fork::Enable(bool enable) {
 207   override_enabled_ = true;
 208   support_enabled_ = enable;
 209 }
 210
 211 void Fork::IncExecCtxCount() {
 212   if (support_enabled_) {
 213     exec_ctx_state_->IncExecCtxCount();
 214   }
 215 }
 216
 217 void Fork::DecExecCtxCount() {
 218   if (support_enabled_) {
 219     exec_ctx_state_->DecExecCtxCount();
 220   }
 221 }
 222
 223 void Fork::SetResetChildPollingEngineFunc(
 224     Fork::child_postfork_func reset_child_polling_engine) {
 225   reset_child_polling_engine_ = reset_child_polling_engine;
 226 }
 227 Fork::child_postfork_func Fork::GetResetChildPollingEngineFunc() {
 228   return reset_child_polling_engine_;
 229 }
 230
 231 bool Fork::BlockExecCtx() {
 232   if (support_enabled_) {
 233     return exec_ctx_state_->BlockExecCtx();
 234   }
 235   return false;
 236 }
 237
 238 void Fork::AllowExecCtx() {
 239   if (support_enabled_) {
 240     exec_ctx_state_->AllowExecCtx();
 241   }
 242 }
 243
 244 void Fork::IncThreadCount() {
 245   if (support_enabled_) {
 246     thread_state_->IncThreadCount();
 247   }
 248 }
 249
 250 void Fork::DecThreadCount() {
 251   if (support_enabled_) {
 252     thread_state_->DecThreadCount();
 253   }
 254 }
 255 void Fork::AwaitThreads() {
 256   if (support_enabled_) {
 257     thread_state_->AwaitThreads();
 258   }
 259 }
 260
 261 internal::ExecCtxState* Fork::exec_ctx_state_ = nullptr;
 262 internal::ThreadState* Fork::thread_state_ = nullptr;
 263 bool Fork::support_enabled_ = false;
 264 bool Fork::override_enabled_ = false;
 265 Fork::child_postfork_func Fork::reset_child_polling_engine_ = nullptr;
 266 }  // namespace grpc_core