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/Memory.h>
  25 #include <folly/Optional.h>
  26 #include <folly/ScopeGuard.h>
  27
  28 #include <cassert>
  29
  30 using std::chrono::milliseconds;
  31
  32 namespace folly {
  33
  34 /**
  35  * We want to select the default interval carefully.
  36  * An interval of 10ms will give us 10ms * WHEEL_SIZE^WHEEL_BUCKETS
  37  * for the largest timeout possible, or about 497 days.
  38  *
  39  * For a lower bound, we want a reasonable limit on local IO, 10ms
  40  * seems short enough
  41  *
  42  * A shorter interval also has CPU implications, less than 1ms might
  43  * start showing up in cpu perf.  Also, it might not be possible to set
  44  * tick interval less than 10ms on older kernels.
  45  */
  46 int HHWheelTimer::DEFAULT_TICK_INTERVAL = 10;
  47
  48 HHWheelTimer::Callback::~Callback() {
  49   if (isScheduled()) {
  50     cancelTimeout();
  51   }
  52 }
  53
  54 void HHWheelTimer::Callback::setScheduled(HHWheelTimer* wheel,
  55                                           std::chrono::milliseconds timeout) {
  56   assert(wheel_ == nullptr);
  57   assert(expiration_ == milliseconds(0));
  58
  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   expiration_ = milliseconds(0);
  78 }
  79
  80 HHWheelTimer::HHWheelTimer(
  81     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_(nullptr) {}
  93
  94 HHWheelTimer::~HHWheelTimer() {
  95   // Ensure this gets done, but right before destruction finishes.
  96   auto destructionPublisherGuard = folly::makeGuard([&] {
  97     // Inform the subscriber that this instance is doomed.
  98     if (processingCallbacksGuard_) {
  99       *processingCallbacksGuard_ = true;
 100     }
 101   });
 102   cancelAll();
 103 }
 104
 105 void HHWheelTimer::scheduleTimeoutImpl(Callback* callback,
 106                                        std::chrono::milliseconds timeout) {
 107   int64_t due = timeToWheelTicks(timeout) + nextTick_;
 108   int64_t diff = due - nextTick_;
 109   CallbackList* list;
 110
 111   if (diff < 0) {
 112     list = &buckets_[0][nextTick_ & WHEEL_MASK];
 113   } else if (diff < WHEEL_SIZE) {
 114     list = &buckets_[0][due & WHEEL_MASK];
 115   } else if (diff < 1 << (2 * WHEEL_BITS)) {
 116     list = &buckets_[1][(due >> WHEEL_BITS) & WHEEL_MASK];
 117   } else if (diff < 1 << (3 * WHEEL_BITS)) {
 118     list = &buckets_[2][(due >> 2 * WHEEL_BITS) & WHEEL_MASK];
 119   } else {
 120     /* in largest slot */
 121     if (diff > LARGEST_SLOT) {
 122       diff = LARGEST_SLOT;
 123       due = diff + nextTick_;
 124     }
 125     list = &buckets_[3][(due >> 3 * WHEEL_BITS) & WHEEL_MASK];
 126   }
 127   list->push_back(*callback);
 128 }
 129
 130 void HHWheelTimer::scheduleTimeout(Callback* callback,
 131                                    std::chrono::milliseconds timeout) {
 132   // Cancel the callback if it happens to be scheduled already.
 133   callback->cancelTimeout();
 134
 135   callback->context_ = RequestContext::saveContext();
 136
 137   if (count_ == 0 && !processingCallbacksGuard_) {
 138     this->AsyncTimeout::scheduleTimeout(interval_.count());
 139   }
 140
 141   callback->setScheduled(this, timeout);
 142   scheduleTimeoutImpl(callback, timeout);
 143   count_++;
 144 }
 145
 146 void HHWheelTimer::scheduleTimeout(Callback* callback) {
 147   CHECK(std::chrono::milliseconds(-1) != defaultTimeout_)
 148       << "Default timeout was not initialized";
 149   scheduleTimeout(callback, defaultTimeout_);
 150 }
 151
 152 bool HHWheelTimer::cascadeTimers(int bucket, int tick) {
 153   CallbackList cbs;
 154   cbs.swap(buckets_[bucket][tick]);
 155   while (!cbs.empty()) {
 156     auto* cb = &cbs.front();
 157     cbs.pop_front();
 158     scheduleTimeoutImpl(cb, cb->getTimeRemaining(now_));
 159   }
 160
 161   // If tick is zero, timeoutExpired will cascade the next bucket.
 162   return tick == 0;
 163 }
 164
 165 void HHWheelTimer::timeoutExpired() noexcept {
 166   // If the last smart pointer for "this" is reset inside the callback's
 167   // timeoutExpired(), then the guard will detect that it is time to bail from
 168   // this method.
 169   auto isDestroyed = false;
 170   // If scheduleTimeout is called from a callback in this function, it may
 171   // cause inconsistencies in the state of this object. As such, we need
 172   // to treat these calls slightly differently.
 173   CHECK(!processingCallbacksGuard_);
 174   processingCallbacksGuard_ = &isDestroyed;
 175   auto reEntryGuard = folly::makeGuard([&] {
 176     if (!isDestroyed) {
 177       processingCallbacksGuard_ = nullptr;
 178     }
 179   });
 180
 181   // timeoutExpired() can only be invoked directly from the event base loop.
 182   // It should never be invoked recursively.
 183   //
 184   milliseconds catchup = now_ + interval_;
 185   // If catchup is enabled, we may have missed multiple intervals, use
 186   // currentTime() to check exactly.
 187   if (++expirationsSinceCatchup_ >= catchupEveryN_) {
 188     catchup = std::chrono::duration_cast<milliseconds>(
 189       std::chrono::steady_clock::now().time_since_epoch());
 190     expirationsSinceCatchup_ = 0;
 191   }
 192   while (now_ < catchup) {
 193     now_ += interval_;
 194
 195     int idx = nextTick_ & WHEEL_MASK;
 196     if (0 == idx) {
 197       // Cascade timers
 198       if (cascadeTimers(1, (nextTick_ >> WHEEL_BITS) & WHEEL_MASK) &&
 199           cascadeTimers(2, (nextTick_ >> (2 * WHEEL_BITS)) & WHEEL_MASK)) {
 200         cascadeTimers(3, (nextTick_ >> (3 * WHEEL_BITS)) & WHEEL_MASK);
 201       }
 202     }
 203
 204     nextTick_++;
 205     CallbackList* cbs = &buckets_[0][idx];
 206     while (!cbs->empty()) {
 207       auto* cb = &cbs->front();
 208       cbs->pop_front();
 209       count_--;
 210       cb->wheel_ = nullptr;
 211       cb->expiration_ = milliseconds(0);
 212       RequestContextScopeGuard rctx(cb->context_);
 213       cb->timeoutExpired();
 214       if (isDestroyed) {
 215         // The HHWheelTimer itself has been destroyed. The other callbacks
 216         // will have been cancelled from the destructor. Bail before causing
 217         // damage.
 218         return;
 219       }
 220     }
 221   }
 222   if (count_ > 0) {
 223     this->AsyncTimeout::scheduleTimeout(interval_.count());
 224   }
 225 }
 226
 227 size_t HHWheelTimer::cancelAll() {
 228   size_t count = 0;
 229
 230   if (count_ != 0) {
 231     const uint64_t numElements = WHEEL_BUCKETS * WHEEL_SIZE;
 232     auto maxBuckets = std::min(numElements, count_);
 233     auto buckets = folly::make_unique<CallbackList[]>(maxBuckets);
 234     size_t countBuckets = 0;
 235     for (auto& tick : buckets_) {
 236       for (auto& bucket : tick) {
 237         if (bucket.empty()) {
 238           continue;
 239         }
 240         for (auto& cb : bucket) {
 241           count++;
 242         }
 243         std::swap(bucket, buckets[countBuckets++]);
 244         if (count >= count_) {
 245           break;
 246         }
 247       }
 248     }
 249
 250     for (size_t i = 0; i < countBuckets; ++i) {
 251       auto& bucket = buckets[i];
 252       while (!bucket.empty()) {
 253         auto& cb = bucket.front();
 254         cb.cancelTimeout();
 255         cb.callbackCanceled();
 256       }
 257     }
 258   }
 259
 260   return count;
 261 }
 262
 263 } // folly