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   struct PrivateConstructorTag {};
  31
  32  public:
  33   WTCallback(PrivateConstructorTag, EventBase* base) : base_(base) {}
  34
  35   // Only allow creation by this factory, to ensure heap allocation.
  36   static std::shared_ptr<WTCallback> create(EventBase* base) {
  37     // optimization opportunity: memory pool
  38     auto cob = std::make_shared<WTCallback>(PrivateConstructorTag{}, base);
  39     // Capture shared_ptr of cob in lambda so that Core inside Promise will
  40     // hold a ref count to it. The ref count will be released when Core goes
  41     // away which happens when both Promise and Future go away
  42     cob->promise_.setInterruptHandler(
  43         [cob](const folly::exception_wrapper&) { cob->interruptHandler(); });
  44     return cob;
  45   }
  46
  47   Future<Unit> getFuture() {
  48     return promise_.getFuture();
  49   }
  50
  51   void releasePromise() {
  52     // Don't need promise anymore. Break the circular reference as promise_
  53     // is holding a ref count to us via Core. Core won't go away until both
  54     // Promise and Future go away.
  55     promise_ = Promise<Unit>::makeEmpty();
  56   }
  57
  58  protected:
  59   EventBase* base_;
  60   Promise<Unit> promise_;
  61
  62   void timeoutExpired() noexcept override {
  63     promise_.setValue();
  64     // Don't need Promise anymore, break the circular reference
  65     releasePromise();
  66   }
  67
  68   void interruptHandler() {
  69     // Capture shared_ptr of self in lambda, if we don't do this, object
  70     // may go away before the lambda is executed from event base thread.
  71     // This is not racing with timeoutExpired anymore because this is called
  72     // through Future, which means Core is still alive and keeping a ref count
  73     // on us, so what timeouExpired is doing won't make the object go away
  74     base_->runInEventBaseThread([me = shared_from_this()] {
  75       me->cancelTimeout();
  76       // Don't need Promise anymore, break the circular reference
  77       me->releasePromise();
  78     });
  79   }
  80 };
  81
  82 } // namespace
  83
  84 ThreadWheelTimekeeper::ThreadWheelTimekeeper()
  85     : thread_([this] { eventBase_.loopForever(); }),
  86       wheelTimer_(
  87           HHWheelTimer::newTimer(&eventBase_, std::chrono::milliseconds(1))) {
  88   eventBase_.waitUntilRunning();
  89   eventBase_.runInEventBaseThread([this]{
  90     // 15 characters max
  91     eventBase_.setName("FutureTimekeepr");
  92   });
  93 }
  94
  95 ThreadWheelTimekeeper::~ThreadWheelTimekeeper() {
  96   eventBase_.runInEventBaseThreadAndWait([this]{
  97     wheelTimer_->cancelAll();
  98     eventBase_.terminateLoopSoon();
  99   });
 100   thread_.join();
 101 }
 102
 103 Future<Unit> ThreadWheelTimekeeper::after(Duration dur) {
 104   auto cob = WTCallback::create(&eventBase_);
 105   auto f = cob->getFuture();
 106   //
 107   // Even shared_ptr of cob is captured in lambda this is still somewhat *racy*
 108   // because it will be released once timeout is scheduled. So technically there
 109   // is no gurantee that EventBase thread can safely call timeout callback.
 110   // However due to fact that we are having circular reference here:
 111   // WTCallback->Promise->Core->WTCallbak, so three of them won't go away until
 112   // we break the circular reference. The break happens either in
 113   // WTCallback::timeoutExpired or WTCallback::interruptHandler. Former means
 114   // timeout callback is being safely executed. Latter captures shared_ptr of
 115   // WTCallback again in another lambda for canceling timeout. The moment
 116   // canceling timeout is executed in EventBase thread, the actual timeout
 117   // callback has either been executed, or will never be executed. So we are
 118   // fine here.
 119   //
 120   if (!eventBase_.runInEventBaseThread([this, cob, dur]{
 121         wheelTimer_->scheduleTimeout(cob.get(), dur);
 122       })) {
 123     // Release promise to break the circular reference. Because if
 124     // scheduleTimeout fails, there is nothing to *promise*. Internally
 125     // Core would automatically set an exception result when Promise is
 126     // destructed before fulfilling.
 127     // This is either called from EventBase thread, or here.
 128     // They are somewhat racy but given the rare chance this could fail,
 129     // I don't see it is introducing any problem yet.
 130     cob->releasePromise();
 131   }
 132   return f;
 133 }
 134
 135 namespace detail {
 136
 137 std::shared_ptr<Timekeeper> getTimekeeperSingleton() {
 138   return timekeeperSingleton_.try_get();
 139 }
 140
 141 } // namespace detail
 142
 143 } // namespace folly