folly/executors/IOThreadPoolExecutor.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
  17 #include <folly/executors/IOThreadPoolExecutor.h>
  18
  19 #include <glog/logging.h>
  20
  21 #include <folly/detail/MemoryIdler.h>
  22
  23 namespace folly {
  24
  25 using folly::detail::MemoryIdler;
  26
  27 /* Class that will free jemalloc caches and madvise the stack away
  28  * if the event loop is unused for some period of time
  29  */
  30 class MemoryIdlerTimeout : public AsyncTimeout, public EventBase::LoopCallback {
  31  public:
  32   explicit MemoryIdlerTimeout(EventBase* b) : AsyncTimeout(b), base_(b) {}
  33
  34   void timeoutExpired() noexcept override {
  35     idled = true;
  36   }
  37
  38   void runLoopCallback() noexcept override {
  39     if (idled) {
  40       MemoryIdler::flushLocalMallocCaches();
  41       MemoryIdler::unmapUnusedStack(MemoryIdler::kDefaultStackToRetain);
  42
  43       idled = false;
  44     } else {
  45       std::chrono::steady_clock::duration idleTimeout =
  46           MemoryIdler::defaultIdleTimeout.load(std::memory_order_acquire);
  47
  48       idleTimeout = MemoryIdler::getVariationTimeout(idleTimeout);
  49
  50       scheduleTimeout(
  51           std::chrono::duration_cast<std::chrono::milliseconds>(idleTimeout)
  52               .count());
  53     }
  54
  55     // reschedule this callback for the next event loop.
  56     base_->runBeforeLoop(this);
  57   }
  58
  59  private:
  60   EventBase* base_;
  61   bool idled{false};
  62 };
  63
  64 IOThreadPoolExecutor::IOThreadPoolExecutor(
  65     size_t numThreads,
  66     std::shared_ptr<ThreadFactory> threadFactory,
  67     EventBaseManager* ebm,
  68     bool waitForAll)
  69     : ThreadPoolExecutor(numThreads, std::move(threadFactory), waitForAll),
  70       nextThread_(0),
  71       eventBaseManager_(ebm) {
  72   setNumThreads(numThreads);
  73 }
  74
  75 IOThreadPoolExecutor::~IOThreadPoolExecutor() {
  76   stop();
  77 }
  78
  79 void IOThreadPoolExecutor::add(Func func) {
  80   add(std::move(func), std::chrono::milliseconds(0));
  81 }
  82
  83 void IOThreadPoolExecutor::add(
  84     Func func,
  85     std::chrono::milliseconds expiration,
  86     Func expireCallback) {
  87   RWSpinLock::ReadHolder r{&threadListLock_};
  88   if (threadList_.get().empty()) {
  89     throw std::runtime_error("No threads available");
  90   }
  91   auto ioThread = pickThread();
  92
  93   auto task = Task(std::move(func), expiration, std::move(expireCallback));
  94   auto wrappedFunc = [ ioThread, task = std::move(task) ]() mutable {
  95     runTask(ioThread, std::move(task));
  96     ioThread->pendingTasks--;
  97   };
  98
  99   ioThread->pendingTasks++;
 100   if (!ioThread->eventBase->runInEventBaseThread(std::move(wrappedFunc))) {
 101     ioThread->pendingTasks--;
 102     throw std::runtime_error("Unable to run func in event base thread");
 103   }
 104 }
 105
 106 std::shared_ptr<IOThreadPoolExecutor::IOThread>
 107 IOThreadPoolExecutor::pickThread() {
 108   auto& me = *thisThread_;
 109   auto& ths = threadList_.get();
 110   // When new task is added to IOThreadPoolExecutor, a thread is chosen for it
 111   // to be executed on, thisThread_ is by default chosen, however, if the new
 112   // task is added by the clean up operations on thread destruction, thisThread_
 113   // is not an available thread anymore, thus, always check whether or not
 114   // thisThread_ is an available thread before choosing it.
 115   if (me && std::find(ths.cbegin(), ths.cend(), me) != ths.cend()) {
 116     return me;
 117   }
 118   auto n = ths.size();
 119   if (n == 0) {
 120     return me;
 121   }
 122   auto thread = ths[nextThread_++ % n];
 123   return std::static_pointer_cast<IOThread>(thread);
 124 }
 125
 126 EventBase* IOThreadPoolExecutor::getEventBase() {
 127   return pickThread()->eventBase;
 128 }
 129
 130 EventBase* IOThreadPoolExecutor::getEventBase(
 131     ThreadPoolExecutor::ThreadHandle* h) {
 132   auto thread = dynamic_cast<IOThread*>(h);
 133
 134   if (thread) {
 135     return thread->eventBase;
 136   }
 137
 138   return nullptr;
 139 }
 140
 141 EventBaseManager* IOThreadPoolExecutor::getEventBaseManager() {
 142   return eventBaseManager_;
 143 }
 144
 145 std::shared_ptr<ThreadPoolExecutor::Thread> IOThreadPoolExecutor::makeThread() {
 146   return std::make_shared<IOThread>(this);
 147 }
 148
 149 void IOThreadPoolExecutor::threadRun(ThreadPtr thread) {
 150   this->threadPoolHook_.registerThread();
 151
 152   const auto ioThread = std::static_pointer_cast<IOThread>(thread);
 153   ioThread->eventBase = eventBaseManager_->getEventBase();
 154   thisThread_.reset(new std::shared_ptr<IOThread>(ioThread));
 155
 156   auto idler = std::make_unique<MemoryIdlerTimeout>(ioThread->eventBase);
 157   ioThread->eventBase->runBeforeLoop(idler.get());
 158
 159   ioThread->eventBase->runInEventBaseThread(
 160       [thread] { thread->startupBaton.post(); });
 161   while (ioThread->shouldRun) {
 162     ioThread->eventBase->loopForever();
 163   }
 164   if (isJoin_) {
 165     while (ioThread->pendingTasks > 0) {
 166       ioThread->eventBase->loopOnce();
 167     }
 168   }
 169   idler.reset();
 170   if (isWaitForAll_) {
 171     // some tasks, like thrift asynchronous calls, create additional
 172     // event base hookups, let's wait till all of them complete.
 173     ioThread->eventBase->loop();
 174   }
 175
 176   std::lock_guard<std::mutex> guard(ioThread->eventBaseShutdownMutex_);
 177   ioThread->eventBase = nullptr;
 178   eventBaseManager_->clearEventBase();
 179 }
 180
 181 // threadListLock_ is writelocked
 182 void IOThreadPoolExecutor::stopThreads(size_t n) {
 183   std::vector<ThreadPtr> stoppedThreads;
 184   stoppedThreads.reserve(n);
 185   for (size_t i = 0; i < n; i++) {
 186     const auto ioThread =
 187         std::static_pointer_cast<IOThread>(threadList_.get()[i]);
 188     for (auto& o : observers_) {
 189       o->threadStopped(ioThread.get());
 190     }
 191     ioThread->shouldRun = false;
 192     stoppedThreads.push_back(ioThread);
 193     std::lock_guard<std::mutex> guard(ioThread->eventBaseShutdownMutex_);
 194     if (ioThread->eventBase) {
 195       ioThread->eventBase->terminateLoopSoon();
 196     }
 197   }
 198   for (auto thread : stoppedThreads) {
 199     stoppedThreads_.add(thread);
 200     threadList_.remove(thread);
 201   }
 202 }
 203
 204 // threadListLock_ is readlocked
 205 uint64_t IOThreadPoolExecutor::getPendingTaskCountImpl(
 206     const folly::RWSpinLock::ReadHolder&) {
 207   uint64_t count = 0;
 208   for (const auto& thread : threadList_.get()) {
 209     auto ioThread = std::static_pointer_cast<IOThread>(thread);
 210     size_t pendingTasks = ioThread->pendingTasks;
 211     if (pendingTasks > 0 && !ioThread->idle) {
 212       pendingTasks--;
 213     }
 214     count += pendingTasks;
 215   }
 216   return count;
 217 }
 218
 219 } // namespace folly