folly/io/async/HHWheelTimer.cpp

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