folly/futures/ThreadWheelTimekeeper.cpp

   1 /*
   2  * Copyright 2017 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 #include "ThreadWheelTimekeeper.h"
  17
  18 #include <folly/Singleton.h>
  19 #include <folly/futures/Future.h>
  20 #include <future>
  21
  22 namespace folly {
  23
  24 namespace {
  25 Singleton<ThreadWheelTimekeeper> timekeeperSingleton_;
  26
  27 // Our Callback object for HHWheelTimer
  28 struct WTCallback : public std::enable_shared_from_this<WTCallback>,
  29                     public folly::HHWheelTimer::Callback {
  30  public:
  31   // Only allow creation by this factory, to ensure heap allocation.
  32   static std::shared_ptr<WTCallback> create(EventBase* base) {
  33     // optimization opportunity: memory pool
  34     auto cob = std::shared_ptr<WTCallback>(new WTCallback(base));
  35     // Capture shared_ptr of cob in lambda so that Core inside Promise will
  36     // hold a ref count to it. The ref count will be released when Core goes
  37     // away which happens when both Promise and Future go away
  38     cob->promise_->setInterruptHandler([cob](const folly::exception_wrapper&) {
  39         cob->interruptHandler();
  40       });
  41     return cob;
  42   }
  43
  44   Future<Unit> getFuture() {
  45     return promise_->getFuture();
  46   }
  47
  48   void releasePromise() {
  49     // Don't need promise anymore. Break the circular reference as promise_
  50     // is holding a ref count to us via Core. Core won't go away until both
  51     // Promise and Future go away.
  52     promise_.reset();
  53   }
  54
  55  protected:
  56   EventBase* base_;
  57   std::shared_ptr<Promise<Unit>> promise_;
  58
  59   explicit WTCallback(EventBase* base)
  60       : base_(base) {
  61     promise_ = std::make_shared<Promise<Unit>>();
  62   }
  63
  64   void timeoutExpired() noexcept override {
  65     promise_->setValue();
  66     // Don't need Promise anymore, break the circular reference
  67     releasePromise();
  68   }
  69
  70   void interruptHandler() {
  71     // Capture shared_ptr of self in lambda, if we don't do this, object
  72     // may go away before the lambda is executed from event base thread.
  73     // This is not racing with timeoutExpired anymore because this is called
  74     // through Future, which means Core is still alive and keeping a ref count
  75     // on us, so what timeouExpired is doing won't make the object go away
  76     auto me = shared_from_this();
  77     base_->runInEventBaseThread([me] {
  78         me->cancelTimeout();
  79         // Don't need Promise anymore, break the circular reference
  80         me->releasePromise();
  81       });
  82   }
  83 };
  84
  85 } // namespace
  86
  87 ThreadWheelTimekeeper::ThreadWheelTimekeeper()
  88     : thread_([this] { eventBase_.loopForever(); }),
  89       wheelTimer_(
  90           HHWheelTimer::newTimer(&eventBase_, std::chrono::milliseconds(1))) {
  91   eventBase_.waitUntilRunning();
  92   eventBase_.runInEventBaseThread([this]{
  93     // 15 characters max
  94     eventBase_.setName("FutureTimekeepr");
  95   });
  96 }
  97
  98 ThreadWheelTimekeeper::~ThreadWheelTimekeeper() {
  99   eventBase_.runInEventBaseThreadAndWait([this]{
 100     wheelTimer_->cancelAll();
 101     eventBase_.terminateLoopSoon();
 102   });
 103   thread_.join();
 104 }
 105
 106 Future<Unit> ThreadWheelTimekeeper::after(Duration dur) {
 107   auto cob = WTCallback::create(&eventBase_);
 108   auto f = cob->getFuture();
 109   //
 110   // Even shared_ptr of cob is captured in lambda this is still somewhat *racy*
 111   // because it will be released once timeout is scheduled. So technically there
 112   // is no gurantee that EventBase thread can safely call timeout callback.
 113   // However due to fact that we are having circular reference here:
 114   // WTCallback->Promise->Core->WTCallbak, so three of them won't go away until
 115   // we break the circular reference. The break happens either in
 116   // WTCallback::timeoutExpired or WTCallback::interruptHandler. Former means
 117   // timeout callback is being safely executed. Latter captures shared_ptr of
 118   // WTCallback again in another lambda for canceling timeout. The moment
 119   // canceling timeout is executed in EventBase thread, the actual timeout
 120   // callback has either been executed, or will never be executed. So we are
 121   // fine here.
 122   //
 123   if (!eventBase_.runInEventBaseThread([this, cob, dur]{
 124         wheelTimer_->scheduleTimeout(cob.get(), dur);
 125       })) {
 126     // Release promise to break the circular reference. Because if
 127     // scheduleTimeout fails, there is nothing to *promise*. Internally
 128     // Core would automatically set an exception result when Promise is
 129     // destructed before fulfilling.
 130     // This is either called from EventBase thread, or here.
 131     // They are somewhat racy but given the rare chance this could fail,
 132     // I don't see it is introducing any problem yet.
 133     cob->releasePromise();
 134   }
 135   return f;
 136 }
 137
 138 namespace detail {
 139
 140 std::shared_ptr<Timekeeper> getTimekeeperSingleton() {
 141   return timekeeperSingleton_.try_get();
 142 }
 143
 144 } // namespace detail
 145
 146 } // namespace folly