src/core/lib/iomgr/exec_ctx.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 #include <grpc/support/port_platform.h>
  20
  21 #include "src/core/lib/iomgr/exec_ctx.h"
  22
  23 #include <grpc/support/log.h>
  24 #include <grpc/support/sync.h>
  25
  26 #include "src/core/lib/gprpp/thd.h"
  27 #include "src/core/lib/iomgr/combiner.h"
  28 #include "src/core/lib/profiling/timers.h"
  29
  30 static void exec_ctx_run(grpc_closure* closure, grpc_error* error) {
  31 #ifndef NDEBUG
  32   closure->scheduled = false;
  33   if (grpc_trace_closure.enabled()) {
  34     gpr_log(GPR_DEBUG, "running closure %p: created [%s:%d]: %s [%s:%d]",
  35             closure, closure->file_created, closure->line_created,
  36             closure->run ? "run" : "scheduled", closure->file_initiated,
  37             closure->line_initiated);
  38   }
  39 #endif
  40   closure->cb(closure->cb_arg, error);
  41 #ifndef NDEBUG
  42   if (grpc_trace_closure.enabled()) {
  43     gpr_log(GPR_DEBUG, "closure %p finished", closure);
  44   }
  45 #endif
  46   GRPC_ERROR_UNREF(error);
  47 }
  48
  49 static void exec_ctx_sched(grpc_closure* closure, grpc_error* error) {
  50   grpc_closure_list_append(grpc_core::ExecCtx::Get()->closure_list(), closure,
  51                            error);
  52 }
  53
  54 static gpr_timespec g_start_time;
  55 static gpr_cycle_counter g_start_cycle;
  56
  57 static grpc_millis timespan_to_millis_round_down(gpr_timespec ts) {
  58   double x = GPR_MS_PER_SEC * static_cast<double>(ts.tv_sec) +
  59              static_cast<double>(ts.tv_nsec) / GPR_NS_PER_MS;
  60   if (x < 0) return 0;
  61   if (x > GRPC_MILLIS_INF_FUTURE) return GRPC_MILLIS_INF_FUTURE;
  62   return static_cast<grpc_millis>(x);
  63 }
  64
  65 static grpc_millis timespec_to_millis_round_down(gpr_timespec ts) {
  66   return timespan_to_millis_round_down(gpr_time_sub(ts, g_start_time));
  67 }
  68
  69 static grpc_millis timespan_to_millis_round_up(gpr_timespec ts) {
  70   double x = GPR_MS_PER_SEC * static_cast<double>(ts.tv_sec) +
  71              static_cast<double>(ts.tv_nsec) / GPR_NS_PER_MS +
  72              static_cast<double>(GPR_NS_PER_SEC - 1) /
  73                  static_cast<double>(GPR_NS_PER_SEC);
  74   if (x < 0) return 0;
  75   if (x > GRPC_MILLIS_INF_FUTURE) return GRPC_MILLIS_INF_FUTURE;
  76   return static_cast<grpc_millis>(x);
  77 }
  78
  79 static grpc_millis timespec_to_millis_round_up(gpr_timespec ts) {
  80   return timespan_to_millis_round_up(gpr_time_sub(ts, g_start_time));
  81 }
  82
  83 gpr_timespec grpc_millis_to_timespec(grpc_millis millis,
  84                                      gpr_clock_type clock_type) {
  85   // special-case infinities as grpc_millis can be 32bit on some platforms
  86   // while gpr_time_from_millis always takes an int64_t.
  87   if (millis == GRPC_MILLIS_INF_FUTURE) {
  88     return gpr_inf_future(clock_type);
  89   }
  90   if (millis == GRPC_MILLIS_INF_PAST) {
  91     return gpr_inf_past(clock_type);
  92   }
  93
  94   if (clock_type == GPR_TIMESPAN) {
  95     return gpr_time_from_millis(millis, GPR_TIMESPAN);
  96   }
  97   return gpr_time_add(gpr_convert_clock_type(g_start_time, clock_type),
  98                       gpr_time_from_millis(millis, GPR_TIMESPAN));
  99 }
 100
 101 grpc_millis grpc_timespec_to_millis_round_down(gpr_timespec ts) {
 102   return timespec_to_millis_round_down(
 103       gpr_convert_clock_type(ts, g_start_time.clock_type));
 104 }
 105
 106 grpc_millis grpc_timespec_to_millis_round_up(gpr_timespec ts) {
 107   return timespec_to_millis_round_up(
 108       gpr_convert_clock_type(ts, g_start_time.clock_type));
 109 }
 110
 111 grpc_millis grpc_cycle_counter_to_millis_round_down(gpr_cycle_counter cycles) {
 112   return timespan_to_millis_round_down(
 113       gpr_cycle_counter_sub(cycles, g_start_cycle));
 114 }
 115
 116 grpc_millis grpc_cycle_counter_to_millis_round_up(gpr_cycle_counter cycles) {
 117   return timespan_to_millis_round_up(
 118       gpr_cycle_counter_sub(cycles, g_start_cycle));
 119 }
 120
 121 namespace grpc_core {
 122 GPR_TLS_CLASS_DEF(ExecCtx::exec_ctx_);
 123 GPR_TLS_CLASS_DEF(ApplicationCallbackExecCtx::callback_exec_ctx_);
 124
 125 // WARNING: for testing purposes only!
 126 void ExecCtx::TestOnlyGlobalInit(gpr_timespec new_val) {
 127   g_start_time = new_val;
 128   gpr_tls_init(&exec_ctx_);
 129 }
 130
 131 void ExecCtx::GlobalInit(void) {
 132   // gpr_now(GPR_CLOCK_MONOTONIC) incurs a syscall. We don't actually know the
 133   // exact cycle the time was captured, so we use the average of cycles before
 134   // and after the syscall as the starting cycle.
 135   const gpr_cycle_counter cycle_before = gpr_get_cycle_counter();
 136   g_start_time = gpr_now(GPR_CLOCK_MONOTONIC);
 137   const gpr_cycle_counter cycle_after = gpr_get_cycle_counter();
 138   g_start_cycle = (cycle_before + cycle_after) / 2;
 139   gpr_tls_init(&exec_ctx_);
 140 }
 141
 142 bool ExecCtx::Flush() {
 143   bool did_something = 0;
 144   GPR_TIMER_SCOPE("grpc_exec_ctx_flush", 0);
 145   for (;;) {
 146     if (!grpc_closure_list_empty(closure_list_)) {
 147       grpc_closure* c = closure_list_.head;
 148       closure_list_.head = closure_list_.tail = nullptr;
 149       while (c != nullptr) {
 150         grpc_closure* next = c->next_data.next;
 151         grpc_error* error = c->error_data.error;
 152         did_something = true;
 153         exec_ctx_run(c, error);
 154         c = next;
 155       }
 156     } else if (!grpc_combiner_continue_exec_ctx()) {
 157       break;
 158     }
 159   }
 160   GPR_ASSERT(combiner_data_.active_combiner == nullptr);
 161   return did_something;
 162 }
 163
 164 grpc_millis ExecCtx::Now() {
 165   if (!now_is_valid_) {
 166     now_ = timespec_to_millis_round_down(gpr_now(GPR_CLOCK_MONOTONIC));
 167     now_is_valid_ = true;
 168   }
 169   return now_;
 170 }
 171
 172 void ExecCtx::Run(const DebugLocation& location, grpc_closure* closure,
 173                   grpc_error* error) {
 174   (void)location;
 175   if (closure == nullptr) {
 176     GRPC_ERROR_UNREF(error);
 177     return;
 178   }
 179 #ifndef NDEBUG
 180   if (closure->scheduled) {
 181     gpr_log(GPR_ERROR,
 182             "Closure already scheduled. (closure: %p, created: [%s:%d], "
 183             "previously scheduled at: [%s: %d], newly scheduled at [%s: %d]",
 184             closure, closure->file_created, closure->line_created,
 185             closure->file_initiated, closure->line_initiated, location.file(),
 186             location.line());
 187     abort();
 188   }
 189   closure->scheduled = true;
 190   closure->file_initiated = location.file();
 191   closure->line_initiated = location.line();
 192   closure->run = false;
 193   GPR_ASSERT(closure->cb != nullptr);
 194 #endif
 195   exec_ctx_sched(closure, error);
 196 }
 197
 198 void ExecCtx::RunList(const DebugLocation& location, grpc_closure_list* list) {
 199   (void)location;
 200   grpc_closure* c = list->head;
 201   while (c != nullptr) {
 202     grpc_closure* next = c->next_data.next;
 203 #ifndef NDEBUG
 204     if (c->scheduled) {
 205       gpr_log(GPR_ERROR,
 206               "Closure already scheduled. (closure: %p, created: [%s:%d], "
 207               "previously scheduled at: [%s: %d], newly scheduled at [%s:%d]",
 208               c, c->file_created, c->line_created, c->file_initiated,
 209               c->line_initiated, location.file(), location.line());
 210       abort();
 211     }
 212     c->scheduled = true;
 213     c->file_initiated = location.file();
 214     c->line_initiated = location.line();
 215     c->run = false;
 216     GPR_ASSERT(c->cb != nullptr);
 217 #endif
 218     exec_ctx_sched(c, c->error_data.error);
 219     c = next;
 220   }
 221   list->head = list->tail = nullptr;
 222 }
 223
 224 }  // namespace grpc_core