folly/io/async/HHWheelTimer.cpp

   1 /*
   2  * Copyright 2015 Facebook, Inc.
   3  *
   4  * Licensed to the Apache Software Foundation (ASF) under one
   5  * or more contributor license agreements. See the NOTICE file
   6  * distributed with this work for additional information
   7  * regarding copyright ownership. The ASF licenses this file
   8  * to you under the Apache License, Version 2.0 (the
   9  * "License"); you may not use this file except in compliance
  10  * with the License. You may obtain a copy of the License at
  11  *
  12  *   http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing,
  15  * software distributed under the License is distributed on an
  16  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  17  * KIND, either express or implied. See the License for the
  18  * specific language governing permissions and limitations
  19  * under the License.
  20  */
  21 #include <folly/io/async/HHWheelTimer.h>
  22 #include <folly/io/async/Request.h>
  23
  24 #include <folly/ScopeGuard.h>
  25
  26 #include <cassert>
  27
  28 using std::chrono::milliseconds;
  29
  30 namespace folly {
  31
  32 /**
  33  * We want to select the default interval carefully.
  34  * An interval of 10ms will give us 10ms * WHEEL_SIZE^WHEEL_BUCKETS
  35  * for the largest timeout possible, or about 497 days.
  36  *
  37  * For a lower bound, we want a reasonable limit on local IO, 10ms
  38  * seems short enough
  39  *
  40  * A shorter interval also has CPU implications, less than 1ms might
  41  * start showing up in cpu perf.  Also, it might not be possible to set
  42  * tick interval less than 10ms on older kernels.
  43  */
  44 int HHWheelTimer::DEFAULT_TICK_INTERVAL = 10;
  45
  46 HHWheelTimer::Callback::~Callback() {
  47   if (isScheduled()) {
  48     cancelTimeout();
  49   }
  50 }
  51
  52 void HHWheelTimer::Callback::setScheduled(HHWheelTimer* wheel,
  53                                           std::chrono::milliseconds timeout) {
  54   assert(wheel_ == nullptr);
  55   assert(expiration_ == milliseconds(0));
  56
  57   wheel_ = wheel;
  58
  59   // Only update the now_ time if we're not in a timeout expired callback
  60   if (wheel_->count_  == 0 && !wheel_->processingCallbacksGuard_) {
  61     wheel_->now_ = getCurTime();
  62   }
  63
  64   expiration_ = wheel_->now_ + timeout;
  65 }
  66
  67 void HHWheelTimer::Callback::cancelTimeoutImpl() {
  68   if (--wheel_->count_ <= 0) {
  69     assert(wheel_->count_ == 0);
  70     wheel_->AsyncTimeout::cancelTimeout();
  71   }
  72   hook_.unlink();
  73
  74   wheel_ = nullptr;
  75   expiration_ = milliseconds(0);
  76 }
  77
  78 HHWheelTimer::HHWheelTimer(folly::EventBase* eventBase,
  79                            std::chrono::milliseconds intervalMS,
  80                            AsyncTimeout::InternalEnum internal)
  81   : AsyncTimeout(eventBase, internal)
  82   , interval_(intervalMS)
  83   , nextTick_(1)
  84   , count_(0)
  85   , catchupEveryN_(DEFAULT_CATCHUP_EVERY_N)
  86   , expirationsSinceCatchup_(0)
  87   , processingCallbacksGuard_(false)
  88 {
  89 }
  90
  91 HHWheelTimer::~HHWheelTimer() {
  92 }
  93
  94 void HHWheelTimer::destroy() {
  95   assert(count_ == 0);
  96   DelayedDestruction::destroy();
  97 }
  98
  99 void HHWheelTimer::scheduleTimeoutImpl(Callback* callback,
 100                                        std::chrono::milliseconds timeout) {
 101   int64_t due = timeToWheelTicks(timeout) + nextTick_;
 102   int64_t diff = due - nextTick_;
 103   CallbackList* list;
 104
 105   if (diff < 0) {
 106     list = &buckets_[0][nextTick_ & WHEEL_MASK];
 107   } else if (diff < WHEEL_SIZE) {
 108     list = &buckets_[0][due & WHEEL_MASK];
 109   } else if (diff < 1 << (2 * WHEEL_BITS)) {
 110     list = &buckets_[1][(due >> WHEEL_BITS) & WHEEL_MASK];
 111   } else if (diff < 1 << (3 * WHEEL_BITS)) {
 112     list = &buckets_[2][(due >> 2 * WHEEL_BITS) & WHEEL_MASK];
 113   } else {
 114     /* in largest slot */
 115     if (diff > LARGEST_SLOT) {
 116       diff = LARGEST_SLOT;
 117       due = diff + nextTick_;
 118     }
 119     list = &buckets_[3][(due >> 3 * WHEEL_BITS) & WHEEL_MASK];
 120   }
 121   list->push_back(*callback);
 122 }
 123
 124 void HHWheelTimer::scheduleTimeout(Callback* callback,
 125                                    std::chrono::milliseconds timeout) {
 126   // Cancel the callback if it happens to be scheduled already.
 127   callback->cancelTimeout();
 128
 129   callback->context_ = RequestContext::saveContext();
 130
 131   if (count_ == 0 && !processingCallbacksGuard_) {
 132     this->AsyncTimeout::scheduleTimeout(interval_.count());
 133   }
 134
 135   callback->setScheduled(this, timeout);
 136   scheduleTimeoutImpl(callback, timeout);
 137   count_++;
 138 }
 139
 140 bool HHWheelTimer::cascadeTimers(int bucket, int tick) {
 141   CallbackList cbs;
 142   cbs.swap(buckets_[bucket][tick]);
 143   while (!cbs.empty()) {
 144     auto* cb = &cbs.front();
 145     cbs.pop_front();
 146     scheduleTimeoutImpl(cb, cb->getTimeRemaining(now_));
 147   }
 148
 149   // If tick is zero, timeoutExpired will cascade the next bucket.
 150   return tick == 0;
 151 }
 152
 153 void HHWheelTimer::timeoutExpired() noexcept {
 154   // If destroy() is called inside timeoutExpired(), delay actual destruction
 155   // until timeoutExpired() returns
 156   DestructorGuard dg(this);
 157   // If scheduleTimeout is called from a callback in this function, it may
 158   // cause inconsistencies in the state of this object. As such, we need
 159   // to treat these calls slightly differently.
 160   processingCallbacksGuard_ = true;
 161   auto reEntryGuard = folly::makeGuard([&] {
 162     processingCallbacksGuard_ = false;
 163   });
 164
 165   // timeoutExpired() can only be invoked directly from the event base loop.
 166   // It should never be invoked recursively.
 167   //
 168   milliseconds catchup = now_ + interval_;
 169   // If catchup is enabled, we may have missed multiple intervals, use
 170   // currentTime() to check exactly.
 171   if (++expirationsSinceCatchup_ >= catchupEveryN_) {
 172     catchup = std::chrono::duration_cast<milliseconds>(
 173       std::chrono::steady_clock::now().time_since_epoch());
 174     expirationsSinceCatchup_ = 0;
 175   }
 176   while (now_ < catchup) {
 177     now_ += interval_;
 178
 179     int idx = nextTick_ & WHEEL_MASK;
 180     if (0 == idx) {
 181       // Cascade timers
 182       if (cascadeTimers(1, (nextTick_ >> WHEEL_BITS) & WHEEL_MASK) &&
 183           cascadeTimers(2, (nextTick_ >> (2 * WHEEL_BITS)) & WHEEL_MASK)) {
 184         cascadeTimers(3, (nextTick_ >> (3 * WHEEL_BITS)) & WHEEL_MASK);
 185       }
 186     }
 187
 188     nextTick_++;
 189     CallbackList* cbs = &buckets_[0][idx];
 190     while (!cbs->empty()) {
 191       auto* cb = &cbs->front();
 192       cbs->pop_front();
 193       count_--;
 194       cb->wheel_ = nullptr;
 195       cb->expiration_ = milliseconds(0);
 196       auto old_ctx =
 197         RequestContext::setContext(cb->context_);
 198       cb->timeoutExpired();
 199       RequestContext::setContext(old_ctx);
 200     }
 201   }
 202   if (count_ > 0) {
 203     this->AsyncTimeout::scheduleTimeout(interval_.count());
 204   }
 205 }
 206
 207 size_t HHWheelTimer::cancelAll() {
 208   decltype(buckets_) buckets;
 209
 210 // Work around std::swap() bug in libc++
 211 //
 212 // http://llvm.org/bugs/show_bug.cgi?id=22106
 213 #if FOLLY_USE_LIBCPP
 214   for (size_t i = 0; i < WHEEL_BUCKETS; ++i) {
 215     for (size_t ii = 0; ii < WHEEL_SIZE; ++ii) {
 216       std::swap(buckets_[i][ii], buckets[i][ii]);
 217     }
 218   }
 219 #else
 220   std::swap(buckets, buckets_);
 221 #endif
 222
 223   size_t count = 0;
 224
 225   for (auto& tick : buckets) {
 226     for (auto& bucket : tick) {
 227       while (!bucket.empty()) {
 228         auto& cb = bucket.front();
 229         cb.cancelTimeout();
 230         cb.callbackCanceled();
 231         count++;
 232       }
 233     }
 234   }
 235
 236   return count;
 237 }
 238
 239 } // folly