folly/executors/IOThreadPoolExecutor.h

   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 #pragma once
  18
  19 #include <folly/executors/IOExecutor.h>
  20 #include <folly/executors/ThreadPoolExecutor.h>
  21 #include <folly/io/async/EventBaseManager.h>
  22
  23 namespace folly {
  24
  25 /**
  26  * A Thread Pool for IO bound tasks
  27  *
  28  * @note Uses event_fd for notification, and waking an epoll loop.
  29  * There is one queue (NotificationQueue specifically) per thread/epoll.
  30  * If the thread is already running and not waiting on epoll,
  31  * we don't make any additional syscalls to wake up the loop,
  32  * just put the new task in the queue.
  33  * If any thread has been waiting for more than a few seconds,
  34  * its stack is madvised away. Currently however tasks are scheduled round
  35  * robin on the queues, so unless there is no work going on,
  36  * this isn't very effective.
  37  * Since there is one queue per thread, there is hardly any contention
  38  * on the queues - so a simple spinlock around an std::deque is used for
  39  * the tasks. There is no max queue size.
  40  * By default, there is one thread per core - it usually doesn't make sense to
  41  * have more IO threads than this, assuming they don't block.
  42  *
  43  * @note ::getEventBase() will return an EventBase you can schedule IO work on
  44  * directly, chosen round-robin.
  45  *
  46  * @note N.B. For this thread pool, stop() behaves like join() because
  47  * outstanding tasks belong to the event base and will be executed upon its
  48  * destruction.
  49  */
  50 class IOThreadPoolExecutor : public ThreadPoolExecutor, public IOExecutor {
  51  public:
  52   explicit IOThreadPoolExecutor(
  53       size_t numThreads,
  54       std::shared_ptr<ThreadFactory> threadFactory =
  55           std::make_shared<NamedThreadFactory>("IOThreadPool"),
  56       folly::EventBaseManager* ebm = folly::EventBaseManager::get(),
  57       bool waitForAll = false);
  58
  59   ~IOThreadPoolExecutor() override;
  60
  61   void add(Func func) override;
  62   void add(
  63       Func func,
  64       std::chrono::milliseconds expiration,
  65       Func expireCallback = nullptr) override;
  66
  67   folly::EventBase* getEventBase() override;
  68
  69   static folly::EventBase* getEventBase(ThreadPoolExecutor::ThreadHandle*);
  70
  71   folly::EventBaseManager* getEventBaseManager();
  72
  73  private:
  74   struct FOLLY_ALIGN_TO_AVOID_FALSE_SHARING IOThread : public Thread {
  75     IOThread(IOThreadPoolExecutor* pool)
  76         : Thread(pool), shouldRun(true), pendingTasks(0) {}
  77     std::atomic<bool> shouldRun;
  78     std::atomic<size_t> pendingTasks;
  79     folly::EventBase* eventBase;
  80     std::mutex eventBaseShutdownMutex_;
  81   };
  82
  83   ThreadPtr makeThread() override;
  84   std::shared_ptr<IOThread> pickThread();
  85   void threadRun(ThreadPtr thread) override;
  86   void stopThreads(size_t n) override;
  87   uint64_t getPendingTaskCountImpl(const RWSpinLock::ReadHolder&) override;
  88
  89   size_t nextThread_;
  90   folly::ThreadLocal<std::shared_ptr<IOThread>> thisThread_;
  91   folly::EventBaseManager* eventBaseManager_;
  92 };
  93
  94 } // namespace folly