folly/io/async/HHWheelTimer.cpp

   1 /*
   2  * Copyright 2016 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/Optional.h>
  25 #include <folly/ScopeGuard.h>
  26
  27 #include <cassert>
  28
  29 using std::chrono::milliseconds;
  30
  31 namespace folly {
  32
  33 /**
  34  * We want to select the default interval carefully.
  35  * An interval of 10ms will give us 10ms * WHEEL_SIZE^WHEEL_BUCKETS
  36  * for the largest timeout possible, or about 497 days.
  37  *
  38  * For a lower bound, we want a reasonable limit on local IO, 10ms
  39  * seems short enough
  40  *
  41  * A shorter interval also has CPU implications, less than 1ms might
  42  * start showing up in cpu perf.  Also, it might not be possible to set
  43  * tick interval less than 10ms on older kernels.
  44  */
  45 int HHWheelTimer::DEFAULT_TICK_INTERVAL = 10;
  46
  47 HHWheelTimer::Callback::~Callback() {
  48   if (isScheduled()) {
  49     cancelTimeout();
  50   }
  51 }
  52
  53 void HHWheelTimer::Callback::setScheduled(HHWheelTimer* wheel,
  54                                           std::chrono::milliseconds timeout) {
  55   assert(wheel_ == nullptr);
  56   assert(expiration_ == milliseconds(0));
  57
  58   wheelGuard_ = DestructorGuard(wheel);
  59   wheel_ = wheel;
  60
  61   // Only update the now_ time if we're not in a timeout expired callback
  62   if (wheel_->count_  == 0 && !wheel_->processingCallbacksGuard_) {
  63     wheel_->now_ = getCurTime();
  64   }
  65
  66   expiration_ = wheel_->now_ + timeout;
  67 }
  68
  69 void HHWheelTimer::Callback::cancelTimeoutImpl() {
  70   if (--wheel_->count_ <= 0) {
  71     assert(wheel_->count_ == 0);
  72     wheel_->AsyncTimeout::cancelTimeout();
  73   }
  74   hook_.unlink();
  75
  76   wheel_ = nullptr;
  77   wheelGuard_ = folly::none;
  78   expiration_ = milliseconds(0);
  79 }
  80
  81 HHWheelTimer::HHWheelTimer(folly::TimeoutManager* timeoutMananger,
  82                            std::chrono::milliseconds intervalMS,
  83                            AsyncTimeout::InternalEnum internal,
  84                            std::chrono::milliseconds defaultTimeoutMS)
  85     : AsyncTimeout(timeoutMananger, internal),
  86       interval_(intervalMS),
  87       defaultTimeout_(defaultTimeoutMS),
  88       nextTick_(1),
  89       count_(0),
  90       catchupEveryN_(DEFAULT_CATCHUP_EVERY_N),
  91       expirationsSinceCatchup_(0),
  92       processingCallbacksGuard_(false) {}
  93
  94 HHWheelTimer::~HHWheelTimer() {
  95   CHECK(count_ == 0);
  96 }
  97
  98 void HHWheelTimer::destroy() {
  99   if (getDestructorGuardCount() == count_) {
 100     // Every callback holds a DestructorGuard.  In this simple case,
 101     // All timeouts should already be gone.
 102     assert(count_ == 0);
 103
 104     // Clean them up in opt builds and move on
 105     cancelAll();
 106   }
 107   // else, we are only marking pending destruction, but one or more
 108   // HHWheelTimer::SharedPtr's (and possibly their timeouts) are still holding
 109   // this HHWheelTimer.  We cannot assert that all timeouts have been cancelled,
 110   // and will just have to wait for them all to complete on their own.
 111   DelayedDestruction::destroy();
 112 }
 113
 114 void HHWheelTimer::scheduleTimeoutImpl(Callback* callback,
 115                                        std::chrono::milliseconds timeout) {
 116   int64_t due = timeToWheelTicks(timeout) + nextTick_;
 117   int64_t diff = due - nextTick_;
 118   CallbackList* list;
 119
 120   if (diff < 0) {
 121     list = &buckets_[0][nextTick_ & WHEEL_MASK];
 122   } else if (diff < WHEEL_SIZE) {
 123     list = &buckets_[0][due & WHEEL_MASK];
 124   } else if (diff < 1 << (2 * WHEEL_BITS)) {
 125     list = &buckets_[1][(due >> WHEEL_BITS) & WHEEL_MASK];
 126   } else if (diff < 1 << (3 * WHEEL_BITS)) {
 127     list = &buckets_[2][(due >> 2 * WHEEL_BITS) & WHEEL_MASK];
 128   } else {
 129     /* in largest slot */
 130     if (diff > LARGEST_SLOT) {
 131       diff = LARGEST_SLOT;
 132       due = diff + nextTick_;
 133     }
 134     list = &buckets_[3][(due >> 3 * WHEEL_BITS) & WHEEL_MASK];
 135   }
 136   list->push_back(*callback);
 137 }
 138
 139 void HHWheelTimer::scheduleTimeout(Callback* callback,
 140                                    std::chrono::milliseconds timeout) {
 141   // Cancel the callback if it happens to be scheduled already.
 142   callback->cancelTimeout();
 143
 144   callback->context_ = RequestContext::saveContext();
 145
 146   if (count_ == 0 && !processingCallbacksGuard_) {
 147     this->AsyncTimeout::scheduleTimeout(interval_.count());
 148   }
 149
 150   callback->setScheduled(this, timeout);
 151   scheduleTimeoutImpl(callback, timeout);
 152   count_++;
 153 }
 154
 155 void HHWheelTimer::scheduleTimeout(Callback* callback) {
 156   CHECK(std::chrono::milliseconds(-1) != defaultTimeout_)
 157       << "Default timeout was not initialized";
 158   scheduleTimeout(callback, defaultTimeout_);
 159 }
 160
 161 bool HHWheelTimer::cascadeTimers(int bucket, int tick) {
 162   CallbackList cbs;
 163   cbs.swap(buckets_[bucket][tick]);
 164   while (!cbs.empty()) {
 165     auto* cb = &cbs.front();
 166     cbs.pop_front();
 167     scheduleTimeoutImpl(cb, cb->getTimeRemaining(now_));
 168   }
 169
 170   // If tick is zero, timeoutExpired will cascade the next bucket.
 171   return tick == 0;
 172 }
 173
 174 void HHWheelTimer::timeoutExpired() noexcept {
 175   // If destroy() is called inside timeoutExpired(), delay actual destruction
 176   // until timeoutExpired() returns
 177   DestructorGuard dg(this);
 178   // If scheduleTimeout is called from a callback in this function, it may
 179   // cause inconsistencies in the state of this object. As such, we need
 180   // to treat these calls slightly differently.
 181   processingCallbacksGuard_ = true;
 182   auto reEntryGuard = folly::makeGuard([&] {
 183     processingCallbacksGuard_ = false;
 184   });
 185
 186   // timeoutExpired() can only be invoked directly from the event base loop.
 187   // It should never be invoked recursively.
 188   //
 189   milliseconds catchup = now_ + interval_;
 190   // If catchup is enabled, we may have missed multiple intervals, use
 191   // currentTime() to check exactly.
 192   if (++expirationsSinceCatchup_ >= catchupEveryN_) {
 193     catchup = std::chrono::duration_cast<milliseconds>(
 194       std::chrono::steady_clock::now().time_since_epoch());
 195     expirationsSinceCatchup_ = 0;
 196   }
 197   while (now_ < catchup) {
 198     now_ += interval_;
 199
 200     int idx = nextTick_ & WHEEL_MASK;
 201     if (0 == idx) {
 202       // Cascade timers
 203       if (cascadeTimers(1, (nextTick_ >> WHEEL_BITS) & WHEEL_MASK) &&
 204           cascadeTimers(2, (nextTick_ >> (2 * WHEEL_BITS)) & WHEEL_MASK)) {
 205         cascadeTimers(3, (nextTick_ >> (3 * WHEEL_BITS)) & WHEEL_MASK);
 206       }
 207     }
 208
 209     nextTick_++;
 210     CallbackList* cbs = &buckets_[0][idx];
 211     while (!cbs->empty()) {
 212       auto* cb = &cbs->front();
 213       cbs->pop_front();
 214       count_--;
 215       cb->wheel_ = nullptr;
 216       cb->expiration_ = milliseconds(0);
 217       auto old_ctx =
 218         RequestContext::setContext(cb->context_);
 219       cb->timeoutExpired();
 220       RequestContext::setContext(old_ctx);
 221     }
 222   }
 223   if (count_ > 0) {
 224     this->AsyncTimeout::scheduleTimeout(interval_.count());
 225   }
 226 }
 227
 228 size_t HHWheelTimer::cancelAll() {
 229   size_t count = 0;
 230
 231   if (count_ != 0) {
 232     const size_t numElements = WHEEL_BUCKETS * WHEEL_SIZE;
 233     size_t maxBuckets = std::min(numElements, count_);
 234     auto buckets = folly::make_unique<CallbackList[]>(maxBuckets);
 235     size_t countBuckets = 0;
 236     for (auto& tick : buckets_) {
 237       for (auto& bucket : tick) {
 238         if (bucket.empty()) {
 239           continue;
 240         }
 241         for (auto& cb : bucket) {
 242           count++;
 243         }
 244         std::swap(bucket, buckets[countBuckets++]);
 245         if (count >= count_) {
 246           break;
 247         }
 248       }
 249     }
 250
 251     for (size_t i = 0; i < countBuckets; ++i) {
 252       auto& bucket = buckets[i];
 253       while (!bucket.empty()) {
 254         auto& cb = bucket.front();
 255         cb.cancelTimeout();
 256         cb.callbackCanceled();
 257       }
 258     }
 259   }
 260
 261   return count;
 262 }
 263
 264 } // folly